fix(kafka): handle tombstone events without value (#5017)

Co-authored-by: Bertram Vogel <[email protected]>
Co-authored-by: Andrea Amorosi <[email protected]>

Author: 3 people

packages/kafka/src/consumer.ts

@@ -196,7 +196,11 @@ const deserializeRecord = async (
     },
     originalKey: key,
     get value() {
+      if (value === undefined) {
+        return undefined;
+      }
       if (isNull(value)) return null;
+
       const deserializedValue = deserialize({
         value: value,
         deserializer: deserializerValue,

packages/kafka/src/types/types.ts

@@ -201,7 +201,7 @@ type Record = {
   /**
    * Base64-encoded value of the record
    */
-  value: string | null;
+  value?: string | null;
   /**
    * Array of record headers
    */

packages/kafka/tests/unit/consumer.test.ts

@@ -609,4 +609,32 @@ describe('Kafka consumer', () => {
     // Assess
     expect(result).toBeNull();
   });
+
+  it('handles tombstone events without a message value', async () => {
+    // Prepare
+    const event = structuredClone(jsonTestEvent);
+    delete event.records['mytopic-0'][0].value;
+
+    const handler = kafkaConsumer<string, unknown>(
+      async (event) => {
+        await setTimeout(1); // simulate some processing time
+        const firstRecord = event.records[0];
+        if (firstRecord) {
+          return firstRecord.value;
+        }
+        return undefined;
+      },
+      {
+        value: {
+          type: SchemaType.JSON,
+        },
+      }
+    );
+
+    // Act
+    const result = await handler(event, context);
+
+    // Assess
+    expect(result).toBeUndefined();
+  });
 });