diff --git a/packages/parser/src/schemas/kafka.ts b/packages/parser/src/schemas/kafka.ts index d69f37e718..5b922d7791 100644 --- a/packages/parser/src/schemas/kafka.ts +++ b/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'); }) ) ), diff --git a/packages/parser/tests/events/kafka/base.json b/packages/parser/tests/events/kafka/base.json index 4c17ea2490..51b03ac832 100644 --- a/packages/parser/tests/events/kafka/base.json +++ b/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 + ] + } + ] } ] } diff --git a/packages/parser/tests/unit/envelopes/kafka.test.ts b/packages/parser/tests/unit/envelopes/kafka.test.ts index 6eb741f767..c621ffff21 100644 --- a/packages/parser/tests/unit/envelopes/kafka.test.ts +++ b/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"}'], }); }); diff --git a/packages/parser/tests/unit/schema/kafka.test.ts b/packages/parser/tests/unit/schema/kafka.test.ts index 845d20fcdf..464cf29e09 100644 --- a/packages/parser/tests/unit/schema/kafka.test.ts +++ b/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>({ + 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-ë', + }, + ], + }, ], }, });