Fix NULL key/value when consuming legacy MessageSets with flexible Fetch versions (v12+) - #5550
Open
Thomas (delthas) wants to merge 3 commits into
Open
Conversation
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
Thomas (delthas)
force-pushed
the
fix/legacy-msgset-flexver-fetch
branch
2 times, most recently
from
July 15, 2026 12:47
79f8b8f to
55a8d8d
Compare
Thomas (delthas)
marked this pull request as ready for review
July 15, 2026 12:48
Ankith L (Ankith-Confluent)
self-requested a review
July 22, 2026 10:23
Member
|
/sem-approve |
Ankith L (Ankith-Confluent)
requested changes
Jul 22, 2026
Ankith L (Ankith-Confluent)
left a comment
Ankith L (Ankith-Confluent)
left a comment
Member
There was a problem hiding this comment.
Thanks for the PR Thomas (@delthas)!
The PR looks good. Just one minor comment from my side.
Thomas (delthas)
force-pushed
the
fix/legacy-msgset-flexver-fetch
branch
from
July 28, 2026 09:31
55a8d8d to
9e03b4e
Compare
Member
|
/sem-approve |
Member
|
Hi Thomas (@delthas) |
…tch versions Since the Fetch RPC was upgraded to flexible versions (v12+, KIP-951 prework, v2.5.0), messages stored in the legacy MessageSet format (MsgVersion v0..v1, brokers or topics with log.message.format.version < 0.11) were returned to the application with NULL key and value. The legacy MessageSet reader used rd_kafka_buf_read_kbytes() to read the Message Key and Value, which parses a compact (varint-prefixed) length when the buffer is flagged as a flexible-version response. Legacy MessageSets always use fixed-width (int32-prefixed) encodings regardless of the FetchResponse version, so the first 0x00 length byte was read as a varint compact NULL, and the reader then drifted into the middle of the message, misparsing the rest of the MessageSet with errors such as 'Unsupported Message(Set) MagicByte'. Add rd_kafka_buf_read_kbytes_fixed() which always reads the fixed-width encoding, and use it for the legacy Message Key and Value.
The mock broker rejected any produced MessageSet with a MagicByte other than 2 with UNSUPPORTED_VERSION, making it impossible to test consuming from clusters with log.message.format.version < 0.11. Accept uncompressed legacy v0..v1 MessageSets: validate the MessageSet layout, count its messages, and assign an absolute log offset to each message in the set. Compressed legacy MessageSets are still rejected with UNSUPPORTED_VERSION.
Regression test for consuming messages stored in the legacy MessageSet format (MsgVersion v0..v1) with flexible Fetch versions (v12+): cap the ProduceRequest version on the mock cluster so the producer writes MsgVersion v1 (Produce <= v2) or v0 (Produce <= v1) MessageSets, then verify that a consumer fetching with a flexible Fetch version receives all messages with intact key and value.
Thomas (delthas)
force-pushed
the
fix/legacy-msgset-flexver-fetch
branch
from
July 28, 2026 14:06
9e03b4e to
e7ff1d0
Compare
Author
|
Hi Ankith L (@Ankith-Confluent), Done. |
Member
|
/sem-approve |
Ankith L (Ankith-Confluent)
requested review from
Kaushik Raina (k-raina)
and removed request for
Ojasva Jain (ojasvajain)
August 4, 2026 05:05
Ankith L (Ankith-Confluent)
approved these changes
Aug 4, 2026
Ankith L (Ankith-Confluent)
left a comment
Ankith L (Ankith-Confluent)
left a comment
Member
There was a problem hiding this comment.
I have approved the PR Thomas (@delthas)
LGTM!
We will merge it after another round of review.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #5549
Symptom
When consuming from a cluster or topic using the legacy message format (
log.message.format.version< 0.11, MsgVersion v0..v1), fetches succeed and offsets advance, but every message is delivered to the application with NULL key and NULL value (offsets and timestamps are correct), together with consumer errors such as:Happening since v2.5.0. On v2.5.x fetches fail/stall instead; from v2.6.0 onwards messages are delivered with NULL key/value as described.
Found in production through node-rdkafka 3.6.1 (librdkafka 2.12.0), consuming from Apache Kafka 3.9.1 brokers that pin
log.message.format.version=0.10.2.2for rolling-upgrade compatibility.Root cause
Since the Fetch RPC was upgraded to flexible versions (#4584, v2.5.0), the FetchResponse buffer is flagged as a flexible-version response. The legacy MessageSet reader (
rd_kafka_msgset_reader_msg_v0_1()) reads the Message Key and Value withrd_kafka_buf_read_kbytes(), which switches to compact (varint-prefixed) parsing when that flag is set — but legacy MessageSets always use fixed-width (int32-prefixed) encodings regardless of the Fetch version. The first0x00byte of the length prefix parses as a compact NULL, and the reader then drifts into the middle of the message bytes, misparsing the rest of the MessageSet.Fix
Add
rd_kafka_buf_read_kbytes_fixed(), which always reads the fixed-width encoding, and use it for the legacy Message Key and Value.Testing
inter.broker.protocol.version=0.10.2.2/log.message.format.version=0.10.2.2.