Bug: Kafka ProtobufDeserializer reads Confluent message indexes as plain varints, dropping fields or failing for non-first message types

Aperta
  • Qualcosa non va?
#8,665 4 commenti 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@alphasafal ci sta già lavorando.

Dal 25/10/2026.

  • #8670 di @alphasafal — aperta

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

bug event_sources
Expected Behaviour

A Protobuf value produced with Confluent Schema Registry decodes into every field, whatever message index the serializer wrote. Confluent's serializers write the message index array as zigzag encoded varints (array length, then each index), with [0] as a single 0x00 byte; see _encode_varints / _read_index_array in confluent_kafka.schema_registry.protobuf.

Current Behaviour

ProtobufDeserializer._remove_message_index (kafka/deserializer/protobuf.py) reads the array with _DecodeSignedVarint, so it reads twice the real length. Only index [0] works. For any other index:

  • it skips payload bytes as if they were indexes, and the record either fails with DecodeError: Wire format was corrupt, or
  • it decodes without error but with fields silently missing.

The second case is already in the repo's fixture: record offset 17 in tests/functional/kafka_consumer/_protobuf/schemas/complex_schema_with_confuent.py has the prefix 04 02 00 (zigzag for [1, 0]) and decodes without userId and name. test_confluent_schema_registry_complex_schema only asserts email and age, so it passes.

Code snippet
import base64

from google.protobuf.descriptor_pb2 import FieldDescriptorProto

from aws_lambda_powertools.utilities.kafka import ConsumerRecords, SchemaConfig, kafka_consumer


def confluent_framed(message, indexes):
    # Message index array as confluent_kafka's ProtobufSerializer writes it: zigzag varints, [0] as a single 0x00
    def zigzag(v):
        v, out = (v << 1) ^ (v >> 63), bytearray()
        while v & ~0x7F:
            out.append((v & 0x7F) | 0x80)
            v >>= 7
        return bytes(out + bytes([v]))

    prefix = b"\x00" if indexes == [0] else zigzag(len(indexes)) + b"".join(zigzag(i) for i in indexes)
    return base64.b64encode(prefix + message.SerializeToString()).decode()


message = FieldDescriptorProto(name="order_id", number=1, json_name="orderId")


@kafka_consumer(schema_config=SchemaConfig(value_schema_type="PROTOBUF", value_schema=FieldDescriptorProto))
def handler(event: ConsumerRecords, context):
    return next(event.records).value


for indexes in ([0], [1, 0], [2]):
    event = {
        "eventSource": "aws:kafka",
        "records": {
            "orders-0": [
                {
                    "topic": "orders", "partition": 0, "offset": 1, "timestamp": 1, "timestampType": "CREATE_TIME",
                    "key": None, "headers": [],
                    "value": confluent_framed(message, indexes),
                    "valueSchemaMetadata": {"schemaId": "1", "dataFormat": "PROTOBUF"},
                },
            ],
        },
    }
    try:
        print(f"message index {indexes}: {handler(event, None)}")
    except Exception as exc:
        print(f"message index {indexes}: {type(exc).__name__}: {exc}")
Possible Solution

Decode the array as zigzag varints: read the length with _DecodeVarint and apply (n >> 1) ^ -(n & 1), then skip that many zigzag varints. Also assert userId and name in test_confluent_schema_registry_complex_schema, and add a parametrised test for indexes [0], [1], [2], [1, 0] and a multi-byte index. I have this ready with tests (all Kafka tests pass) and am happy to open the PR.

Trade-off to flag: producers still on Confluent's deprecated format (use.deprecated.format=True, plain varints) decode correctly today, and after this change only index [0] would keep working for them. Confluent's client warns that format is slated for removal and its default is zigzag, so I'd fix the default. If you want to keep the deprecated format supported, an opt-in flag on SchemaConfig could select it. Happy to add that.

Steps to Reproduce

Run the snippet above. No Kafka or AWS access needed.

Powertools for AWS Lambda (Python) version

latest (develop @ 5785377)

AWS Lambda function runtime

3.13

Packaging format used

PyPi

Debugging logs
Output of the snippet on develop:

message index [0]: {'name': 'order_id', 'number': 1, 'json_name': 'orderId'}
message index [1, 0]: KafkaConsumerDeserializationError: Error trying to deserialize protobuf data - DecodeError: Error parsing message with type 'google.protobuf.FieldDescriptorProto': Wire format was corrupt
message index [2]: KafkaConsumerDeserializationError: Error trying to deserialize protobuf data - DecodeError: Error parsing message with type 'google.protobuf.FieldDescriptorProto': Wire format was corrupt
Lingua principale
Python
Stelle
3.3k
Fork
511
Merge medio
3g 8h
PR unite (30g)
92

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di aws-powertools/powertools-lambda-python

Tutte le issue di aws-powertools/powertools-lambda-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.