Added event class MskFirehoseEvent.java for Firehose Lambda transformation when MSK is the source

aws-lambda-java-events/README.md

@@ -44,6 +44,8 @@
 * `KinesisFirehoseEvent`
 * `LambdaDestinationEvent`
 * `LexEvent`
+* `MSKFirehoseEvent`
+* `MSKFirehoseResponse`
 * `RabbitMQEvent`
 * `S3BatchEvent`
 * `S3BatchResponse`

aws-lambda-java-events/src/main/java/com/amazonaws/services/lambda/runtime/events/MSKFirehoseEvent.java

@@ -0,0 +1,68 @@
+package com.amazonaws.services.lambda.runtime.events;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Created by vermshas on 6/28/24.
+ * {
+ *     "invocationId": "",
+ *     "sourceMSKArn": "",
+ *     "deliveryStreamArn": "",
+ *     "region": "us-east-1",
+ *     "records": [
+ *         {
+ *             "recordId": "00000000000000000000000000000000000000000000000000000000000000",
+ *             "approximateArrivalTimestamp": 1716369573887,
+ *             "mskRecordMetadata": {
+ *                 "offset": "0",
+ *                 "partitionId": "1",
+ *                 "approximateArrivalTimestamp": 1716369573887
+ *             },
+ *             "kafkaRecordValue": ""
+ *         }
+ *     ]
+ * }
+ */
+
+@Data
+@Builder(setterPrefix = "with")
+@NoArgsConstructor
+@AllArgsConstructor
+
+public class MSKFirehoseEvent {
+
+    private String invocationId;
+
+    private String deliveryStreamArn;
+
+    private String sourceMSKArn;
+
+    private String region;
+
+    private List<Record> records;
+
+    @Data
+    @Builder(setterPrefix = "with")
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class Record {
+
+        private ByteBuffer kafkaRecordValue;
+
+        private String recordId;
+
+        private Long approximateArrivalEpoch;
+
+        private Long approximateArrivalTimestamp;
+
+        private Map<String, String> mskRecordMetadata;
+
+    }
+}

aws-lambda-java-events/src/main/java/com/amazonaws/services/lambda/runtime/events/MSKFirehoseResponse.java

@@ -0,0 +1,54 @@
+package com.amazonaws.services.lambda.runtime.events;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import lombok.*;
+
+/**
+ * Response model for Amazon Data Firehose Lambda transformation with MSK as a source.
+ */
+
+@Data
+@Builder(setterPrefix = "with")
+@NoArgsConstructor
+@AllArgsConstructor
+
+public class MSKFirehoseResponse {
+
+    public enum Result {
+
+        /**
+         * Indicates that processing of this item succeeded.
+         */
+        Ok,
+
+        /**
+         * Indicate that the processing of this item failed
+         */
+        ProcessingFailed,
+
+        /**
+         * Indicates that this item should be silently dropped
+         */
+        Dropped
+    }
+    public List<Record> records;
+
+    @Data
+    @Builder(setterPrefix = "with")
+    @NoArgsConstructor
+
+    public static class Record {
+        public String recordId;
+        public Result result;
+        public ByteBuffer kafkaRecordValue;
+
+        public Record(String recordId, Result result, ByteBuffer kafkaRecordValue) {
+            super();
+            this.recordId = recordId;
+            this.result = result;
+            this.kafkaRecordValue = kafkaRecordValue;
+        }
+    }
+}

aws-lambda-java-tests/src/test/resources/msk_firehose_event.json

@@ -0,0 +1,18 @@
+{
+  "invocationId": "12345621-4787-0000-a418-36e56Example",
+  "sourceMSKArn": "",
+  "deliveryStreamArn": "",
+  "region": "us-east-1",
+  "records": [
+    {
+      "recordId": "00000000000000000000000000000000000000000000000000000000000000",
+      "approximateArrivalTimestamp": 1716369573887,
+      "mskRecordMetadata": {
+        "offset": "0",
+        "partitionId": "1",
+        "approximateArrivalTimestamp": 1716369573887
+      },
+      "kafkaRecordValue": "eyJOYW1lIjoiSGVsbG8gV29ybGQifQ=="
+    }
+  ]
+}

samples/msk-firehose-event-handler/src/main/java/example/MSKFirehoseEventHandler.java

@@ -0,0 +1,34 @@
+package example;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import model.MSKFirehoseResponse;
+import events.MSKFirehoseEvent;
+import org.json.JSONObject;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A sample MSKFirehoseEvent handler
+ * For more information see the developer guide - <a href="https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html">...</a>
+ */
+public class Sample implements RequestHandler<MSKFirehoseEvent, MSKFirehoseResponse> {
+
+    @Override
+    public MSKFirehoseResponse handleRequest(MSKFirehoseEvent MSKFirehoseEvent, Context context) {
+        List<MSKFirehoseResponse.Record> records = new ArrayList<>();
+
+        for (MSKFirehoseEvent.Record record : MSKFirehoseEvent.getRecords()) {
+            String recordData = new String(record.getKafkaRecordValue().array());
+            // Your business logic
+            JSONObject jsonObject = new JSONObject(recordData);
+            records.add(new MSKFirehoseResponse.Record(record.getRecordId(), MSKFirehoseResponse.Result.Ok, encode(jsonObject.toString())));
+        }
+        return new MSKFirehoseResponse(records);
+    }
+    private ByteBuffer encode(String content) {
+        return ByteBuffer.wrap(content.getBytes());
+    }
+}

samples/msk-firehose-event-handler/src/test/java/example/MSKFirehoseEventHandlerTest.java

@@ -0,0 +1,27 @@
+package example;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.tests.annotations.Event;
+import com.amazonaws.services.lambda.runtime.events.MSKFirehoseEvent;
+import com.amazonaws.services.lambda.runtime.events.MSKFirehoseResponse;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class MSKFirehoseEventHandlerTest {
+
+    private Context context; // intentionally null as it's not used in the test
+
+    @ParameterizedTest
+    @Event(value = "event.json", type = MSKFirehoseEvent.class)
+    public void testEventHandler(MSKFirehoseEvent event) {
+        Sample Sample = new Sample();
+        MSKFirehoseResponse response = Sample.handleRequest(event, context);
+
+        String expectedString = "{\"Name\":\"Hello World\"}";
+        MSKFirehoseResponse.Record firstRecord = response.getRecords().get(0);
+        Assertions.assertEquals(expectedString, UTF_8.decode(firstRecord.getKafkaRecordValue()).toString());
+        Assertions.assertEquals(MSKFirehoseResponse.Result.Ok, firstRecord.getResult());
+    }
+}

samples/msk-firehose-event-handler/src/test/resources/event.json

@@ -0,0 +1,18 @@
+{
+  "invocationId": "12345621-4787-0000-a418-36e56Example",
+  "sourceMSKArn": "",
+  "deliveryStreamArn": "",
+  "region": "us-east-1",
+  "records": [
+    {
+      "recordId": "00000000000000000000000000000000000000000000000000000000000000",
+      "approximateArrivalTimestamp": 1716369573887,
+      "mskRecordMetadata": {
+        "offset": "0",
+        "partitionId": "1",
+        "approximateArrivalTimestamp": 1716369573887
+      },
+      "kafkaRecordValue": "eyJOYW1lIjoiSGVsbG8gV29ybGQifQ=="
+    }
+  ]
+}