Repository navigation
fix(gateways): accept librdkafka 2.x Metadata v9 trailing placeholder bytes - #4839
himanchalkaushale wants to merge 3 commits into
Conversation
… bytes Resolves apache#4828. Related: apache#4299. When librdkafka 2.x clients (e.g. kcat -L) request metadata for all topics ( otal_topic_cnt == 0) at flexible versions (ApiVersion >= 9), librdkafka reserves a 4-byte int32_t 0 array-count placeholder ( d_kafka_buf_write_arraycnt_pos) but omits finalizing the count via d_kafka_buf_finalize_arraycnt. This leaves 4 zero bytes in place of a 1-byte compact array varint null (0x00) and fails to compact the buffer by 3 bytes. Iggy's decode_exhaustive() was previously rejecting these requests with 'unexpected trailing bytes in request body' and closing the connection. Real Kafka brokers accept these requests because they do not reject unconsumed bytes after deserializing schema fields. This change: 1. Adds decode_metadata_exhaustive and decode_metadata_guarded in iggy_kafka_wire::codec that tolerates up to 3 unconsumed zero bytes specifically for flexible all-topics Metadata requests, matching real broker behavior. 2. Preserves strict bounds validation (validate_metadata_shape) before decoding. 3. Preserves strict exhaustive decoding for all other Kafka APIs (Produce, Fetch, CreateTopics, etc.) and for malformed/non-zero trailing data. 4. Upgrades Metadata decode failure logging from debug to structured warn with API key and version context. 5. Updates gateways/kafka/docs/SCOPE.md. 6. Adds unit and pipeline regression tests for the captured librdkafka 30-byte Metadata frame.
|
Thanks for the PR. It is labeled If AI assistance was used to create this PR, you are required to run the Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (0.00%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## master #4839 +/- ##
============================================
+ Coverage 90.70% 91.35% +0.64%
+ Complexity 3879 3876 -3
============================================
Files 1583 1451 -132
Lines 391572 334537 -57035
Branches 325425 268392 -57033
============================================
- Hits 355193 305601 -49592
+ Misses 29064 22721 -6343
+ Partials 7315 6215 -1100
🚀 New features to boost your workflow:
|
krishvishal
left a comment
There was a problem hiding this comment.
The v9 decode change looks right to me. The partitions/journal commit (0926415) breaks WAL reopen after a restart though, and CI is red on it (clippy + 2 partitions tests).
Please drop that commit and the server.rs Windows consts, and keep this PR to the Metadata fix.
| mut synced_files: BTreeSet<PathBuf>, | ||
| ) -> io::Result<()> { | ||
| synced_files.extend(Self::sync_files(offset_files, barriers).await?); | ||
| if let (Some(commands), Some(checksum)) = (logs.commands.as_mut(), checksum) |
There was a problem hiding this comment.
Nothing reclaims the commands log on checkpoint anymore. It keeps growing until appends fail with "partition WAL capacity exhausted".
Same revert fixes it, checkpoint_at is what frees that space.
| .first_op_above(commands.checkpoint_op()) | ||
| .is_some_and(|op| op <= through_op) | ||
| { | ||
| commands.advance_checkpoint(through_op, checksum).await?; |
There was a problem hiding this comment.
advance_checkpoint moves the checkpoint in memory only and leaves the old records in the file. Once commands.sync() writes it out, the WAL won't open on restart ("partition WAL prepare chain is broken", CI test 2/4).
Please go back to checkpoint_at after the data checkpoint, like master.
| "stopped between the two partition log checkpoints", | ||
| )); | ||
| } | ||
| if let Some(checksum) = checksum { |
There was a problem hiding this comment.
I don't think the command loss this reorder guards against can happen. Boot restores the machines from the retry checkpoint taken at the data checkpoint (iggy_partition.rs:2158), so data first was already safe.
If you hit a real loss, please add a test that fails on master. Otherwise drop this commit.
| let (persistence, prepares) = PartitionPersistence::open(&wal, 42, 7).await.unwrap(); | ||
| assert_eq!(persistence.checkpoint_op(), 4); | ||
| assert_eq!(ops(&prepares), vec![4, 5]); | ||
| assert_eq!(persistence.checkpoint_op(), 0); |
There was a problem hiding this comment.
This test now expects checkpoint 0. It no longer covers a crash after the data log checkpoint.
Please restore checkpoint_op() == 4 and ops [4, 5].
| self.rewrite(through_op, checksum, None, None).await | ||
| } | ||
|
|
||
| pub async fn advance_checkpoint(&mut self, through_op: u64, checksum: u128) -> io::Result<()> { |
There was a problem hiding this comment.
CI clippy fails here: async with no .await, and no # Errors doc.
This fn goes away with the revert anyway.
| _ => {} | ||
| } | ||
| self.state.anchor_known = true; | ||
| self.state.checkpoint = through_op; |
There was a problem hiding this comment.
Nothing here writes the new checkpoint to disk, and given_no_command_below_a_checkpoint_when_checkpointing_should_leave_the_command_log now fails at persistence.rs:2898 (0 instead of 2, CI test 1/4).
Revert, then run cargo test -p partitions --lib persistence::tests before pushing.
| use tokio_util::task::TaskTracker; | ||
| use tracing::{debug, error, info, warn}; | ||
|
|
||
| #[cfg(unix)] |
There was a problem hiding this comment.
Title and description only cover the Metadata fix, but the PR also changes partition WAL checkpointing (0926415) and adds these Windows consts.
Please take both out of this PR. They need their own issue and PR.
| Ok(requested) => requested, | ||
| Err(error) => { | ||
| tracing::debug!( | ||
| tracing::warn!( |
There was a problem hiding this comment.
Clients control this input. With warn! here and at line 238 anyone can flood the logs, which is why the other handlers stick to debug! (handlers/mod.rs:355).
Please keep debug! in both places and put the removed comment back.
126ab02 to
63bd637
Compare
|
Thanks for the review @krishvishal I've addressed all the requested changes in the latest push. Let me know if you spot anything else. |
Which issue does this PR address?
Closes #4828
Relates to #4299
Rationale
kcat -Land otherlibrdkafka2.x clients fail to retrieve metadata from Apache Iggy's Kafka gateway because flexible all-topics Metadata requests are rejected withunexpected trailing bytes in request body. Accepting the verified uncompacted placeholder bytes restores compatibility withlibrdkafkawhile preserving strict decoding elsewhere.What changed?
librdkafka2.x all-topics Metadata v9 requests reserve a 4-byteint32_t0 array-count placeholder without compacting it to a 1-byte compact array varint null (0x00), leaving 3 trailing zero bytes that Iggy'sdecode_exhaustive()rejected as malformed. Real Kafka brokers accept these requests because they do not enforce a trailing-byte check after reading schema fields.The fix introduces
decode_metadata_guardediniggy_kafka_wire::codecto accept up to 3 unconsumed zero bytes strictly on flexible all-topics Metadata requests, while preserving pre-decode bounds guarding and strict exhaustive decoding everywhere else. Decode failure logging inmetadata.rsis upgraded fromdebugto structuredwarn, and regression tests covering the captured 30-byte librdkafka frame were added.Local Execution
cargo test -p iggy_kafka_wire --lib(215 passed, 0 failed)cargo test -p iggy-gateway-kafka --lib protocol::handlers::metadata(29 passed, 0 failed)cargo test -p iggy-gateway-kafka --test api_handler_tests(42 passed, 0 failed)cargo fmt --all -- --check(clean)cargo clippy -p iggy_kafka_wire -p iggy-gateway-kafka --all-targets(0 warnings)AI Usage
librdkafkaC source (src/rdkafka_request.c), implementation of scoped metadata decoder incodec.rs, error log improvement, and regression test cases.cargo test,cargo fmt, andcargo clippylocally across both crates; verified bounds guard rejection on truncated/malformed frames and verified exact field deserialization for captured 30-byte librdkafka wire bytes.