This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch kafka_idempotence_stub in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6cf16a02fc9af1d1ac7be56cbeb4223a3db8d7ae Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Sep 21 15:04:26 2026 +0200 first round of review --- gateways/kafka/docs/IDEMPOTENCE.md | 10 +++- gateways/kafka/docs/MANUAL_TESTING.md | 17 +++--- gateways/kafka/docs/SCOPE.md | 7 ++- gateways/kafka/docs/kafka_api_keys_reference.md | 13 +++-- gateways/kafka/src/main.rs | 29 ++++++++++ gateways/kafka/src/protocol/bounds_guard.rs | 30 ++++++++++ gateways/kafka/src/protocol/handlers/produce.rs | 6 ++ gateways/kafka/src/server.rs | 7 ++- gateways/kafka/tests/idempotence_tests.rs | 77 +++++++++++++++++++++++-- 9 files changed, 170 insertions(+), 26 deletions(-) diff --git a/gateways/kafka/docs/IDEMPOTENCE.md b/gateways/kafka/docs/IDEMPOTENCE.md index 0c346a207..2d135739a 100644 --- a/gateways/kafka/docs/IDEMPOTENCE.md +++ b/gateways/kafka/docs/IDEMPOTENCE.md @@ -124,7 +124,11 @@ a birthday collision, not a remote one. The id is a pool key, not a dedup identity. Under the design above, the dedup identity is the session's own random client id, minted at register. The producer id only decides which connection serves a producer. Kafka still requires it to be unique across the cluster, which is what the -instance number buys. It does not have to survive a restart. +instance number buys. It does not have to survive a restart while nothing keys state on it, +which is true of allocate-only: the gateway hands an id out and forgets it. That stops being true +the moment the pool lands or Produce persists, because a restarted allocator replays ids a live +producer still holds, and `producer_epoch` is always 0 so the pair cannot tell the generations +apart. Treat generation reuse as a pool blocker, not a detail. An empty `transactional_id` reads as absent. A wire null decodes to `None`, but `kafka-protocol`'s own `Default` is `Some("")`, and a producer that is idempotent-only names no @@ -179,7 +183,9 @@ to every consumer. ## Invariants this design rests on -Both are absences, so nothing fails loudly if they are lost. +Both are absences. Losing either is caught: `golden_wire_fixtures_tests.rs` pins the ApiVersions +v1 and v3 bodies byte-exactly, so adding a finalized feature or any advertised key fails both +goldens. **Never advertise `transaction.version >= 2` in the ApiVersions `finalized_features`.** `TransactionManager.maybeUpdateTransactionV2Enabled` reads it, and under TV2 `maybeAddPartition` diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index 04e82d0c9..b77d864ee 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -38,13 +38,14 @@ kafka listener bound on 127.0.0.1:9093 ```bash # Terminal 2 -# Keys 0/1/2/19 match ci-wire-fixtures.sh: the only keys any test actually loads a .bin -# fixture for. Metadata (3) and ApiVersions (18) requests are built synthetically in-test -# instead, so fixtures for those keys are generated but unused - `generate` still accepts -# them if you want them for manual `send`/`verify` below. +# Keys 0/1/2/19/22 match ci-wire-fixtures.sh's FIXTURE_API_KEYS: the only keys any test +# actually loads a .bin fixture for. Omit key 22 and the six InitProducerId cases skip +# silently, which reads as a pass. Metadata (3) and ApiVersions (18) requests are built +# synthetically in-test instead, so fixtures for those keys are generated but unused - +# `generate` still accepts them if you want them for manual `send`/`verify` below. cargo run -p kafka-message-gen -- generate \ --output gateways/kafka/tools/kafka-tool/kafka_messages \ - --api-key 0 --api-key 1 --api-key 2 --api-key 19 + --api-key 0 --api-key 1 --api-key 2 --api-key 19 --api-key 22 ``` --- @@ -269,14 +270,14 @@ Tester: ___________ Gateway commit: ___________ kcat version (if used): ___________ -[ ] A1–A9 Smoke tests -[ ] B1–B4 Version firewall (all 6 keys × 4 boundary versions) +[ ] A1–A10 Smoke tests +[ ] B1–B4 Version firewall (all 7 keys × 4 boundary versions) [ ] C1–C4 Unsupported API keys [ ] D1–D10 Flexible vs legacy encoding [ ] E1–E4 Metadata stub semantics [ ] F1–F6 TCP / connection behavior [ ] G1–G3 kcat client (record errors for G2/G3) -[ ] H1–H3 Adversarial input +[ ] H1–H6 Adversarial input Automated regression: [ ] cargo test -p iggy-gateway-kafka — all passed (see `TEST_SUITE.md` for why this checklist diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index 44d05a20a..d5084c1f9 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -12,7 +12,7 @@ Foundation layer only: a TCP listener on the Kafka wire port that decodes reques | Length-prefixed frame read/write with `max_frame_size` cap | Done | `src/server.rs` | | Request header v1/v2 auto-detection | Done | `src/protocol/header.rs` (delegates to `kafka_protocol::messages::ApiKey`) | | Version negotiation firewall (`SUPPORTED_RANGES`) | Done | `src/protocol/api.rs` | -| Request decode + stub encode for 6 API keys | Done | `src/protocol/api.rs`, `responses.rs` (via the `kafka_protocol` crate) | +| Request decode + stub encode for 7 API keys | Done | `src/protocol/api.rs`, `src/protocol/handlers/` (via the `kafka_protocol` crate) | | Produce hot path: RecordBatch as opaque `Bytes` | Done | `src/protocol/responses.rs` | | Pre-decode bounds guard against unbounded allocation | Done | `src/protocol/bounds_guard.rs` | | Graceful errors (corrupt decode, invalid header) | Done | `src/protocol/api.rs`, `src/server.rs` | @@ -101,7 +101,10 @@ Three things enforce that, in the order a client meets them: throws and `NetworkClient.doSend` keeps the request off the wire; librdkafka's four request builders return `__UNSUPPORTED_FEATURE`, which is fatal there. 2. **InitProducerId (22) with a `transactional_id`** answers `UNSUPPORTED_VERSION` (35), so a - producer that got past step 1 fails before it can open a transaction. + producer that got past step 1 fails before it can open a transaction. Terminal on the Java + client, whose `InitProducerIdHandler` treats any unrecognised code as fatal. Not terminal on + librdkafka, which retries 35 here indefinitely; librdkafka is stopped by step 1 instead, and + [`IDEMPOTENCE.md`](IDEMPOTENCE.md) records why that matters when FindCoordinator is advertised. 3. **Produce with a non-empty `transactional_id`** answers `UNSUPPORTED_VERSION` (35) per partition, so a raw client that skipped both earlier gates still cannot write transactional records. `acks=0` stays silent, and no case closes the connection. diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md index 58baa0e43..19c004215 100644 --- a/gateways/kafka/docs/kafka_api_keys_reference.md +++ b/gateways/kafka/docs/kafka_api_keys_reference.md @@ -264,8 +264,9 @@ Key new minimums: | Category | Count | Notes | | ---------- | :-----: | ------- | | 🔴 Bridge (data path) | 7 | Produce, Fetch, Metadata, SaslHandshake, ApiVersions, SaslAuthenticate, ShareFetch | -| 🟠 Required Stub (client state machine) | 12 | ListOffsets, consumer group (8-14), CreateTopics, ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) | -| 🟡 Optional Stub (admin/observability) | 47 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely | +| 🟠 Required Stub (client state machine) | 13 | ListOffsets, consumer group (8-14), CreateTopics, InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) | +| 🟡 Optional Stub (admin/observability) | 42 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely | +| ❌ Unadvertised (transactions) | 4 | AddPartitionsToTxn (24), AddOffsetsToTxn (25), EndTxn (26), TxnOffsetCommit (28). Absent from ApiVersions, so a conforming client never sends one | | ❌ Reject (broker/KRaft internal) | 22 | Return `INVALID_REQUEST` with valid frame — never close the TCP connection | | **Total API Keys in this document** | **88** | Key IDs 0-88 with a gap at 73 | @@ -284,12 +285,12 @@ Key new minimums: | ApiVersions | v0-v3 | v4 | 1 version behind | | CreateTopics | v2-v5 | v7 | 2 versions behind | -### Missing from `SUPPORTED_RANGES` (82 of the 88 API keys in this document) +### Missing from `SUPPORTED_RANGES` (81 of the 88 API keys in this document) Every key not in `SUPPORTED_RANGES` closes the connection - the same policy applied to every -other unlisted key, not a special case for these. No api-specific response schema exists for an -unlisted key, so any body the gateway could send would be misparsed by the client against the -schema it expected. This includes: +other unlisted key, not a special case for these. The gateway declines to define a response for a +key it does not advertise, and a conforming client never sends one, so no response shape has to +be agreed. This includes: - **Client bootstrap blockers**: OffsetCommit (8), OffsetFetch (9), FindCoordinator (10) - **Classic consumer group protocol**: JoinGroup (11), Heartbeat (12), LeaveGroup (13), SyncGroup (14) diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs index 46ff2b6b1..ce1763293 100644 --- a/gateways/kafka/src/main.rs +++ b/gateways/kafka/src/main.rs @@ -341,6 +341,35 @@ mod tests { ); } + /// A bad instance id must fail startup loudly. Silently defaulting to 0 would let two + /// gateways mint colliding producer ids, which Kafka requires to be unique cluster-wide, + /// and the message has to name the variable and the offending value or an operator cannot + /// act on it. + #[test] + #[serial] + fn given_an_unparseable_instance_id_when_loading_config_should_reject_and_name_it() { + for raw in ["abc", "-1", "", " 7", "7.0"] { + unsafe { + std::env::set_var("IGGY_KAFKA_INSTANCE_ID", raw); + } + let loaded = load_config(); + unsafe { + std::env::remove_var("IGGY_KAFKA_INSTANCE_ID"); + } + let error = loaded.err().unwrap_or_else(|| { + panic!("instance id `{raw}` must be rejected, not silently defaulted to 0") + }); + assert!( + error.contains("IGGY_KAFKA_INSTANCE_ID"), + "`{raw}` rejection must name the variable, got: {error}" + ); + assert!( + error.contains(raw), + "`{raw}` rejection must quote the offending value, got: {error}" + ); + } + } + #[test] fn parse_positive_rejects_zero() { assert!(parse_positive::<usize>("KEY", "0").is_err()); diff --git a/gateways/kafka/src/protocol/bounds_guard.rs b/gateways/kafka/src/protocol/bounds_guard.rs index 61f77edf1..a3795a322 100644 --- a/gateways/kafka/src/protocol/bounds_guard.rs +++ b/gateways/kafka/src/protocol/bounds_guard.rs @@ -780,6 +780,36 @@ mod tests { assert!(validate_metadata_shape(0, &body, TEST_MAX_FRAME_SIZE).is_err()); } + /// Every sibling guard carries a rejection POC; without one, short-circuiting this guard to + /// `Ok(())` leaves the whole suite green, so nothing proved it rejected a hostile frame. + #[test] + fn init_producer_id_v5_huge_compact_string_rejected() { + // Compact string length varint far past the frame: nothing follows it to read. + let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]); + assert!(validate_init_producer_id_shape(5, &body).is_err()); + } + + #[test] + fn init_producer_id_v0_truncated_legacy_string_rejected() { + // Declares 32767 bytes of transactional id, supplies none. + let body = Bytes::from_static(&[0x7F, 0xFF]); + assert!(validate_init_producer_id_shape(0, &body).is_err()); + } + + #[test] + fn init_producer_id_v5_null_transactional_id_accepted() { + // Null compact string, transaction_timeout_ms, then the v3+ producer id/epoch pair + // (both -1, "no producer id"), then tagged fields. + let body = Bytes::from_static(&[ + 0x00, // transactional_id: null compact string + 0x00, 0x00, 0x75, 0x30, // transaction_timeout_ms: 30000 + 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, // producer_id: -1 + 0xFF, 0xFF, // producer_epoch: -1 + 0x00, // tagged fields + ]); + assert!(validate_init_producer_id_shape(5, &body).is_ok()); + } + #[test] fn metadata_v0_null_array_all_topics_accepted() { let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF]); // -1: all topics diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs index 4c452a6d5..8e7a1a2f4 100644 --- a/gateways/kafka/src/protocol/handlers/produce.rs +++ b/gateways/kafka/src/protocol/handlers/produce.rs @@ -147,6 +147,12 @@ pub fn encode_response(version: i16, req: &ProduceRequest) -> Result<Bytes> { /// A transactional batch must never be answered as if it were ordinary records: nothing here /// tracks a last stable offset or writes an abort marker, so an aborted transaction's records /// would reach every consumer. 35 is fatal for the producer; 42 and 43 are only abortable. +/// +/// This reads the request-level `transactional_id` only. A record batch carries its own +/// transactional bit in attributes, and the records stay opaque bytes here, so a hand-built +/// frame setting the bit without the request field still gets the retriable stub error. Java +/// and librdkafka both set the request field whenever they set the batch bit, so no real client +/// reaches that gap; it has to close before records are ever persisted. fn partition_error_code(req: &ProduceRequest) -> i16 { if is_transactional(req.transactional_id.as_ref()) { ERROR_UNSUPPORTED_VERSION diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs index 066d3701e..8f369532f 100644 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@ -205,9 +205,12 @@ impl KafkaGateway { ) -> Result<()> { let local_addr = listener.local_addr()?; let broker = BrokerAdvertise::from_server_config(&self.config, local_addr)?; + // instance_id is logged because it is the only way to tell from a running process which + // half of the producer-id space this gateway owns. Two gateways left on the default + // collide silently, and a config file cannot be diffed against a live deployment. info!( - "kafka listener bound on {} (advertised as {}:{})", - local_addr, broker.host, broker.port + "kafka listener bound on {} (advertised as {}:{}, instance id {})", + local_addr, broker.host, broker.port, self.config.instance_id ); let state = Arc::new(GatewayState::new( broker, diff --git a/gateways/kafka/tests/idempotence_tests.rs b/gateways/kafka/tests/idempotence_tests.rs index 1d879ebdd..2a03b7e67 100644 --- a/gateways/kafka/tests/idempotence_tests.rs +++ b/gateways/kafka/tests/idempotence_tests.rs @@ -34,15 +34,16 @@ use bytes::{BufMut, Bytes, BytesMut}; use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; +use iggy_gateway_kafka::GatewayConfig; use iggy_gateway_kafka::protocol::api::{ API_KEY_API_VERSIONS, API_KEY_INIT_PRODUCER_ID, API_KEY_PRODUCE, ERROR_NONE, - ERROR_UNSUPPORTED_VERSION, GatewayState, handle_request, handle_request_bounded, - is_supported_version, + ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNSUPPORTED_VERSION, GatewayState, handle_request, + handle_request_bounded, is_supported_version, }; use codec::Decoder; use scope::default_broker; -use server::spawn_test_server; +use server::{spawn_test_server, spawn_test_server_with_config}; use tcp::{ByteRead, build_request_frame, read_byte_with_timeout, round_trip}; use wire::{build_api_versions_flexible_request, build_init_producer_id_request}; @@ -180,6 +181,14 @@ async fn given_a_transactional_id_when_init_producer_id_should_answer_unsupporte response.error_code, ERROR_UNSUPPORTED_VERSION, "v{version} must refuse a transactional producer" ); + // A real broker's error path leaves both at their schema defaults, and -1 is the + // "no producer id" sentinel the whole bit-63-clear layout exists to keep distinct + // from a real allocation. + assert_eq!( + response.producer_id, -1, + "v{version} a refusal must not hand back an allocated id" + ); + assert_eq!(response.producer_epoch, 0, "v{version} producer_epoch"); } } @@ -220,10 +229,10 @@ async fn given_no_transactional_id_when_producing_should_keep_the_retriable_stub ) .await .expect_response("acks=1 expects a response"); - assert_ne!( + assert_eq!( first_produce_partition_error(&body), - ERROR_UNSUPPORTED_VERSION, - "a non-transactional produce must not inherit the transactional refusal" + ERROR_NOT_LEADER_OR_FOLLOWER, + "a non-transactional produce keeps the retriable stub error, not the transactional refusal" ); } @@ -351,6 +360,62 @@ async fn given_a_live_server_when_a_transactional_request_is_refused_should_keep } } +/// Config value has to reach the allocator, not just the struct. Nothing else pins that hop: +/// one test pins env to config, another pins `GatewayState::new` to the high bits, and the +/// server's own `config.instance_id` argument sat between them uncovered. +#[tokio::test] +async fn given_a_server_configured_with_an_instance_id_when_init_producer_id_should_reflect_it() { + let instance_id = 0x0042u16; + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: MAX_FRAME_SIZE, + max_connections: 1024, + idle_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(5), + write_timeout: Duration::from_secs(5), + shutdown_drain_timeout: Duration::from_secs(5), + instance_id, + }) + .await; + + let request = build_init_producer_id_request(4, None); + let (_correlation_id, body) = + round_trip(addr, API_KEY_INIT_PRODUCER_ID, 4, 7_200, &request).await; + let response = decode_init_producer_id_response(4, &body); + assert_eq!(response.error_code, ERROR_NONE); + assert_eq!( + response.producer_id >> COUNTER_BITS, + i64::from(instance_id), + "the configured instance number must reach the allocator, not stop at the config struct" + ); +} + +/// One allocator per process, shared across connections. Building a `GatewayState` per accepted +/// connection instead would restart every counter at 0 and put duplicate producer ids on the +/// wire, which Kafka requires to be unique; the in-process test above cannot see that because it +/// never opens a second connection. +#[tokio::test] +async fn given_one_server_when_two_connections_init_should_receive_distinct_ids() { + let (addr, _shutdown) = spawn_test_server().await; + let request = build_init_producer_id_request(4, None); + + let (_first_id, first_body) = + round_trip(addr, API_KEY_INIT_PRODUCER_ID, 4, 7_300, &request).await; + let (_second_id, second_body) = + round_trip(addr, API_KEY_INIT_PRODUCER_ID, 4, 7_301, &request).await; + + let first = decode_init_producer_id_response(4, &first_body); + let second = decode_init_producer_id_response(4, &second_body); + assert_eq!(first.error_code, ERROR_NONE); + assert_eq!(second.error_code, ERROR_NONE); + assert_ne!( + first.producer_id, second.producer_id, + "separate connections must draw from one allocator" + ); +} + #[tokio::test] async fn given_a_live_server_when_init_producer_id_round_trips_should_return_an_allocated_id() { let (addr, _shutdown) = spawn_test_server().await;
