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

Abierto
  • ¿Algo no está bien?
#8,665 4 comentarios 0 reacciones 1 asignado Ver en GitHub

Los mantenedores suelen responder en 1 día

@alphasafal ya está trabajando en esto.

Desde el 25/10/2026.

  • #8670 de @alphasafal — abierto

Evaluación

Este issue todavía no se ha evaluado.

Descripción

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
Lenguaje dominante
Python
Estrellas
3.3k
Forks
511
Merge medio
3 d 4 h
PR fusionados (30 d)
98

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de aws-powertools/powertools-lambda-python

Todos los issues de aws-powertools/powertools-lambda-python

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.