Skip to content

fix(kafka): decode Confluent Protobuf message indexes as zigzag varints - #8670

Open
alphasafal wants to merge 1 commit into
aws-powertools:developfrom
alphasafal:fix/kafka-protobuf-confluent-zigzag-message-index
Open

alphasafal wants to merge 1 commit into
aws-powertools:developfrom
alphasafal:fix/kafka-protobuf-confluent-zigzag-message-index

Conversation

@alphasafal

Copy link
Copy Markdown

Issue number: closes #8665

Summary

Changes

Confluent's Protobuf serializer writes the message index array as zigzag encoded varints (array length, then each index), with [0] as a single 0x00 byte. ProtobufDeserializer._remove_message_index read it with _DecodeSignedVarint, so it read twice the real length and skipped payload bytes as if they were indexes. Records for any message type other than the first in the schema failed with DecodeError, or decoded with fields silently missing.

This change:

  • reads the array length and each index with _DecodeVarint plus zigzag decoding ((n >> 1) ^ -(n & 1)), in a small _decode_zigzag_varint helper
  • raises ValueError for a negative array length (wrapped as KafkaConsumerDeserializationError by deserialize)
  • fixes the default without a flag, as agreed in the issue

Tests in tests/functional/kafka_consumer/_protobuf/test_kafka_consumer_with_protobuf.py:

  • test_confluent_schema_registry_complex_schema now also asserts userId and name. The fixture record with the 04 02 00 prefix was losing them, and the test only checked email and age.
  • new test_confluent_schema_registry_message_index_is_zigzag_decoded, parametrised over indexes [0], [1], [2], [1, 0] and [70] (a multi-byte index), framed the same way confluent-kafka's serializer frames them

The tightened test and four of the five new cases fail on develop ([0] already worked) and pass with this change. All Kafka tests pass (60), and ruff format / ruff check are clean.

User experience

Before, for a Confluent-framed record whose message type is not the first in its schema:

message index [1, 0]: KafkaConsumerDeserializationError: ... DecodeError: ... Wire format was corrupt
message index [2]: KafkaConsumerDeserializationError: ... DecodeError: ... Wire format was corrupt

and the repo's own fixture record with index [1, 0] decoded without userId and name.

After, every index decodes into all fields:

message index [0]: {'name': 'order_id', 'number': 1, 'json_name': 'orderId'}
message index [1, 0]: {'name': 'order_id', 'number': 1, 'json_name': 'orderId'}
message index [2]: {'name': 'order_id', 'number': 1, 'json_name': 'orderId'}

(output of the snippet in #8665)


By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Disclaimer: We value your time and bandwidth. As such, any pull requests created on non-triaged issues might not be successful.

Confluent's Protobuf serializer writes the message index array as zigzag
encoded varints (array length, then each index), with [0] as a single 0x00
byte. ProtobufDeserializer read it with _DecodeSignedVarint, so it read
twice the real length and consumed payload bytes as indexes. Records for
any message type other than the first in the schema either failed to
decode or decoded with fields silently missing.

Decode the array as zigzag varints. Tighten
test_confluent_schema_registry_complex_schema to assert userId and name,
which the 04 02 00 fixture record was losing, and add a parametrised test
for indexes [0], [1], [2], [1, 0] and a multi-byte index.

Closes aws-powertools#8665
@alphasafal
alphasafal requested a review from a team as a code owner October 25, 2026 19:39
@alphasafal
alphasafal requested review from leandrodamascena and a balanced review from Copilot October 25, 2026 19:39
@boring-cyborg boring-cyborg Bot added the tests label Oct 25, 2026
@powertools-for-aws-oss-automation powertools-for-aws-oss-automation Bot added the size/M Denotes a PR that changes 30-99 lines, ignoring generated files. label Oct 25, 2026

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota.

@sonarqubecloud

Copy link
Copy Markdown

@leandrodamascena leandrodamascena added this to the 3.40.0 milestone Oct 25, 2026

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

size/M Denotes a PR that changes 30-99 lines, ignoring generated files. tests

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

3 participants