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

Ouverte
  • Quelque chose ne va pas ?
#8,665 4 commentaires 0 réactions 1 personne assignée Voir sur GitHub

Les mainteneurs répondent en général sous 1 jour

@alphasafal y travaille déjà.

Depuis le 25/10/2026.

  • #8670 par @alphasafal — ouverte

Évaluation

Cette issue n'a pas encore été évaluée.

Description

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
Langage dominant
Python
Étoiles
3.3k
Forks
511
Merge moyen
3 j 8 h
PR mergées (30 j)
92

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de aws-powertools/powertools-lambda-python

Toutes les issues de aws-powertools/powertools-lambda-python

Issues similaires

Plus d'issues Python

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.