ethanlin01x commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3695775980
########## gateways/kafka/docs/TEST_SUITE.md: ########## @@ -0,0 +1,155 @@ +# Kafka gateway — automated regression test suite + +Regression tests live under [`tests/`](../tests/). Run from the workspace root: + +```bash +cargo test -p iggy-gateway-kafka +``` + +**Current count:** 103 tests across 12 suites (as of #3421 foundation). + +## Prerequisites + +### Wire fixtures (required for `decode_validation_tests` and some handler tests) + +```bash +./gateways/kafka/scripts/ci-wire-fixtures.sh generate +``` + +Fixtures are gitignored under `tools/kafka-tool/kafka_messages/`. CI runs the same script before `rust-gateway` test jobs and removes the directory afterward. Tests that need fixtures skip gracefully when a file is missing (`handler_regression_tests`) or panic with a clear path (`decode_validation_tests`). + +--- + +## Test file catalog + +| File | Suite focus | Test count (approx.) | Depends on fixtures | +| ------ | ------------- | ---------------------- | --------------------- | +| [`codec_tests.rs`](../tests/codec_tests.rs) | Primitive encode/decode round-trips, varint, compact strings, tagged fields | 9 | No | +| [`decode_safety_tests.rs`](../tests/decode_safety_tests.rs) | Adversarial wire input — malformed lengths, truncated bodies | 6 | No | +| [`header_tests.rs`](../tests/header_tests.rs) | Request/response header v1/v2, version lookup table | 10 | No | +| [`api_handler_tests.rs`](../tests/api_handler_tests.rs) | ApiVersions, Metadata stub, unsupported key/version | 7 | No | +| [`golden_wire_fixtures_tests.rs`](../tests/golden_wire_fixtures_tests.rs) | Byte-exact golden responses (ApiVersions v1, Metadata v0) | 2 | No | +| [`decode_validation_tests.rs`](../tests/decode_validation_tests.rs) | kafka-tool fixture decode + response structure per version | 14 | **Yes** | +| [`version_firewall_tests.rs`](../tests/version_firewall_tests.rs) | Version boundary matrix, unsupported keys, corrupt bodies | 17 | Partial | +| [`metadata_regression_tests.rs`](../tests/metadata_regression_tests.rs) | Metadata v0–v9, topic counts, broker advertise | 7 | No | Review Comment: SCOPE.md seems worth keeping — it records decisions the code can't show. TEST_SUITE.md I'd trim now rather than after merge: the catalog and coverage matrix just duplicate the test files, and they already drifted in this PR. I'd keep only the fixture policy and how to run the suites. ########## gateways/kafka/src/server.rs: ########## @@ -0,0 +1,617 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use bytes::{BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::broadcast; +use tokio::time::{timeout, timeout_at}; +use tokio_util::task::TaskTracker; +use tracing::{debug, error, info, warn}; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::api::{ + BrokerAdvertise, DEFAULT_KAFKA_PORT, ERROR_INVALID_REQUEST, HandleOutcome, + encode_error_only_response, handle_request, +}; +use crate::protocol::codec::Decoder; +use crate::protocol::header::{ + RequestHeader, ResponseHeader, request_header_version, response_header_version, +}; +use std::io; + +const READ_CHUNK: usize = 65536; + +#[derive(Debug, Clone)] +pub struct ServerConfig { + pub bind_addr: String, + /// Hostname or IP advertised in Metadata (`KAFKA_ADVERTISED_HOST`). Required when `bind_addr` + /// uses a wildcard address (`0.0.0.0` / `::`). + pub advertised_host: Option<String>, + /// Port advertised in Metadata (`KAFKA_ADVERTISED_PORT`). Defaults to the bind port. + pub advertised_port: Option<u16>, + pub max_frame_size: usize, + pub read_timeout: Duration, + pub write_timeout: Duration, +} + +impl Default for ServerConfig { + fn default() -> Self { + Self { + bind_addr: format!("127.0.0.1:{DEFAULT_KAFKA_PORT}"), + advertised_host: None, + advertised_port: None, + max_frame_size: 8 * 1024 * 1024, + read_timeout: Duration::from_secs(15), + write_timeout: Duration::from_secs(10), + } + } +} + +impl BrokerAdvertise { + /// Resolve the broker endpoint advertised in Metadata. + /// + /// `local_addr` is the address the listener is actually bound to (from `listener.local_addr()`). + /// + /// # Errors + /// + /// Returns `InvalidConfig` when `advertised_host` is empty or the listener binds to a wildcard + /// without an explicit advertised host. + pub fn from_server_config(config: &ServerConfig, local_addr: SocketAddr) -> Result<Self> { + let port = config + .advertised_port + .map_or_else(|| i32::from(local_addr.port()), i32::from); + + let host = if let Some(ref advertised) = config.advertised_host { + let trimmed = advertised.trim(); + if trimmed.is_empty() { + return Err(KafkaProtocolError::InvalidConfig( + "KAFKA_ADVERTISED_HOST must not be empty".into(), + )); + } + if trimmed.len() > i16::MAX as usize { + return Err(KafkaProtocolError::InvalidConfig( + "KAFKA_ADVERTISED_HOST exceeds Kafka nullable string limit (32767 bytes)" + .into(), + )); + } + trimmed.to_string() + } else if local_addr.ip().is_unspecified() { + return Err(KafkaProtocolError::InvalidConfig( + "binding to a wildcard address (0.0.0.0 or ::) requires KAFKA_ADVERTISED_HOST \ + to be set to a reachable hostname or IP for Metadata broker advertisement" + .into(), + )); + } else { + local_addr.ip().to_string() + }; + + Ok(Self { host, port }) + } +} + +pub struct KafkaServer { + config: Arc<ServerConfig>, +} + +impl KafkaServer { + #[must_use] + pub fn new(config: ServerConfig) -> Self { + Self { + config: Arc::new(config), + } + } + + /// Accept Kafka wire connections until `shutdown` fires, then drain in-flight tasks. + /// + /// `listener` must already be bound by the caller. This lets tests and `main` bind + /// the port before spawning the task, eliminating the TOCTOU race of bind-drop-rebind. + /// + /// # Errors + /// + /// Returns an error on invalid config or a non-transient `accept()` error. + pub async fn run( + self, + listener: TcpListener, + mut shutdown: broadcast::Receiver<()>, + ) -> Result<()> { + let local_addr = listener.local_addr()?; + let broker = Arc::new(BrokerAdvertise::from_server_config( + &self.config, + local_addr, + )?); + info!( + "kafka listener bound on {} (advertised as {}:{})", + local_addr, broker.host, broker.port + ); + + let tracker = TaskTracker::new(); + let broker = Arc::clone(&broker); + + loop { + tokio::select! { + result = shutdown.recv() => { + match result { + Ok(()) => { + info!("kafka listener shutdown requested"); + tracker.close(); + tracker.wait().await; + break; + } + // Capacity-1 channel: lagged means a signal was sent before we polled - treat as shutdown. + Err(broadcast::error::RecvError::Lagged(_)) => { + info!("kafka listener shutdown requested (lagged)"); + tracker.close(); + tracker.wait().await; + break; + } + Err(broadcast::error::RecvError::Closed) => { + tracker.close(); + tracker.wait().await; + break; + } + } + } + accept_result = listener.accept() => { + match accept_result { + Ok((stream, peer)) => { + if let Err(e) = stream.set_nodelay(true) { + warn!(%peer, "TCP_NODELAY failed: {e}"); + } + if let Err(e) = enable_tcp_keepalive(&stream) { + warn!(%peer, "TCP_KEEPALIVE failed: {e}"); + } + let cfg = Arc::clone(&self.config); + let broker = Arc::clone(&broker); + tracker.spawn(async move { + if let Err(err) = handle_connection(stream, cfg, peer, broker).await { + warn!(%peer, "connection closed with error: {err}"); + } + }); + } + Err(e) if is_transient_accept_error(&e) => { + // Brief backoff on fd exhaustion to avoid busy-spinning. + if matches!(e.raw_os_error(), Some(23 | 24)) { + tokio::time::sleep(Duration::from_millis(10)).await; + } + warn!(%e, "transient accept error, continuing"); + } + Err(e) => return Err(e.into()), + } + } + + } + } + Ok(()) + } +} + +fn is_transient_accept_error(err: &std::io::Error) -> bool { + use std::io::ErrorKind; Review Comment: `use std::io::ErrorKind;` should move to the module-level ########## gateways/kafka/tests/server_e2e_tests.rs: ########## @@ -0,0 +1,480 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! End-to-end TCP tests through `KafkaServer` (full request/response cycle). + +#[path = "common/fixtures.rs"] +mod fixtures; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use bytes::{BufMut, Bytes, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +use iggy_gateway_kafka::protocol::api::{ + API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_LIST_OFFSETS, + API_KEY_METADATA, API_KEY_PRODUCE, ERROR_NONE, ERROR_UNSUPPORTED_VERSION, +}; +use iggy_gateway_kafka::protocol::codec::Decoder; + +use fixtures::load_fixture_body_or_skip; +use server::spawn_test_server; +use std::time::Duration; +use tcp::{ + ByteRead, build_list_offsets_v0_request_with_topic_t, build_metadata_legacy_request, + build_produce_v3_body, build_request_frame, parse_response_payload, read_byte_with_timeout, + read_response_frame, read_response_frame_with_timeout, round_trip, +}; +use wire::{ + OUT_OF_SCOPE_API_KEYS, build_create_topics_empty_request, build_fetch_empty_topics_request, + build_list_offsets_request, build_produce_flexible_empty_request, +}; + +#[tokio::test] +async fn e2e_apiversions_v1_preserves_correlation_id() { + let (addr, _shutdown) = spawn_test_server().await; + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 1, 42_001, &[]).await; + assert_eq!(corr, 42_001); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); +} + +#[tokio::test] +async fn e2e_apiversions_v3_flexible_preserves_correlation_id() { + let (addr, _shutdown) = spawn_test_server().await; + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 3, 42_002, &[]).await; + assert_eq!(corr, 42_002); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_varint().unwrap() - 1).expect("api count fits usize"); + assert_eq!(count, 6); +} + +#[tokio::test] +async fn e2e_metadata_v0_returns_stub_broker() { + let (addr, _shutdown) = spawn_test_server().await; + let mut req = BytesMut::new(); + req.put_i32(0); // empty topics + let (corr, body) = round_trip(addr, API_KEY_METADATA, 0, 77, &req).await; + assert_eq!(corr, 77); + let mut d = Decoder::new(body); + assert_eq!(d.read_i32().unwrap(), 1); + d.read_i32().unwrap(); + let host = d.read_nullable_string().unwrap().unwrap(); + assert_eq!(host, "127.0.0.1"); +} + +#[tokio::test] +async fn e2e_produce_v3_round_trip_with_fixture() { + let (addr, _shutdown) = spawn_test_server().await; + let Some(body) = load_fixture_body_or_skip(0, "Produce", 3) else { + return; + }; + let (corr, resp_body) = round_trip(addr, API_KEY_PRODUCE, 3, 88, &body).await; + assert_eq!(corr, 88); + assert!(!resp_body.is_empty()); +} + +#[tokio::test] +async fn e2e_unsupported_api_key_returns_error_then_closes() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]); + stream.write_all(&frame1).await.unwrap(); + let payload1 = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, body) = parse_response_payload(8, 2, payload1); + assert_eq!(corr, 99); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION); + + // The unsupported-version error is terminal: the server closes the connection. + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "connection must close after the unsupported-version error response" + ); +} + +#[tokio::test] +async fn e2e_sequential_requests_on_one_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let requests = [(API_KEY_API_VERSIONS, 1i16), (API_KEY_METADATA, 0i16)]; + for (i, (key, ver)) in requests.iter().enumerate() { + let meta_body = { + let mut b = BytesMut::new(); + b.put_i32(0); + b + }; + let body: &[u8] = if *key == API_KEY_METADATA { + &meta_body + } else { + &[] + }; + let correlation_id = 1000 + i32::try_from(i).expect("test index fits i32"); + let frame = build_request_frame(*key, *ver, correlation_id, Some("seq-test"), body); + stream.write_all(&frame).await.unwrap(); + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, _) = parse_response_payload(*key, *ver, payload); + assert_eq!(corr, correlation_id); + } +} + +#[tokio::test] +async fn e2e_negative_frame_length_closes_connection() { Review Comment: `e2e_negative_frame_length_closes_connection` exists in both this file and `listener_robustness_tests.rs:240` -- same name, same scenario (negative length prefix must close the connection), only the literal (-1 vs -5) and the read helper differ. The listener_robustness copy uses the timeout-guarded `read_byte_with_timeout`, so I'd keep that one and drop this copy. Related near-overlaps worth a look while at it: `e2e_oversized_frame_is_rejected` vs `e2e_frame_exceeding_max_frame_size_closes_connection` (default cap vs custom `max_frame_size` -- arguably both earn their keep), and `e2e_sequential_requests_on_one_connection` vs `e2e_many_sequential_requests_on_one_connection` + `e2e_mixed_api_key_pipeline_returns_responses_in_order`, which mostly subsume it. ########## gateways/kafka/tools/kafka-tool/Cargo.toml: ########## @@ -0,0 +1,42 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "kafka-message-gen" +version = "0.1.0" +edition = "2024" +description = "Generates binary Kafka protocol messages for testing the Iggy Kafka gateway" +license = "Apache-2.0" +repository = "https://github.com/apache/iggy"; +keywords = ["kafka", "protocol", "testing", "iggy", "wire-format"] +publish = false + +[[bin]] +name = "kafka-message-gen" +path = "src/main.rs" + +[dependencies] +anyhow = { workspace = true } +bytes = { workspace = true } +clap = { workspace = true } +hex = "0.4" +iggy-gateway-kafka = { path = "../.." } +indexmap = "2" Review Comment: `indexmap = "2"` this should be `indexmap = { workspace = true }` like the other deps in this file. ########## gateways/kafka/Cargo.toml: ########## @@ -0,0 +1,59 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "iggy-gateway-kafka" +version = "0.1.0" +description = "Kafka wire protocol gateway foundation for Apache Iggy" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "kafka", "gateway", "streaming"] +homepage = "https://iggy.apache.org"; +documentation = "https://iggy.apache.org/docs"; +repository = "https://github.com/apache/iggy"; +readme = "README.md" +publish = false + +[[bin]] +name = "iggy-gateway-kafka" +path = "src/main.rs" + +[dependencies] +bytes = { workspace = true } +socket2 = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = [ + "rt-multi-thread", + "macros", + "net", + "io-util", + "time", + "sync", + "signal", +] } +tokio-util = { workspace = true, features = ["rt"] } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "time"] } + +[lints.clippy] +enum_glob_use = "deny" +#Ported Kafka wire codec; pedantic cleanup tracked for a follow-up PR. Review Comment: This comment is stale -- the blanket `#![allow(clippy::pedantic)]` directives are gone and the cleanup happened in this PR. (Also missing a space after `#`.) ########## gateways/kafka/docs/MANUAL_TESTING.md: ########## @@ -0,0 +1,275 @@ +# Kafka gateway — manual testing procedure + +Manual validation for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421) foundation: TCP listener, wire decode, version firewall, stub responses. **No Iggy backend** — success means correct Kafka wire behavior, not message persistence. + +See also: [SCOPE.md](SCOPE.md) (supported API keys), [TEST_SUITE.md](TEST_SUITE.md) (automated coverage). + +--- + +## 1. Environment setup + +### Requirements + +| Tool | Purpose | Install | +| ------ | --------- | --------- | +| Rust toolchain | Build gateway + kafka-tool | [rustup.rs](https://rustup.rs) | +| `kafka-message-gen` | Generate/send wire fixtures | `cargo build -p kafka-message-gen` | +| `kcat` (optional) | Real Kafka client smoke test | `brew install kcat` / `apt install kafkacat` | +| `nc` / `netcat` (optional) | Raw byte injection | Usually preinstalled | +| `xxd` or `hexdump` (optional) | Inspect binary responses | Usually preinstalled | + +### Build and start gateway + +```bash +# From iggy workspace root +cargo build -p iggy-gateway-kafka + +# Terminal 1 — start listener (default 127.0.0.1:9093) +RUST_LOG=info cargo run -p iggy-gateway-kafka +``` + +Expected log: + +```text +kafka listener bound on 127.0.0.1:9093 +``` + +### Generate wire fixtures + +```bash +# Terminal 2 +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 3 --api-key 18 --api-key 19 +``` + +--- + +## 2. Pre-flight automated check + +Run before manual testing to catch regressions: + +```bash +cargo test -p iggy-gateway-kafka +``` + +All tests must pass. If `decode_validation_tests` fail, regenerate fixtures (step above). + +--- + +## 3. Manual test cases + +### Category A — Smoke tests (must pass before check-in) + +| ID | Test | Steps | Expected result | Pass criteria | +| ---- | ------ | ------- | ----------------- | --------------- | +| A1 | Gateway starts | Run `iggy-gateway-kafka` | Binds to `:9093`, no panic | Log shows bind address | +| A2 | ApiVersions v1 | `cargo run -p kafka-message-gen -- send --host 127.0.0.1:9093 --api-key 18 --version 1` | Response received | `ec=0`, non-zero byte count | +| A3 | ApiVersions v3 (flexible) | Same with `--version 3` | Response received | `ec=0` | +| A4 | Metadata v0 | `send --api-key 3 --version 0` | Stub broker in response | `ec=0` or topic error 3 (stub) | +| A5 | Produce v3 | `send --api-key 0 --version 3` | Decode + stub ack | `ec=0` | +| A6 | Fetch v4 | `send --api-key 1 --version 4` | Decode + stub response | `ec=0` | +| A7 | ListOffsets v1 | `send --api-key 2 --version 1` | Decode + stub offsets | `ec=0` | +| A8 | CreateTopics v2 | `send --api-key 19 --version 2` | Decode + stub ack | `ec=0` | +| A9 | Verify all scoped keys | `cargo run -p kafka-message-gen -- verify --host 127.0.0.1:9093 --api-key 0 --api-key 1 --api-key 2 --api-key 3 --api-key 18 --api-key 19` | Exit code 0 | No timeouts or I/O errors | + +### Category B — Version firewall (boundary validation) + +For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-message-gen send` with `--version N`. + +| API key | Name | Min | Max | Test versions | +| --------- | ------ | ----- | ----- | --------------- | +| 18 | ApiVersions | 0 | 3 | −1, 0, 3, 4 | +| 3 | Metadata | 0 | 9 | −1, 0, 9, 10 | +| 0 | Produce | 3 | 9 | 2, 3, 9, 10 | +| 1 | Fetch | 4 | 12 | 3, 4, 12, 13 | +| 2 | ListOffsets | 1 | 6 | 0, 1, 6, 7 | +| 19 | CreateTopics | 2 | 5 | 1, 2, 5, 6 | + +| ID | Test | Expected for in-range | Expected for out-of-range | +| ---- | ------ | ---------------------- | --------------------------- | +| B1 | ApiVersions negotiation | `error_code=0`; body lists 6 API keys with correct min/max | `error_code=35` (UNSUPPORTED_VERSION) | +| B2 | Metadata out-of-range | N/A | Topic entries show `error_code=35` | +| B3 | Produce/Fetch/ListOffsets/CreateTopics out-of-range | N/A | Version-aware response with `error_code=35` (top-level or per-topic/partition) | +| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,18,19 only — no consumer-group keys | + +**Validation tip:** Use `--hex` when generating to inspect request bytes: + +```bash +cargo run -p kafka-message-gen -- generate --api-key 18 --version 3 --hex +``` + +### Category C — Unsupported API keys + +| ID | API key | Name | Steps | Expected | +| ---- | --------- | ------ | ------- | ---------- | +| C1 | 8 | OffsetCommit | `send --api-key 8 --version 2` | `ec=35`, connection stays open | +| C2 | 10 | FindCoordinator | `send --api-key 10` | `ec=35` | +| C3 | 17 | SaslHandshake | `send --api-key 17` | `ec=35` | +| C4 | 20 | DeleteTopics | `send --api-key 20` | `ec=35` | + +Follow C1 with A2 on the **same** `nc` session to confirm the connection is not dropped. + +### Category D — Flexible vs legacy wire encoding + +| ID | API key | Version | Encoding | Validation | +| ---- | --------- | --------- | ---------- | ------------ | +| D1 | Produce | 8 | Legacy (i32 arrays) | `send` succeeds, `ec=0` | +| D2 | Produce | 9 | Flexible (compact + tagged fields) | `send` succeeds, `ec=0` | +| D3 | Fetch | 11 | Legacy | `send` succeeds | +| D4 | Fetch | 12 | Flexible | `send` succeeds | +| D5 | Metadata | 8 | Legacy | `send` succeeds | +| D6 | Metadata | 9 | Flexible | `send` succeeds | +| D7 | ListOffsets | 5 | Legacy | `send` succeeds | +| D8 | ListOffsets | 6 | Flexible | `send` succeeds | +| D9 | CreateTopics | 4 | Legacy | `send` succeeds | +| D10 | CreateTopics | 5 | Flexible | `send` succeeds | + +### Category E — Metadata stub semantics + +| ID | Test | Steps | Expected | +| ---- | ------ | ------- | ---------- | +| E1 | Broker advertise address | Start gateway on `127.0.0.1:9093`; Metadata v0 | Broker host=`127.0.0.1`, port=`9093` | +| E2 | Wildcard bind + advertised host | `KAFKA_BIND_ADDR=0.0.0.0:19093` + `KAFKA_ADVERTISED_HOST=kafka.internal`, restart | Metadata broker host/port match advertised values | +| E3 | Unknown topic stub | Metadata with topic name `my-topic` | Topic error `3` (UNKNOWN_TOPIC_OR_PARTITION), name `unknown-topic` | +| E4 | Multiple topics | Metadata request listing 3 topics | 3 topic entries, each with error 3 | + +### Category F — TCP / connection behavior + +| ID | Test | Steps | Expected | +| ---- | ------ | ------- | ---------- | +| F1 | Correlation ID echoed | Send ApiVersions with known correlation_id; decode response header | Response correlation_id matches request | +| F2 | Sequential requests | Send ApiVersions then Metadata on same TCP connection | Both get valid responses | +| F3 | Client disconnect | Connect, send partial frame, close | Gateway logs clean disconnect, no panic | +| F4 | Invalid frame length 0 | `printf '\x00\x00\x00\x00' \| nc 127.0.0.1 9093` | Connection closed, gateway continues serving others | +| F5 | Oversized frame | Send 4-byte length > 8 MiB | Connection rejected/closed, no OOM | +| F6 | Graceful shutdown | Ctrl+C on gateway | Log "shutdown requested", in-flight requests drain | + +### Category G — Real Kafka client (kcat) + +Requires `kcat` installed. Gateway does **not** implement SASL or full broker semantics — expect limited success. + +| ID | Test | Command | Expected (foundation) | +| ---- | ------ | --------- | --------------------- | +| G1 | Broker metadata | `kcat -b 127.0.0.1:9093 -L` | ApiVersions + Metadata handshake; broker appears in metadata | +| G2 | Produce (likely fails later) | `echo "hello" \| kcat -b 127.0.0.1:9093 -t test -P` | May fail at coordinator/group stage — document actual error | +| G3 | Consumer (likely fails later) | `kcat -b 127.0.0.1:9093 -t test -C -o beginning` | May fail without consumer groups — document actual error | + +Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke. + +### Category H — Adversarial / negative input + +| ID | Test | Steps | Expected | +| ---- | ------ | ------- | ---------- | +| H1 | Truncated Produce body | Send valid header + incomplete body | `error_code=42` (INVALID_REQUEST) or connection error; **no panic** | +| H2 | Random bytes | `dd if=/dev/urandom bs=64 count=1 \| nc 127.0.0.1 9093` | Connection closed or protocol error; gateway stays up | +| H3 | Empty body after header | ApiVersions with valid header, empty body | `ec=0` (ApiVersions accepts empty body) | + +--- + +## 4. Validation reference + +### Kafka error codes used in #3421 + +| Code | Name | When returned | +| ------ | ------ | --------------- | +| 0 | NONE | Successful stub response (Fetch/ListOffsets/ApiVersions) | +| 6 | NOT_LEADER_OR_FOLLOWER | Produce stub (retriable; payload not persisted) | +| 3 | UNKNOWN_TOPIC_OR_PARTITION | Metadata stub per-topic error | +| 35 | UNSUPPORTED_VERSION | Out-of-range version or unlisted API key | +| 37 | INVALID_PARTITIONS | CreateTopics: partition count `0` or `< -1` (or any non-positive on v2–v3) | +| 38 | INVALID_REPLICATION_FACTOR | CreateTopics: replication factor `0` or `< -1` (or any non-positive on v2–v3) | +| 41 | NOT_CONTROLLER | CreateTopics stub (topic not created) | +| 42 | INVALID_REQUEST | Produce/Fetch/ListOffsets/CreateTopics decode failure; unsupported request header | + +### Response header rules + +| API key | Request flexible? | Response header version | +| --------- | -------------------- | ------------------------- | +| 18 ApiVersions | v3+ | Always v0 (correlation_id only) | +| 3 Metadata | v9+ | v1 (correlation_id + tagged fields) | +| 0 Produce | v9+ | v1 | +| 1 Fetch | v12+ | v1 | +| Others | Per SCOPE.md | See `header.rs` lookup table | + +### Frame layout (for manual hex inspection) + +```text +Request frame: + [length: i32 BE] + [api_key: i16][api_version: i16][correlation_id: i32] + [client_id: NULLABLE_STRING or COMPACT_NULLABLE_STRING] + [tagged_fields: 0x00] ← flexible requests only + [request body] + +Response frame: + [length: i32 BE] + [correlation_id: i32] + [tagged_fields: 0x00] ← flexible responses only (not ApiVersions) + [response body] +``` + +### Raw netcat smoke test + +```bash +# ApiVersions v3 — after generating fixtures +cat gateways/kafka/tools/kafka-tool/kafka_messages/018_ApiVersions_v3.bin \ + | nc -w 2 127.0.0.1 9093 | xxd | head -20 +``` + +First bytes after length prefix should include your correlation_id from the fixture. + +--- + +## 5. Manual test execution checklist + +Copy this checklist into your PR or test log: + +```text +Date: ___________ +Tester: ___________ +Gateway commit: ___________ +kcat version (if used): ___________ + +[ ] A1–A9 Smoke tests +[ ] B1–B4 Version firewall (all 6 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 + +Automated regression: +[ ] cargo test -p iggy-gateway-kafka — ___/103 passed +[ ] cargo clippy -p iggy-gateway-kafka — clean / warnings noted + +Notes / failures: +_________________________________ +``` + +--- + +## 6. Troubleshooting + +| Symptom | Likely cause | Fix | +| --------- | -------------- | ----- | +| `Connection refused` on 9093 | Gateway not running | Start `iggy-gateway-kafka` | +| `decode_validation_tests` panic | Missing fixtures | Run `kafka-message-gen generate` | Review Comment: This troubleshooting row says `decode_validation_tests` panics on missing fixtures, but after the refactor those tests skip with a note (and only panic when `KAFKA_FIXTURES_REQUIRED` is set). Worth updating so an operator doesn't wait for a panic that never comes. ########## gateways/kafka/src/server.rs: ########## @@ -0,0 +1,617 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use bytes::{BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::broadcast; +use tokio::time::{timeout, timeout_at}; +use tokio_util::task::TaskTracker; +use tracing::{debug, error, info, warn}; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::api::{ + BrokerAdvertise, DEFAULT_KAFKA_PORT, ERROR_INVALID_REQUEST, HandleOutcome, + encode_error_only_response, handle_request, +}; +use crate::protocol::codec::Decoder; +use crate::protocol::header::{ + RequestHeader, ResponseHeader, request_header_version, response_header_version, +}; +use std::io; + +const READ_CHUNK: usize = 65536; + +#[derive(Debug, Clone)] +pub struct ServerConfig { + pub bind_addr: String, + /// Hostname or IP advertised in Metadata (`KAFKA_ADVERTISED_HOST`). Required when `bind_addr` + /// uses a wildcard address (`0.0.0.0` / `::`). + pub advertised_host: Option<String>, + /// Port advertised in Metadata (`KAFKA_ADVERTISED_PORT`). Defaults to the bind port. + pub advertised_port: Option<u16>, + pub max_frame_size: usize, + pub read_timeout: Duration, + pub write_timeout: Duration, +} + +impl Default for ServerConfig { + fn default() -> Self { + Self { + bind_addr: format!("127.0.0.1:{DEFAULT_KAFKA_PORT}"), + advertised_host: None, + advertised_port: None, + max_frame_size: 8 * 1024 * 1024, + read_timeout: Duration::from_secs(15), + write_timeout: Duration::from_secs(10), + } + } +} + +impl BrokerAdvertise { + /// Resolve the broker endpoint advertised in Metadata. + /// + /// `local_addr` is the address the listener is actually bound to (from `listener.local_addr()`). + /// + /// # Errors + /// + /// Returns `InvalidConfig` when `advertised_host` is empty or the listener binds to a wildcard + /// without an explicit advertised host. + pub fn from_server_config(config: &ServerConfig, local_addr: SocketAddr) -> Result<Self> { + let port = config + .advertised_port + .map_or_else(|| i32::from(local_addr.port()), i32::from); + + let host = if let Some(ref advertised) = config.advertised_host { + let trimmed = advertised.trim(); + if trimmed.is_empty() { + return Err(KafkaProtocolError::InvalidConfig( + "KAFKA_ADVERTISED_HOST must not be empty".into(), + )); + } + if trimmed.len() > i16::MAX as usize { + return Err(KafkaProtocolError::InvalidConfig( + "KAFKA_ADVERTISED_HOST exceeds Kafka nullable string limit (32767 bytes)" + .into(), + )); + } + trimmed.to_string() + } else if local_addr.ip().is_unspecified() { + return Err(KafkaProtocolError::InvalidConfig( + "binding to a wildcard address (0.0.0.0 or ::) requires KAFKA_ADVERTISED_HOST \ + to be set to a reachable hostname or IP for Metadata broker advertisement" + .into(), + )); + } else { + local_addr.ip().to_string() + }; + + Ok(Self { host, port }) + } +} + +pub struct KafkaServer { + config: Arc<ServerConfig>, +} + +impl KafkaServer { + #[must_use] + pub fn new(config: ServerConfig) -> Self { + Self { + config: Arc::new(config), + } + } + + /// Accept Kafka wire connections until `shutdown` fires, then drain in-flight tasks. + /// + /// `listener` must already be bound by the caller. This lets tests and `main` bind + /// the port before spawning the task, eliminating the TOCTOU race of bind-drop-rebind. + /// + /// # Errors + /// + /// Returns an error on invalid config or a non-transient `accept()` error. + pub async fn run( + self, + listener: TcpListener, + mut shutdown: broadcast::Receiver<()>, + ) -> Result<()> { + let local_addr = listener.local_addr()?; + let broker = Arc::new(BrokerAdvertise::from_server_config( + &self.config, + local_addr, + )?); + info!( + "kafka listener bound on {} (advertised as {}:{})", + local_addr, broker.host, broker.port + ); + + let tracker = TaskTracker::new(); + let broker = Arc::clone(&broker); Review Comment: This line can be dropped, the accept loop already clones per connection on line 183 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
