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

オープン
  • 何か問題がありますか？
#8,665 コメント 4 件 リアクション 0 件 担当者 1 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

@alphasafal がすでに取り組んでいます。

2026年10月25日 から。

  • #8670 @alphasafal による — オープン

評価

この issue はまだ評価されていません。

説明

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
主要言語
Python
スター
3.3k
フォーク
511
平均マージ
3日 8時間
マージ済み PR（30日）
92

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

aws-powertools/powertools-lambda-python のほかの issue

aws-powertools/powertools-lambda-python の issue をすべて見る

似ている issue

Python の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。