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

Offen
  • Stimmt etwas nicht?
#8,665 4 Kommentare 0 Reaktionen 1 zugewiesene Person Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

@alphasafal arbeitet bereits daran.

Seit 25.10.2026.

  • #8670 von @alphasafal — offen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Beschreibung

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
Vorherrschende Sprache
Python
Sterne
3.3k
Forks
511
Ø Merge
3 T. 4 Std.
Gemergte PRs (30 T.)
98

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus aws-powertools/powertools-lambda-python

Alle Issues in aws-powertools/powertools-lambda-python

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.