Feature request: Model keySchemaMetadata / valueSchemaMetadata in the Parser Kafka schemas #5868

Activity

1. powertools-for-aws-oss-automation commented on Oct 30, 2026

   

   powertools-for-aws-oss-automationBot

   on Oct 30, 2026 – with Powertools for AWS - OSS Automation

   More actions

   Thanks for opening your first issue here! We'll come back to you as soon as we can.
   In the meantime, check out our GitHub Discussions: Join the conversation
2. dreamorosi

   added

   parserThis item relates to the Parser UtilityThis item relates to the Parser Utility

   on Oct 30, 2026
3. dreamorosi

   added the

   Feature

   issue type on Oct 30, 2026
4. dreamorosi

   assigned

   hidingmyname

   on Oct 30, 2026
5. dreamorosi commented on Oct 30, 2026

   

   dreamorosi

   on Oct 30, 2026

   Contributor

   More actions

   Hi @hidingmyname, thanks for this one too. Confirmed: keySchemaMetadata and valueSchemaMetadata are documented in the Lambda Kafka event payload when schema validation is configured on the event source mapping, and KafkaRecordSchema currently strips them. Powertools for AWS Lambda (Python) models both fields in its Parser, as you pointed out.

   Your proposed fix looks good to me, including keeping schemaId optional. Python requires it, but our own @aws-lambda-powertools/kafka treats it as optional (see SchemaMetadata in packages/kafka/src/types/types.ts and the schemaId === undefined branch in the Protobuf deserializer), and the Parser should not reject events that the Kafka utility accepts. Please mention this difference from Python in the PR description.

   I've assigned the issue to you. Please add these tests to the existing describe block in packages/parser/tests/unit/schema/kafka.test.ts, using its existing rawEvent:

   it.each([
     { eventSource: 'aws:kafka', schema: KafkaMskEventSchema },
     { eventSource: 'SelfManagedKafka', schema: KafkaSelfManagedEventSchema },
   ])(
     'preserves the key and value schema metadata ($eventSource)',
     ({ eventSource, schema }) => {
       // Prepare
       const event = { ...structuredClone(rawEvent), eventSource };
       event.records['mytopic-0'][0].keySchemaMetadata = {
         dataFormat: 'AVRO',
         schemaId: '5678',
       };
       event.records['mytopic-0'][0].valueSchemaMetadata = {
         dataFormat: 'AVRO',
         schemaId: 'a1b2c3d4-5678-90ab-cdef-EXAMPLE11111',
       };
   
       // Act
       const result = schema.parse(event);
   
       // Assess
       expect(result.records['mytopic-0'][0]).toEqual(
         expect.objectContaining({
           keySchemaMetadata: { dataFormat: 'AVRO', schemaId: '5678' },
           valueSchemaMetadata: {
             dataFormat: 'AVRO',
             schemaId: 'a1b2c3d4-5678-90ab-cdef-EXAMPLE11111',
           },
         })
       );
     }
   );
   
   it('parses schema metadata without a schemaId', () => {
     // Prepare
     const event = structuredClone(rawEvent);
     event.records['mytopic-0'][0].valueSchemaMetadata = {
       dataFormat: 'PROTOBUF',
     };
   
     // Act
     const result = KafkaMskEventSchema.parse(event);
   
     // Assess
     expect(result.records['mytopic-0'][0].valueSchemaMetadata).toStrictEqual({
       dataFormat: 'PROTOBUF',
     });
   });
   
   it('returns an issue when the schema metadata has no dataFormat', () => {
     // Prepare
     const event = structuredClone(rawEvent);
     (
       event.records['mytopic-0'][0] as Record<string, unknown>
     ).valueSchemaMetadata = { schemaId: '1234' };
   
     // Act
     const result = KafkaMskEventSchema.safeParse(event);
   
     // Assess
     expect(result.success).toBe(false);
     expect(result.error?.issues).toEqual([
       expect.objectContaining({
         path: ['records', 'mytopic-0', 0, 'valueSchemaMetadata', 'dataFormat'],
       }),
     ]);
   });

   All four currently fail on main, and the schema change from your proposal should make them pass. KafkaMskEvent and KafkaSelfManagedEvent should stay inferred from the schema, so no separate type change is needed, and please don't add or modify other tests unless coverage drops below 100%.