Add support for otfMetadata in Kinesis firehose response metadata for iceberg table routing

Author: Smelton01

events/firehose.go

@@ -25,6 +25,13 @@ const (
 	KinesisFirehoseTransformedStateProcessingFailed = "ProcessingFailed"
 )
 
+// Constants used for otf operation for the record
+const (
+	KinesisFirehoseOTFOperationInsert = "insert"
+	KinesisFirehoseOTFOperationUpdate = "update"
+	KinesisFirehoseOTFOperationDelete = "delete"
+)
+
 type KinesisFirehoseResponse struct {
 	Records []KinesisFirehoseResponseRecord `json:"records"`
 }
@@ -37,7 +44,14 @@ type KinesisFirehoseResponseRecord struct {
 }
 
 type KinesisFirehoseResponseRecordMetadata struct {
-	PartitionKeys map[string]string `json:"partitionKeys"`
+	PartitionKeys map[string]string                        `json:"partitionKeys"`
+	OTFMetadata   KinesisFirehoseResponseRecordOTFMetadata `json:"otfMetadata"`
+}
+
+type KinesisFirehoseResponseRecordOTFMetadata struct {
+	DestinationTableName    string `json:"destinationTableName"`
+	DestinationDatabaseName string `json:"destinationDatabaseName"`
+	Operation               string `json:"operation"` // The Operation field must have one of the following values – insert, update, or delete.
 }
 
 type KinesisFirehoseRecordMetadata struct {

events/testdata/kinesis-firehose-response.json

@@ -1,31 +1,36 @@
 {
-   "records": [
-     {
-       "data": "SGVsbG8gV29ybGQ=",
-       "recordId": "record1",
-       "result": "TRANSFORMED_STATE_OK",
-       "metadata": {
-         "partitionKeys": {}
-       }
-     },
-     {
-       "data": "SGVsbG8gV29ybGQ=",
-       "recordId": "record2",
-       "result": "TRANSFORMED_STATE_DROPPED",
-       "metadata": {
-         "partitionKeys": {}
-       }
-     },
-     {
-       "data": "SGVsbG8gV29ybGQ=",
-       "recordId": "record3",
-       "result": "TransformedStateOk",
-       "metadata": {
-         "partitionKeys": {
-           "iamKey1": "iamValue1",
-           "iamKey2": "iamValue2"
-         }
-       }
-     }
-   ]
- }
+  "records": [
+    {
+      "data": "SGVsbG8gV29ybGQ=",
+      "recordId": "record1",
+      "result": "TRANSFORMED_STATE_OK",
+      "metadata": {
+        "partitionKeys": {},
+        "otfMetadata": {
+          "destinationTableName": "",
+          "destinationDatabaseName": "",
+          "operation": ""
+        }
+      }
+    },
+    {
+      "data": "SGVsbG8gV29ybGQ=",
+      "recordId": "record2",
+      "result": "TRANSFORMED_STATE_DROPPED",
+      "metadata": {
+        "partitionKeys": {}
+      }
+    },
+    {
+      "data": "SGVsbG8gV29ybGQ=",
+      "recordId": "record3",
+      "result": "TransformedStateOk",
+      "metadata": {
+        "partitionKeys": {
+          "iamKey1": "iamValue1",
+          "iamKey2": "iamValue2"
+        }
+      }
+    }
+  ]
+}