Repository navigation
fix(kafka): decode Confluent Protobuf message indexes as zigzag varints - #8670
alphasafal wants to merge 3 commits into
Conversation
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
|
leandrodamascena
left a comment
There was a problem hiding this comment.
Thanks @alphasafal, and good catch, the fixture with [1, 0] was silently dropping two fields and nobody noticed. I checked your reading against confluent-kafka 2.16.1 itself: framed [1], [2], [1, 0], [70], [300], [0, 3, 1] and a few more with their encoder, all of them fail or lose fields on develop and decode every field with your change. I pushed one small test for the negative length path so the new raise is covered.
I'm approving now. This goes out in 3.40.0, so it will sit here for a while before I merge it. No rebase needed in the meantime, and you'll be credited in the release notes.



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 single0x00byte.ProtobufDeserializer._remove_message_indexread 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 withDecodeError, or decoded with fields silently missing.This change:
_DecodeVarintplus zigzag decoding ((n >> 1) ^ -(n & 1)), in a small_decode_zigzag_varinthelperValueErrorfor a negative array length (wrapped asKafkaConsumerDeserializationErrorbydeserialize)Tests in
tests/functional/kafka_consumer/_protobuf/test_kafka_consumer_with_protobuf.py:test_confluent_schema_registry_complex_schemanow also assertsuserIdandname. The fixture record with the04 02 00prefix was losing them, and the test only checkedemailandage.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 themThe 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), andruff format/ruff checkare clean.User experience
Before, for a Confluent-framed record whose message type is not the first in its schema:
and the repo's own fixture record with index
[1, 0]decoded withoutuserIdandname.After, every index decodes into all fields:
(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.