fix(parser): decode signed Kafka header bytes as UTF-8

packages/parser/src/schemas/kafka.ts

@@ -22,8 +22,10 @@ const KafkaRecordSchema = z.object({
   headers: z.array(
     z.record(
       z.string(),
-      z.array(z.number()).transform((value) => {
-        return String.fromCodePoint(...value);
+      // Lambda sends header bytes as signed Java bytes, so bytes >= 128 arrive negative.
+      // Buffer.from masks each number to a byte, and decoding matches @aws-lambda-powertools/kafka.
+      z.array(z.number().int().min(-128).max(255)).transform((value) => {
+        return Buffer.from(value).toString('utf8');
       })
     )
   ),

packages/parser/tests/events/kafka/base.json

@@ -17,6 +17,21 @@
             "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
           }
         ]
+      },
+      {
+        "topic": "mymessage-with-unsigned",
+        "partition": 0,
+        "offset": 15,
+        "timestamp": 1545084650987,
+        "timestampType": "CREATE_TIME",
+        "value": "eyJrZXkiOiJ2YWx1ZSJ9",
+        "headers": [
+          {
+            "headerKey": [
+              104, 101, 108, 108, 111, 45, 119, 111, 114, 108, 100, 45, -61, -85
+            ]
+          }
+        ]
       }
     ]
   }

packages/parser/tests/unit/envelopes/kafka.test.ts

@@ -96,7 +96,7 @@ describe('Envelope: Kafka', () => {
       const result = KafkaEnvelope.parse(event, z.string());
 
       // Assess
-      expect(result).toEqual(['{"key":"value"}']);
+      expect(result).toEqual(['{"key":"value"}', '{"key":"value"}']);
     });
 
     it('parses a Kafka event and applies the schema transformation', () => {
@@ -110,7 +110,7 @@ describe('Envelope: Kafka', () => {
       );
 
       // Assess
-      expect(result).toEqual([{ key: 'value' }]);
+      expect(result).toEqual([{ key: 'value' }, { key: 'value' }]);
     });
 
     it('parses a self managed Kafka event', () => {
@@ -122,7 +122,7 @@ describe('Envelope: Kafka', () => {
       const result = KafkaEnvelope.parse(event, z.string());
 
       // Assess
-      expect(result).toEqual(['{"key":"value"}']);
+      expect(result).toEqual(['{"key":"value"}', '{"key":"value"}']);
     });
   });
 
@@ -137,7 +137,7 @@ describe('Envelope: Kafka', () => {
       // Assess
       expect(result).toEqual({
         success: true,
-        data: ['{"key":"value"}'],
+        data: ['{"key":"value"}', '{"key":"value"}'],
       });
     });
 

packages/parser/tests/unit/schema/kafka.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from 'vitest';
+import type { z } from 'zod';
 import {
   KafkaMskEventSchema,
   KafkaSelfManagedEventSchema,
@@ -14,6 +15,11 @@ describe('Schema: Kafka', () => {
     eventsPath: 'kafka',
     filename: 'base',
   });
+  // Typed as the schema input, so header values are the raw byte arrays
+  const rawEvent = getTestEvent<z.input<typeof KafkaMskEventSchema>>({
+    eventsPath: 'kafka',
+    filename: 'base',
+  });
 
   it('parses a Kafka MSK event', () => {
     // Prepare
@@ -47,6 +53,19 @@ describe('Schema: Kafka', () => {
               },
             ],
           },
+          {
+            topic: 'mymessage-with-unsigned',
+            partition: 0,
+            offset: 15,
+            timestamp: 1545084650987,
+            timestampType: 'CREATE_TIME',
+            value: '{"key":"value"}',
+            headers: [
+              {
+                headerKey: 'hello-world-ë',
+              },
+            ],
+          },
         ],
       },
     });
@@ -61,6 +80,42 @@ describe('Schema: Kafka', () => {
     expect(() => KafkaMskEventSchema.parse(event)).toThrow();
   });
 
+  it.each([-129, 256, 1.5])(
+    'returns an issue for a header byte of %s instead of throwing',
+    (byte) => {
+      // Prepare
+      const event = structuredClone(rawEvent);
+      event.records['mytopic-0'][0].headers = [{ headerKey: [byte] }];
+
+      // Act
+      const result = KafkaMskEventSchema.safeParse(event);
+
+      // Assess
+      expect(result.success).toBe(false);
+      expect(result.error?.issues).toEqual([
+        expect.objectContaining({
+          path: ['records', 'mytopic-0', 0, 'headers', 0, 'headerKey', 0],
+        }),
+      ]);
+    }
+  );
+
+  it('decodes a large header without exceeding the call stack', () => {
+    // Prepare
+    const event = structuredClone(rawEvent);
+    event.records['mytopic-0'][0].headers = [
+      { headerKey: new Array(200_000).fill(65) },
+    ];
+
+    // Act
+    const result = KafkaMskEventSchema.parse(event);
+
+    // Assess
+    expect(result.records['mytopic-0'][0].headers).toEqual([
+      { headerKey: 'A'.repeat(200_000) },
+    ]);
+  });
+
   it('parses a Kafka self-managed event', () => {
     // Prepare
     const event = omit(
@@ -92,6 +147,19 @@ describe('Schema: Kafka', () => {
               },
             ],
           },
+          {
+            topic: 'mymessage-with-unsigned',
+            partition: 0,
+            offset: 15,
+            timestamp: 1545084650987,
+            timestampType: 'CREATE_TIME',
+            value: '{"key":"value"}',
+            headers: [
+              {
+                headerKey: 'hello-world-ë',
+              },
+            ],
+          },
         ],
       },
     });