diff --git a/apis/fluentbit/v1alpha2/plugins/output/kafka_types.go b/apis/fluentbit/v1alpha2/plugins/output/kafka_types.go
index 61e0bd2bc..beb0d2089 100644
--- a/apis/fluentbit/v1alpha2/plugins/output/kafka_types.go
+++ b/apis/fluentbit/v1alpha2/plugins/output/kafka_types.go
@@ -12,8 +12,10 @@ import (
 // Kafka output plugin allows to ingest your records into an Apache Kafka service. <br />
 // **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/outputs/kafka**
 type Kafka struct {
-	// Specify data format, options available: json, msgpack.
+	// Specify data format, options available: json, msgpack, raw.
 	Format string `json:"format,omitempty"`
+	// When using the raw format, the value of the record field specified by RawLogKey is sent to Kafka as the payload.
+	RawLogKey string `json:"rawLogKey,omitempty"`
 	// Optional key to store the message
 	MessageKey string `json:"messageKey,omitempty"`
 	// If set, the value of Message_Key_Field in the record will indicate the message key.
@@ -60,6 +62,9 @@ func (k *Kafka) Params(_ plugins.SecretLoader) (*params.KVs, error) {
 	if k.Format != "" {
 		kvs.Insert("Format", k.Format)
 	}
+	if k.RawLogKey != "" {
+		kvs.Insert("Raw_Log_Key", k.RawLogKey)
+	}
 	if k.MessageKey != "" {
 		kvs.Insert("Message_Key", k.MessageKey)
 	}
diff --git a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml
index 214a9c96b..715268190 100644
--- a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml
+++ b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml
@@ -2258,7 +2258,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2277,6 +2278,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml
index 76088d713..f80db4271 100644
--- a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml
+++ b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml
@@ -2258,7 +2258,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2277,6 +2278,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
index 2abf993df..5f69755fb 100644
--- a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
+++ b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
@@ -2256,7 +2256,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2275,6 +2276,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml
index dcee8f101..154ef72ee 100644
--- a/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml
+++ b/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml
@@ -2256,7 +2256,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2275,6 +2276,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
index 8841ee8f2..027b3ac05 100644
--- a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
@@ -2257,7 +2257,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2276,6 +2277,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/config/crd/bases/fluentbit.fluent.io_outputs.yaml b/config/crd/bases/fluentbit.fluent.io_outputs.yaml
index 34c4f25fa..c2fe12bc4 100644
--- a/config/crd/bases/fluentbit.fluent.io_outputs.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_outputs.yaml
@@ -2257,7 +2257,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -2276,6 +2277,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
diff --git a/docs/plugins/fluentbit/output/kafka.md b/docs/plugins/fluentbit/output/kafka.md
index 65ac6917a..c356ccaff 100644
--- a/docs/plugins/fluentbit/output/kafka.md
+++ b/docs/plugins/fluentbit/output/kafka.md
@@ -5,7 +5,8 @@ Kafka output plugin allows to ingest your records into an Apache Kafka service.
 
 | Field | Description | Scheme |
 | ----- | ----------- | ------ |
-| format | Specify data format, options available: json, msgpack. | string |
+| format | Specify data format, options available: json, msgpack, raw. | string |
+| rawLogKey | When using the raw format, the value of the record field specified by RawLogKey is sent to Kafka as the payload. | string |
 | messageKey | Optional key to store the message | string |
 | messageKeyField | If set, the value of Message_Key_Field in the record will indicate the message key. If not set nor found in the record, Message_Key will be used (if set). | string |
 | timestampKey | Set the key to store the record timestamp | string |
diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml
index fa044f4a2..43bf9645d 100644
--- a/manifests/setup/setup.yaml
+++ b/manifests/setup/setup.yaml
@@ -6457,7 +6457,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -6476,6 +6477,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string
@@ -37123,7 +37128,8 @@ spec:
                       So in Topics only a default topic needs to be configured
                     type: boolean
                   format:
-                    description: 'Specify data format, options available: json, msgpack.'
+                    description: 'Specify data format, options available: json, msgpack,
+                      raw.'
                     type: string
                   messageKey:
                     description: Optional key to store the message
@@ -37142,6 +37148,10 @@ spec:
                       Setting the queue_full_retries value to 0 set's an unlimited number of retries.
                     format: int64
                     type: integer
+                  rawLogKey:
+                    description: When using the raw format, the value of the record
+                      field specified by RawLogKey is sent to Kafka as the payload.
+                    type: string
                   rdkafka:
                     additionalProperties:
                       type: string