Fix data loss reading legacy (magic v0/v1) topics over flexible Fetch - #5505
Open
Giuseppe Lillo (giuseppelillo) wants to merge 2 commits into
Open
Fix data loss reading legacy (magic v0/v1) topics over flexible Fetch#5505Giuseppe Lillo (giuseppelillo) wants to merge 2 commits into
Giuseppe Lillo (giuseppelillo) wants to merge 2 commits into
Conversation
Since v2.5.0 the consumer negotiates flexible Fetch (Fetch v12+, KIP-482) whenever the broker supports it, which sets RD_KAFKA_OP_F_FLEXVER on the FetchResponse buffer. The Records/MessageSet payload inside a FetchResponse is encoded in the fixed Kafka record format, independent of the response's flexible framing: legacy magic v0/v1 Message Key/Value use plain int32 length prefixes, not compact (uvarint) ones. rd_kafka_buf_read_kbytes() is flex-sensitive, so leaving FLEXVER set while parsing decoded those length prefixes as compact (uvarint) bytes and misaligned the entire parse. The result was silent data loss on legacy-format topics: records delivered with NULL key/value, spurious "Unsupported MagicByte" errors at bogus offsets, collapsed throughput, and assertion failures in debug builds. magic v2 (RecordBatch) topics were unaffected. Clear RD_KAFKA_OP_F_FLEXVER for the duration of message-set parsing in rd_kafka_msgset_parse() and restore it afterwards, so the payload is always parsed as a non-flexible structure regardless of the FetchResponse framing.
|
Please sign the Contributor License Agreement here before this PR can be approved. |
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds a regression fix and test coverage for consuming legacy (magic v0/v1) message sets over flexible Fetch responses, and extends the mock cluster to allow verbatim message-set injection for targeted protocol tests.
Changes:
- Fix message-set parsing by ensuring legacy MessageSet Key/Value lengths are decoded as fixed int32 even when FetchResponse is flexible-version framed.
- Add a mock-cluster API/command to append a raw (verbatim) message set to a partition log for tests.
- Introduce a new regression test (0154) and wire it into the test runner/CMake; update changelog entry for v2.14.3.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test.c | Registers the new 0154 mock-based regression test in the test runner. |
| tests/CMakeLists.txt | Adds the new test source file to the tests build. |
| tests/0154-legacy_msgver_flexfetch_mock.c | New regression test that pushes a handcrafted magic v1 message set and validates consumption via flexible Fetch. |
| src/rdkafka_op.h | Adds mock command enum value and payload fields for the new “push raw msgset” operation. |
| src/rdkafka_op.c | Ensures mock op payload (data) is freed on op destroy. |
| src/rdkafka_msgset_reader.c | Clears/restores FLEXVER flag while parsing message sets to avoid compact-length decoding. |
| src/rdkafka_mock.h | Exposes rd_kafka_mock_partition_push_msgset_raw() API for verbatim message-set injection. |
| src/rdkafka_mock.c | Implements raw message-set push and avoids MsgVersion 2 header rewriting for verbatim message sets. |
| CHANGELOG.md | Documents the consumer fix in the v2.14.3 release notes. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+314
to
+334
| /** | ||
| * @brief Append a verbatim message set (\p msgset of \p size bytes, | ||
| * containing \p msgcnt messages) to the partition log exactly as | ||
| * provided, without the MsgVersion 2 RecordBatch header rewrite that | ||
| * the normal Produce path performs. | ||
| * | ||
| * This is primarily intended for tests that need the broker to serve a | ||
| * message set in a specific on-disk format (e.g. a legacy magic v0/v1 | ||
| * message set) to a modern consumer. The caller is responsible for baking | ||
| * the correct absolute offsets into the message set; push into a fresh | ||
| * partition so the first message's offset is 0. | ||
| * | ||
| * @param mcluster Mock cluster instance. | ||
| * @param topic Topic to append to (auto-created if needed). | ||
| * @param partition Partition to append to. | ||
| * @param msgset Raw message set bytes. | ||
| * @param size Size of \p msgset in bytes. | ||
| * @param msgcnt Number of messages contained in \p msgset. | ||
| * | ||
| * @return Push operation error code. | ||
| */ |
Comment on lines
+2261
to
+2280
| rd_kafka_resp_err_t | ||
| rd_kafka_mock_partition_push_msgset_raw(rd_kafka_mock_cluster_t *mcluster, | ||
| const char *topic, | ||
| int32_t partition, | ||
| const void *msgset, | ||
| size_t size, | ||
| int32_t msgcnt) { | ||
| rd_kafka_op_t *rko; | ||
|
|
||
| if (size > INT32_MAX) | ||
| return RD_KAFKA_RESP_ERR__INVALID_ARG; | ||
|
|
||
| rko = rd_kafka_op_new(RD_KAFKA_OP_MOCK); | ||
|
|
||
| rko->rko_u.mock.name = rd_strdup(topic); | ||
| rko->rko_u.mock.cmd = RD_KAFKA_MOCK_CMD_PART_PUSH_MSGSET_RAW; | ||
| rko->rko_u.mock.partition = partition; | ||
| rko->rko_u.mock.data = size > 0 ? rd_memdup(msgset, size) : NULL; | ||
| rko->rko_u.mock.size = size; | ||
| rko->rko_u.mock.lo = msgcnt; |
Comment on lines
+53
to
+56
| static void buf_push(uint8_t *buf, size_t *of, const void *data, size_t len) { | ||
| memcpy(buf + *of, data, len); | ||
| *of += len; | ||
| } |
Comment on lines
+154
to
+155
| buf = rd_malloc(1024 * 1024); | ||
| for (i = 0; i < msgcnt; i++) { |
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
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 #5504