feat: read Kafka header and propagate in the payload (#1565)

Signed-off-by: Vigith Maurice <[email protected]>
numaproj · Mar 15, 2024 · 4457ac8 · 4457ac8
1 parent ae89d93
commit 4457ac8
Show file tree

Hide file tree

Showing 13 changed files with 641 additions and 514 deletions.
diff --git a/api/json-schema/schema.json b/api/json-schema/schema.json
@@ -18849,6 +18849,10 @@
         "consumerGroup": {
           "type": "string"
         },
+        "includeHeaders": {
+          "description": "include Kafka headers (this will change the payload struct into a JSON format with _body and _headers)",
+          "type": "boolean"
+        },
         "sasl": {
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.SASL",
           "description": "SASL user to configure SASL connection for kafka broker SASL.enable=true default for SASL."

diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
@@ -18848,6 +18848,10 @@
         "consumerGroup": {
           "type": "string"
         },
+        "includeHeaders": {
+          "description": "include Kafka headers (this will change the payload struct into a JSON format with _body and _headers)",
+          "type": "boolean"
+        },
         "sasl": {
           "description": "SASL user to configure SASL connection for kafka broker SASL.enable=true default for SASL.",
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.SASL"

diff --git a/config/base/crds/full/numaflow.numaproj.io_pipelines.yaml b/config/base/crds/full/numaflow.numaproj.io_pipelines.yaml
@@ -7323,6 +7323,8 @@ spec:
                               type: string
                             consumerGroup:
                               type: string
+                            includeHeaders:
+                              type: boolean
                             sasl:
                               properties:
                                 gssapi:

diff --git a/config/base/crds/full/numaflow.numaproj.io_vertices.yaml b/config/base/crds/full/numaflow.numaproj.io_vertices.yaml
@@ -2975,6 +2975,8 @@ spec:
                         type: string
                       consumerGroup:
                         type: string
+                      includeHeaders:
+                        type: boolean
                       sasl:
                         properties:
                           gssapi:

diff --git a/config/install.yaml b/config/install.yaml
@@ -9932,6 +9932,8 @@ spec:
                               type: string
                             consumerGroup:
                               type: string
+                            includeHeaders:
+                              type: boolean
                             sasl:
                               properties:
                                 gssapi:
@@ -14784,6 +14786,8 @@ spec:
                         type: string
                       consumerGroup:
                         type: string
+                      includeHeaders:
+                        type: boolean
                       sasl:
                         properties:
                           gssapi:

diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
@@ -9932,6 +9932,8 @@ spec:
                               type: string
                             consumerGroup:
                               type: string
+                            includeHeaders:
+                              type: boolean
                             sasl:
                               properties:
                                 gssapi:
@@ -14784,6 +14786,8 @@ spec:
                         type: string
                       consumerGroup:
                         type: string
+                      includeHeaders:
+                        type: boolean
                       sasl:
                         properties:
                           gssapi:

diff --git a/docs/APIs.md b/docs/APIs.md
@@ -2784,6 +2784,18 @@ default for SASL.
 </p>
 </td>
 </tr>
+<tr>
+<td>
+<code>includeHeaders</code></br> <em> bool </em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+include Kafka headers (this will change the payload struct into a JSON
+format with \_body and \_headers)
+</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="numaflow.numaproj.io/v1alpha1.Lifecycle">

diff --git a/pkg/apis/numaflow/v1alpha1/generated.pb.go b/pkg/apis/numaflow/v1alpha1/generated.pb.go
diff --git a/pkg/apis/numaflow/v1alpha1/generated.proto b/pkg/apis/numaflow/v1alpha1/generated.proto
diff --git a/pkg/apis/numaflow/v1alpha1/kafka_source.go b/pkg/apis/numaflow/v1alpha1/kafka_source.go
@@ -30,4 +30,7 @@ type KafkaSource struct {
 	// SASL.enable=true default for SASL.
 	// +optional
 	SASL *SASL `json:"sasl" protobuf:"bytes,6,opt,name=sasl"`
+	// include Kafka headers (this will change the payload struct into a JSON format with _body and _headers)
+	// +optional
+	IncludeHeaders bool `json:"includeHeaders" protobuf:"bytes,7,opt,name=includeHeaders"`
 }
diff --git a/pkg/apis/numaflow/v1alpha1/openapi_generated.go b/pkg/apis/numaflow/v1alpha1/openapi_generated.go