Feature request: [Kafka] Skip N leading bytes of a Kafka Avro payload before deserialization

[amin-farjadi]

Use case

Current Sate

When Kafka messages are produced with a schema-registry-aware serializer (for example
Confluent's KafkaAvroSerializer or AWS Glue's Avro serializer), the producer prepends a
short wire-format wrapper to every payload — typically 5 bytes for Confluent
(1-byte magic byte 0x00 + 4-byte big-endian schema ID) or 18 bytes for Glue
(1-byte header version + 1-byte compression + 16-byte UUID schema-version ID).

When my consumer uses the ESM Schema Registry integration, ESM strips those bytes upstream
and populates valueSchemaMetadata.schemaId — the existing AvroDeserializer behaves
correctly.

Problem

I have my Avro schema offline (checked into my
Lambda), and I do not use the ESM Schema Registry integration. As a result:

• The raw payload — including the 5-byte wire-format prefix — reaches my Lambda function.
• valueSchemaMetadata.schemaId is not populated on the event, so the protobuf-style
  auto-detection in deserializer/protobuf.py does not translate.
• AvroDeserializer.deserialize hands the prefixed bytes straight to
  avro.io.BinaryDecoder. Because the prefix is legal Avro varint noise, the decoder
  either silently returns corrupted values (empty strings, zero ints) or raises
  KafkaConsumerDeserializationError, depending on the specific prefix bytes.

I want to keep using Powertools' kafka_consumer decorator with my offline Avro schema
without having to hand-roll a base64-decode + slice + re-encode dance in every handler.

Solution/User Experience

Add two new optional parameters on SchemaConfig:

• value_schema_id_prefix_length: int = 0
• key_schema_id_prefix_length: int = 0

They tell the Avro deserializer how many leading bytes to skip after base64-decoding the
payload and before handing it to the Avro BinaryDecoder. Default is 0, preserving
today's behaviour for anyone not using a schema-registry wire format.

from aws_lambda_powertools.utilities.kafka import SchemaConfig, kafka_consumer

schema_config = SchemaConfig(
    value_schema_type="AVRO",
    value_schema=AVRO_SCHEMA,           # offline, checked into the Lambda
    value_schema_id_prefix_length=5,    # skip Confluent's magic byte + schema ID
)

@kafka_consumer(schema_config=schema_config)
def handler(event, context):
    for record in event.records:
        # record.value is the fully-deserialized Avro payload,
        # with the 5-byte prefix stripped transparently.
        ...

Scope of the change:

• Applies to both value and key sides, symmetrical with every other SchemaConfig
  option.
• Applies to Avro only for now. Protobuf already has schema-registry framing logic
  keyed on schemaId length; JSON-with-registry is uncommon and can follow later if there
  is demand.
• Non-negative-integer validation on the parameters at SchemaConfig construction time.
• The offset is included in the deserializer cache key so that concurrent handlers using
  the same Avro schema with different prefix lengths do not collide.

Alternative solutions

• Require users to pre-process the payload themselves (base64-decode, slice, re-encode before the decorator runs). Works today but is boilerplate every user of an offline schema has to repeat, and defeats the ergonomic value of kafka_consumer.
• Add ESM Schema Registery integration. For a lamdba that's integrating with 1 Kafka stream, it is not worth adding more infrastructure.

Acknowledgment

• This feature request meets Powertools for AWS Lambda (Python) Tenets
• Should this be considered in other Powertools for AWS Lambda languages? i.e. Java, TypeScript, and .NET

[amin-farjadi]

@leandrodamascena I have a solution for this ready if the request is approved. It is not a big change so, I'm confident it can get merged quickly.