krishvishal commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3803283962
########## gateways/kafka/tests/version_firewall_tests.rs: ########## @@ -0,0 +1,742 @@ +// 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. + +//! Version negotiation firewall - boundary tests for every scoped API key. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/fixtures.rs"] +mod fixtures; +#[path = "common/scope.rs"] +mod scope; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::time::Duration; + +use bytes::Bytes; +use tokio::io::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_INVALID_REQUEST, ERROR_NONE, + ERROR_UNSUPPORTED_VERSION, advertised_min_version, handle_request, is_supported_version, + supported_api_ranges, +}; + +use codec::Decoder; +use fixtures::{fixture_exists, load_fixture_body, load_fixture_body_or_skip}; +use scope::{SCOPED_API_KEYS, default_broker}; +use server::spawn_test_server; +use tcp::{ + ByteRead, build_list_offsets_v0_request_with_topic_t, build_metadata_legacy_request, + build_produce_flexible_body, build_produce_v2_body, build_produce_v3_body, build_request_frame, + parse_response_payload, read_byte_with_timeout, round_trip, scan_for_error_code, +}; +use wire::{ + OUT_OF_SCOPE_API_KEYS, build_api_versions_flexible_request, build_create_topics_empty_request, + build_fetch_empty_topics_request, build_list_offsets_request, + build_metadata_all_topics_flexible, build_metadata_all_topics_legacy, + build_metadata_flexible_request_v10, +}; + +#[test] +fn supported_ranges_table_has_six_entries() { + assert_eq!(supported_api_ranges().len(), 6); +} + +#[test] +fn is_supported_version_matches_scope_table() { + for &(api_key, _, min_ver, max_ver) in SCOPED_API_KEYS { + assert!( + !is_supported_version(api_key, min_ver - 1), + "key {api_key} must reject v{}", + min_ver - 1 + ); + assert!( + is_supported_version(api_key, min_ver), + "key {api_key} must accept min v{min_ver}" + ); + assert!( + is_supported_version(api_key, max_ver), + "key {api_key} must accept max v{max_ver}" + ); + assert!( + !is_supported_version(api_key, max_ver + 1), + "key {api_key} must reject v{}", + max_ver + 1 + ); + } +} + +#[test] +fn apiversions_advertises_exact_supported_ranges_v1() { + let body = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_i32().unwrap()).expect("api count fits usize"); + assert_eq!(count, supported_api_ranges().len()); + + for expected in supported_api_ranges() { + let key = d.read_i16().unwrap(); + let min = d.read_i16().unwrap(); + let max = d.read_i16().unwrap(); + assert_eq!(key, expected.api_key); + assert_eq!( Review Comment: **Verifies shape, not values.** `min` was read out of a response that `encode_api_versions_response` (`src/protocol/api.rs:440`) built by calling `advertised_min_version` on that same table row. The assertion compares the function against itself. Concrete broken implementation that passes both this test and `apiversions_advertises_exact_supported_ranges_v3_flexible` (`:116-144`): ```rust pub const fn advertised_min_version(api_key: i16, firewall_min: i16) -> i16 { if api_key == API_KEY_PRODUCE || api_key == API_KEY_FETCH { 0 } else { firewall_min } } ``` The response then advertises Fetch min=0, the test compares 0 == 0, green. Real clients send Fetch v0 and get a naked close, because `is_supported_version` still enforces 4. The actual value guard in this PR is the single byte-exact golden at `tests/golden_wire_fixtures_tests.rs:43-54`, where `00 01 00 04 00 0C` pins Fetch 4-12. That test is doing all the work here, and there is **no flexible/v3 golden** - so the v3 path's advertised values rest entirely on the tautology. Fix: rename these two to say what they check (`apiversions_v1_response_shape_matches_range_table`), and add a byte-exact `golden_apiversions_v3_flexible_response_fixture` alongside the v1 one. Tradeoff: a second golden doubles the churn when `SUPPORTED_RANGES` legitimately widens - acceptable, since widening the advertised set is a client-visible change that should require touching a golden. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,558 @@ +// 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 bytes::{Buf, Bytes}; +use kafka_protocol::messages::api_versions_response::ApiVersion; +use kafka_protocol::messages::metadata_response::{MetadataResponseBroker, MetadataResponseTopic}; +use kafka_protocol::messages::{ + ApiVersionsRequest, ApiVersionsResponse, BrokerId, CreateTopicsRequest, FetchRequest, + ListOffsetsRequest, MetadataRequest, MetadataResponse, ProduceRequest, TopicName, +}; +use kafka_protocol::protocol::{Decodable, StrBytes}; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::responses::{ + encode_create_topics_error_response, encode_create_topics_response, + encode_fetch_error_response, encode_fetch_response, encode_list_offsets_error_response, + encode_list_offsets_response, encode_message, encode_produce_error_response, + encode_produce_response, +}; + +pub const API_KEY_PRODUCE: i16 = 0; +pub const API_KEY_FETCH: i16 = 1; +pub const API_KEY_LIST_OFFSETS: i16 = 2; +pub const API_KEY_METADATA: i16 = 3; +pub const API_KEY_API_VERSIONS: i16 = 18; +pub const API_KEY_CREATE_TOPICS: i16 = 19; + +pub const DEFAULT_KAFKA_PORT: u16 = 9093; + +pub const ERROR_NONE: i16 = 0; +pub const ERROR_UNKNOWN_TOPIC_OR_PARTITION: i16 = 3; +/// Retriable; Produce stub uses this until the Iggy bridge persists records. +pub const ERROR_NOT_LEADER_OR_FOLLOWER: i16 = 6; +pub const ERROR_UNSUPPORTED_VERSION: i16 = 35; +pub const ERROR_INVALID_PARTITIONS: i16 = 37; +pub const ERROR_INVALID_REPLICATION_FACTOR: i16 = 38; +/// `CreateTopics` stub: do not claim topics were created (no controller / no Iggy bridge). +pub const ERROR_NOT_CONTROLLER: i16 = 41; +pub const ERROR_INVALID_REQUEST: i16 = 42; + +/// Result of handling one Kafka request body. +#[derive(Debug)] +pub enum HandleOutcome { + /// Write this response body (with a response header). + Respond(Bytes), + /// Produce with `acks=0`: write nothing, keep the connection open. + NoResponse, + /// No parseable response exists for this request; close the TCP connection. + Close, +} + +impl HandleOutcome { + /// Return the response body, or panic with `msg` if the outcome is not [`Self::Respond`]. + /// + /// # Panics + /// + /// Panics when the outcome is [`Self::NoResponse`] or [`Self::Close`]. + #[must_use] + pub fn expect_response(self, msg: &str) -> Bytes { + match self { + Self::Respond(body) => body, + Self::NoResponse => panic!("{msg}: got NoResponse"), + Self::Close => panic!("{msg}: got Close"), + } + } + + #[must_use] + pub const fn is_no_response(&self) -> bool { + matches!(self, Self::NoResponse) + } + + #[must_use] + pub const fn is_close(&self) -> bool { + matches!(self, Self::Close) + } +} + +#[derive(Debug, Clone)] +pub struct BrokerAdvertise { + pub host: String, + pub port: i32, +} + +impl Default for BrokerAdvertise { + fn default() -> Self { + Self { + host: "127.0.0.1".to_string(), + port: i32::from(DEFAULT_KAFKA_PORT), + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct ApiVersionRange { + pub api_key: i16, + pub min_version: i16, + pub max_version: i16, +} + +static SUPPORTED_RANGES: &[ApiVersionRange] = &[ + ApiVersionRange { + api_key: API_KEY_PRODUCE, + min_version: 3, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_FETCH, + min_version: 4, + max_version: 12, + }, + ApiVersionRange { + api_key: API_KEY_LIST_OFFSETS, + min_version: 1, + max_version: 6, + }, + ApiVersionRange { + api_key: API_KEY_METADATA, + min_version: 0, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_API_VERSIONS, + min_version: 0, + max_version: 3, + }, + ApiVersionRange { + api_key: API_KEY_CREATE_TOPICS, + min_version: 2, + max_version: 5, + }, +]; + +#[must_use] +pub fn supported_api_ranges() -> &'static [ApiVersionRange] { + SUPPORTED_RANGES +} + +/// Handles one decoded request frame and returns how the connection should proceed. +pub fn handle_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> HandleOutcome { + if api_key == API_KEY_PRODUCE { + return handle_produce_request(api_version, body); + } + handle_other_request(api_key, api_version, body, broker) +} + +/// Decode `T` from the whole request body and reject unconsumed trailing bytes. +/// +/// `kafka_protocol`'s `Decodable` stops once it has read the fields its schema defines; it does +/// not know (or care) whether the caller handed it an exact-length body, so the trailing-bytes +/// check has to live here. +fn decode_exhaustive<T: Decodable>(version: i16, mut body: Bytes) -> Result<T> { + let value = + T::decode(&mut body, version).map_err(|e| KafkaProtocolError::Malformed(e.to_string()))?; + if body.has_remaining() { + return Err(KafkaProtocolError::Malformed( + "unexpected trailing bytes in request body".to_string(), + )); + } + Ok(value) +} + +/// Turn an encode [`Result`] into a [`HandleOutcome`], closing the connection when encoding +/// fails rather than propagating - there is no parseable response to send in that case. +fn respond_or_close(result: Result<Bytes>, api_name: &str) -> HandleOutcome { + match result { + Ok(body) => HandleOutcome::Respond(body), + Err(error) => { + tracing::warn!(%error, "failed to encode {api_name} response; closing connection"); + HandleOutcome::Close + } + } +} + +/// Produce is the only request the wire protocol allows to go unanswered +/// (`acks=0`), so it gets its own path that may return [`HandleOutcome::NoResponse`]. +/// +/// The firewall check runs AFTER decoding the request, not before: `ApiVersions` advertises +/// Produce min=0 (see [`advertised_min_version`]) while the firewall's real floor is 3, so a +/// spec-compliant client can legitimately send Produce v0-2 with `acks=0`. Rejecting those +/// versions before reading `acks` would send an error response the client never expects, +/// desyncing the next correlation id it reads. +fn handle_produce_request(api_version: i16, body: Bytes) -> HandleOutcome { + // Above the encoder max there is no response parseable at the client's version, so close + // rather than decode (same policy the other APIs apply). + if api_version > supported_max_version(API_KEY_PRODUCE).unwrap_or(i16::MAX) { Review Comment: **Opposite fail-safe directions for the same table lookup.** This is `supported_max_version(API_KEY_PRODUCE).unwrap_or(i16::MAX)` (fail-open). `api.rs:391` is `.map_or(0, |r| r.max_version)` (fail-closed). Same table, same missing-row case, opposite defaults. Today both rows exist so neither default fires, which is what makes the divergence invisible. `handle_request:159-161` dispatches Produce on a hard-coded `api_key ==`, not from the table. So deleting the Produce row to disable the API leaves Produce v0-2 with `acks=0` silently accepted and discarded on an API the operator believes is off, while the other three APIs correctly close. Latent until someone edits `SUPPORTED_RANGES`, which `docs/SCOPE.md:27` explicitly invites ("Expand `SUPPORTED_RANGES` only after ... manually tested"). Fix: `unwrap_or(i16::MIN)` here, or gate the Produce dispatch on the row existing so a missing row means "unknown api key -> Close" like every other unlisted key. The value is dead today, so this changes only the behaviour of a future edit. ########## gateways/kafka/tests/version_firewall_tests.rs: ########## @@ -0,0 +1,742 @@ +// 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. + +//! Version negotiation firewall - boundary tests for every scoped API key. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/fixtures.rs"] +mod fixtures; +#[path = "common/scope.rs"] +mod scope; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::time::Duration; + +use bytes::Bytes; +use tokio::io::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_INVALID_REQUEST, ERROR_NONE, + ERROR_UNSUPPORTED_VERSION, advertised_min_version, handle_request, is_supported_version, + supported_api_ranges, +}; + +use codec::Decoder; +use fixtures::{fixture_exists, load_fixture_body, load_fixture_body_or_skip}; +use scope::{SCOPED_API_KEYS, default_broker}; +use server::spawn_test_server; +use tcp::{ + ByteRead, build_list_offsets_v0_request_with_topic_t, build_metadata_legacy_request, + build_produce_flexible_body, build_produce_v2_body, build_produce_v3_body, build_request_frame, + parse_response_payload, read_byte_with_timeout, round_trip, scan_for_error_code, +}; +use wire::{ + OUT_OF_SCOPE_API_KEYS, build_api_versions_flexible_request, build_create_topics_empty_request, + build_fetch_empty_topics_request, build_list_offsets_request, + build_metadata_all_topics_flexible, build_metadata_all_topics_legacy, + build_metadata_flexible_request_v10, +}; + +#[test] +fn supported_ranges_table_has_six_entries() { + assert_eq!(supported_api_ranges().len(), 6); +} + +#[test] +fn is_supported_version_matches_scope_table() { + for &(api_key, _, min_ver, max_ver) in SCOPED_API_KEYS { + assert!( + !is_supported_version(api_key, min_ver - 1), + "key {api_key} must reject v{}", + min_ver - 1 + ); + assert!( + is_supported_version(api_key, min_ver), + "key {api_key} must accept min v{min_ver}" + ); + assert!( + is_supported_version(api_key, max_ver), + "key {api_key} must accept max v{max_ver}" + ); + assert!( + !is_supported_version(api_key, max_ver + 1), + "key {api_key} must reject v{}", + max_ver + 1 + ); + } +} + +#[test] +fn apiversions_advertises_exact_supported_ranges_v1() { + let body = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_i32().unwrap()).expect("api count fits usize"); + assert_eq!(count, supported_api_ranges().len()); + + for expected in supported_api_ranges() { + let key = d.read_i16().unwrap(); + let min = d.read_i16().unwrap(); + let max = d.read_i16().unwrap(); + assert_eq!(key, expected.api_key); + assert_eq!( + min, + advertised_min_version(expected.api_key, expected.min_version) + ); + assert_eq!(max, expected.max_version); + } + assert_eq!(d.read_i32().unwrap(), 0); // throttle + assert_eq!(d.remaining(), 0); +} + +#[test] +fn apiversions_advertises_exact_supported_ranges_v3_flexible() { + let body = handle_request( + API_KEY_API_VERSIONS, + 3, + build_api_versions_flexible_request("iggy-test", "0.1.0"), + &default_broker(), + ) + .expect_response("test request has acks != 0 and expects a response"); + 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, supported_api_ranges().len()); + + for expected in supported_api_ranges() { + let key = d.read_i16().unwrap(); + let min = d.read_i16().unwrap(); + let max = d.read_i16().unwrap(); + d.read_tagged_fields().unwrap(); + assert_eq!(key, expected.api_key); + assert_eq!( + min, + advertised_min_version(expected.api_key, expected.min_version) + ); + assert_eq!(max, expected.max_version); + } + assert_eq!(d.read_i32().unwrap(), 0); + d.read_tagged_fields().unwrap(); + assert_eq!(d.remaining(), 0); +} + +#[test] +fn apiversions_advertises_produce_min_zero_while_firewall_stays_three() { + let range = supported_api_ranges() + .iter() + .find(|r| r.api_key == API_KEY_PRODUCE) + .expect("produce range"); + assert_eq!(range.min_version, 3); + assert_eq!( + advertised_min_version(API_KEY_PRODUCE, range.min_version), + 0 + ); + assert!(!is_supported_version(API_KEY_PRODUCE, 0)); +} + +#[test] +fn apiversions_all_versions_return_success() { + for version in 0i16..=3 { + let request = if version >= 3 { + build_api_versions_flexible_request("iggy-test", "0.1.0") + } else { + Bytes::new() + }; + let body = handle_request(API_KEY_API_VERSIONS, version, request, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0, "ApiVersions v{version}"); + } +} + +#[test] +fn apiversions_out_of_range_returns_unsupported_in_body() { + let body = handle_request(API_KEY_API_VERSIONS, 99, Bytes::new(), &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION); +} + +fn metadata_request_one_topic() -> Bytes { + build_metadata_legacy_request(&["test-topic"]) +} + +#[test] +fn metadata_below_min_version_closes_connection() { + assert!( + handle_request( + API_KEY_METADATA, + -1, + metadata_request_one_topic(), + &default_broker(), + ) + .is_close(), + "Metadata below supported min must close rather than return a clamped body" + ); +} + +#[test] +fn metadata_above_max_version_closes_connection() { + // v10 request uses flexible encoding; a clamped v9 reply would not survive client parsing. + assert!( + handle_request( + API_KEY_METADATA, + 10, + build_metadata_flexible_request_v10(&["test-topic"]), + &default_broker(), + ) + .is_close(), + "Metadata above supported max must close rather than return a clamped body" + ); +} + +#[tokio::test] +async fn e2e_metadata_above_max_version_closes_tcp_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let body = build_metadata_flexible_request_v10(&["orders"]); + let frame = build_request_frame(API_KEY_METADATA, 10, 44, Some("n9-test"), &body); + stream.write_all(&frame).await.expect("write metadata v10"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "unsupported Metadata version must close the connection" + ); +} + +#[test] +fn produce_below_min_version_with_nonzero_acks_closes_connection() { + // Produce v2 is below both the firewall min (3) and `kafka_protocol`'s schema floor (3-13) + // - no encodable response exists at this version, so a client expecting a reply (acks != 0) + // gets a close instead of the pre-migration downgraded error response. acks=0 still keeps + // the connection open - see `produce_advertises_min_zero_but_firewall_rejects_below_v3` and + // `api::handle_produce_request`'s hand-peeked acks path. + let body = handle_request( + API_KEY_PRODUCE, + 2, + build_produce_v2_body(1, 0), + &default_broker(), + ); + assert!( + body.is_close(), + "Produce v2 with acks != 0 has no encodable response shape and must close" + ); +} + +#[test] +fn fetch_below_min_version_closes_connection() { + // Fetch v3 is below both the firewall min (4) and `kafka_protocol`'s schema floor (4-18) - + // no encodable response exists at this version, so this closes instead of the + // pre-migration downgraded error response. + assert!( + handle_request(API_KEY_FETCH, 3, Bytes::new(), &default_broker()).is_close(), + "Fetch v3 has no encodable response shape and must close" + ); +} + +#[test] +fn fetch_unsupported_version_above_max_closes_connection() { + // Fetch v13+ response shape differs from the v12 encoder; a clamped body is unparsable. + assert!( + handle_request(API_KEY_FETCH, 13, Bytes::new(), &default_broker()).is_close(), + "Fetch above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn produce_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_PRODUCE, 13, Bytes::new(), &default_broker()).is_close(), + "Produce above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn create_topics_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_CREATE_TOPICS, 7, Bytes::new(), &default_broker()).is_close(), + "CreateTopics above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn list_offsets_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_LIST_OFFSETS, 7, Bytes::new(), &default_broker()).is_close(), + "ListOffsets above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn list_offsets_v0_closes_connection() { + // `kafka_protocol` has no encoder for ListOffsets v0's legacy `old_style_offsets` shape (it + // predates the schema the crate generates from), so a v0 request - already below the + // firewall's min=1 - now closes instead of getting the pre-migration downgraded response. + assert!( + handle_request(API_KEY_LIST_OFFSETS, 0, Bytes::new(), &default_broker()).is_close(), + "ListOffsets v0 has no encodable response shape and must close" + ); +} + +#[test] +fn create_topics_below_min_version_closes_connection() { + // CreateTopics v1 is below both the firewall min (2) and `kafka_protocol`'s schema floor + // (2-7) - no encodable response exists at this version, so this closes instead of the + // pre-migration downgraded error response. + assert!( + handle_request(API_KEY_CREATE_TOPICS, 1, Bytes::new(), &default_broker()).is_close(), + "CreateTopics v1 has no encodable response shape and must close" + ); +} + +#[test] +fn unsupported_api_keys_close_connection() { + for key in [8, 9, 10, 11, 17, 20, 42, 999] { + let outcome = handle_request(key, 0, Bytes::new(), &default_broker()); + assert!( + outcome.is_close(), + "unknown api_key {key} must close (no parseable response schema)" + ); + } +} + +#[test] +fn supported_produce_versions_accept_valid_fixture() { + for version in 3i16..=9 { + let Some(body) = load_fixture_body_or_skip(0, "Produce", version) else { + continue; + }; + let resp = handle_request(API_KEY_PRODUCE, version, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + assert!(!resp.is_empty(), "Produce v{version} response empty"); + } +} + +#[test] +fn supported_fetch_versions_accept_valid_fixture() { + for version in 4i16..=12 { + let Some(body) = load_fixture_body_or_skip(1, "Fetch", version) else { + continue; + }; + let resp = handle_request(API_KEY_FETCH, version, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + assert!(!resp.is_empty(), "Fetch v{version} response empty"); + } +} + +#[test] +fn corrupt_produce_body_with_acks_stays_silent() { + // `kafka_protocol` decodes Produce in one shot, so a decode failure never exposes `acks` + // (unlike the pre-migration field-by-field decoder, which could still answer with + // INVALID_REQUEST once it knew acks was nonzero). Every Produce decode failure now stays + // silent regardless of whether acks was readable before the truncation. + let body = Bytes::from_static(&[ + 0xFF, 0xFF, // null transactional_id + 0x00, 0x01, // acks = 1 + 0x00, 0x00, 0x00, 0x00, // timeout_ms = 0 + 0xFF, 0xFF, 0xFF, // truncated topics count + ]); + assert!( + handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).is_no_response(), + "malformed Produce body must stay silent regardless of acks" + ); +} + +#[test] +fn corrupt_produce_body_before_acks_is_silent() { + // Decode fails before acks is read: the client's response expectation is unknowable, and an + // error response could desync an acks=0 fire-and-forget client, so the server stays silent. + let body = Bytes::from_static(&[0xFF, 0xFF]); // null transactional_id, then EOF + let outcome = handle_request(API_KEY_PRODUCE, 3, body, &default_broker()); + assert!( + outcome.is_no_response(), + "produce decode failure before acks must be silent" + ); +} + +#[test] +fn corrupt_fetch_body_returns_invalid_request_error() { + let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF]); + let resp = handle_request(API_KEY_FETCH, 4, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(resp); + assert_eq!(d.read_i32().unwrap(), 0); + assert_eq!(d.read_i32().unwrap(), 1); + assert_eq!(d.read_nullable_string().unwrap(), Some(String::new())); + assert_eq!(d.read_i32().unwrap(), 1); + assert_eq!(d.read_i32().unwrap(), 0); + assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST); +} + +// ── ListOffsets v0 (no encodable representation in kafka_protocol) ───────── + +#[test] +fn list_offsets_v0_with_topic_closes_connection() { + let request_body = build_list_offsets_v0_request_with_topic_t(); + assert!( + handle_request(API_KEY_LIST_OFFSETS, 0, request_body, &default_broker()).is_close(), + "ListOffsets v0 has no encodable response shape and must close, even with a well-formed body" + ); +} + +// ── Comprehensive scoped-API coverage (correlation id, boundary versions) ── + +fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes { + match api_key { + API_KEY_METADATA => { + if version >= 9 { + build_metadata_all_topics_flexible(version) + } else { + build_metadata_all_topics_legacy(version) + } + } + API_KEY_API_VERSIONS => { + if version >= 3 { + build_api_versions_flexible_request("iggy-test", "0.1.0") + } else { + Bytes::new() + } + } + API_KEY_PRODUCE => { + if fixture_exists(api_key, name, version) { Review Comment: **This bypasses `KAFKA_FIXTURES_REQUIRED` with a silent synthetic fallback.** Also at `:437` and `:444`. These three arms use `fixture_exists` + `load_fixture_body` directly instead of `load_fixture_body_or_skip`, so the env-var escalation added in `tests/common/fixtures.rs:63-71` does not apply here. Compounding it: `cmd_generate` (`tools/kafka-tool/src/main.rs:630`) does `Err(e) => warn!("SKIP {} v{}: {e}", name, v)` and still returns `Ok`, so `set -e` in `scripts/ci-wire-fixtures.sh` cannot observe a partial generation. Concrete failure: suppose `build_framed` starts failing for Produce v9 only, after a `kafka-protocol` bump changes an encoder's version floor. `fixtures_canary_tests.rs` passes because other `.bin` files exist. `load_fixture_body_or_skip` is not called here, so no panic fires. `request_body_for_scoped_api` silently substitutes `build_produce_flexible_body(1, 0)`. And `each_scoped_api_max_version_preserves_correlation_id_e2e` (`:475`) passes while asserting only `corr == corr` and `!resp_body.is_empty()`. This is the one suite that iterates the whole scope matrix, so it is the worst place to leave that hole. Fix: route these three arms through a helper that honours the env var - e.g. `load_fixture_body_or_fallback(api_key, name, version, || synthetic)` that panics when `KAFKA_FIXTURES_REQUIRED=1` and the fixture is missing. Tradeoff: the min/max e2e tests then hard-fail in CI whenever the generator regresses for any (key, version) in the matrix. That is the intent, but it converts a silent degradation into a red build that a `kafka-protocol` bump will trip. ########## gateways/kafka/tests/listener_robustness_tests.rs: ########## @@ -0,0 +1,585 @@ +// 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. + +//! TCP listener robustness - framing, pipelining, concurrency, edge cases. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::time::Duration; + +use bytes::{BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time; + +use iggy_gateway_kafka::GatewayConfig; +use iggy_gateway_kafka::protocol::api::{ + API_KEY_API_VERSIONS, API_KEY_FETCH, API_KEY_METADATA, API_KEY_PRODUCE, ERROR_INVALID_REQUEST, +}; + +use codec::Decoder; +use server::{spawn_test_server, spawn_test_server_with_config}; +use tcp::{ + ByteRead, build_request_frame, concat_frames, parse_response_payload, read_byte_with_timeout, + read_response_frame, read_response_frame_with_timeout, scan_for_error_code, +}; + +#[tokio::test] +async fn e2e_pipelined_requests_receive_responses_in_order() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let frame1 = build_request_frame(API_KEY_API_VERSIONS, 1, 1, Some("pipe-test"), &[]); + let frame2 = build_request_frame(API_KEY_API_VERSIONS, 1, 2, Some("pipe-test"), &[]); + let frame3 = build_request_frame(API_KEY_API_VERSIONS, 1, 3, Some("pipe-test"), &[]); + stream + .write_all(&concat_frames(&[frame1, frame2, frame3])) + .await + .expect("pipelined write"); + + for expected_corr in 1..=3 { + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, body) = parse_response_payload(API_KEY_API_VERSIONS, 1, payload); + assert_eq!(corr, expected_corr); + assert_eq!(Decoder::new(body).read_i16().unwrap(), 0); + } +} + +#[tokio::test] +async fn e2e_partial_length_prefix_then_remainder_accepted() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 42, Some("partial-test"), &[]); + assert!(frame.len() > 6, "test frame long enough to split"); + + stream.write_all(&frame[..2]).await.expect("partial prefix"); + time::sleep(Duration::from_millis(50)).await; + stream.write_all(&frame[2..]).await.expect("remainder"); + + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, _) = parse_response_payload(API_KEY_API_VERSIONS, 1, payload); + assert_eq!(corr, 42); +} + +#[tokio::test] +async fn e2e_frame_within_custom_max_frame_size_accepted() { + let max_frame = 512; + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: max_frame, + 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), + }) + .await; + + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 55, Some("max-frame-test"), &[]); + assert!( + frame.len() <= max_frame, + "ApiVersions frame must fit test max ({max_frame})" + ); + + stream.write_all(&frame).await.expect("write"); + let payload = read_response_frame(&mut stream, max_frame).await; + assert_eq!( + parse_response_payload(API_KEY_API_VERSIONS, 1, payload).0, + 55 + ); +} + +#[tokio::test] +async fn e2e_frame_exceeding_max_frame_size_closes_connection() { + let max_frame = 64; + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: max_frame, + 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), + }) + .await; + + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let mut frame = BytesMut::new(); + frame.put_i32(200); + frame.resize(4 + 200, 0); + stream.write_all(&frame).await.expect("oversized frame"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "oversized frame should close connection (EOF)" + ); +} + +#[tokio::test] +async fn e2e_truncated_frame_body_closes_connection() { + // A truncated in-flight body closes only once the server's read_timeout elapses, so use a + // short read_timeout and wait longer than it to observe a genuine close, not a mere stall. + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: 8 * 1024 * 1024, + max_connections: 1024, + idle_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(1), + write_timeout: Duration::from_secs(5), + shutdown_drain_timeout: Duration::from_secs(5), + }) + .await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let full = build_request_frame(API_KEY_API_VERSIONS, 1, 66, Some("trunc-test"), &[]); + let payload_len = u32::from_be_bytes([full[0], full[1], full[2], full[3]]) as usize; + assert!(full.len() >= 4 + payload_len); + + stream + .write_all(&full[..4 + payload_len / 2]) + .await + .expect("half body"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(3)).await, + ByteRead::Closed, + "truncated body should close connection after read_timeout" + ); +} + +#[tokio::test] +async fn e2e_multiple_concurrent_connections_are_independent() { + let (addr, _shutdown) = spawn_test_server().await; + + let (r1, r2, r3) = tokio::join!( + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 101, &[]), + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 102, &[]), + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 103, &[]), + ); + + assert_eq!(r1.0, 101); + assert_eq!(r2.0, 102); + assert_eq!(r3.0, 103); +} + +#[tokio::test] +async fn e2e_client_disconnect_mid_frame_allows_new_connection() { + let (addr, _shutdown) = spawn_test_server().await; + + { + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let full = build_request_frame(API_KEY_API_VERSIONS, 1, 77, Some("abort-test"), &[]); + stream.write_all(&full[..8]).await.expect("partial write"); + drop(stream); + } + + time::sleep(Duration::from_millis(100)).await; + + let (corr, body) = tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 78, &[]).await; + assert_eq!(corr, 78); + assert_eq!(Decoder::new(body).read_i16().unwrap(), 0); +} + +#[tokio::test] +async fn e2e_response_frames_have_positive_big_endian_length_prefix() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let request = wire::build_api_versions_flexible_request("iggy-test", "0.1.0"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 3, 200, Some("len-test"), &request); + stream.write_all(&frame).await.expect("write"); + + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .await + .expect("length prefix"); Review Comment: **This test "verifies" endianness by hanging, and its second assertion cannot fail.** `e2e_response_frames_have_positive_big_endian_length_prefix` asserts `len > 0` and `!body.is_empty()`. The second is implied by the first, so it adds nothing. More importantly, nothing here can distinguish big-endian from little-endian by assertion. If the server wrote the length prefix little-endian, `len` becomes a large number and the `read_exact` at `:229-231` blocks until nextest kills the test at 300 s. A hang is not a failure signal - and with `[profile.ci] retries = 3` it burns 15 minutes before reporting. Also `:583` in this file uses a raw `stream.read(&mut buf).await.expect("read after shutdown")` while everything else here correctly uses `read_byte_with_timeout`. Fix: assert `len_buf == (expected_body_len as i32).to_be_bytes()` against an independently known body length, or delete the test - `parse_response_payload` already depends on big-endian framing in roughly 23 places. For `:583`, use `read_byte_with_timeout(..., 2s) == ByteRead::Closed`. ########## gateways/kafka/tests/common/wire.rs: ########## @@ -0,0 +1,491 @@ +// 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. + +//! Kafka wire request builders aligned with SCOPE.md / protocol spec. +//! +//! Callers must also declare `#[path = "common/codec.rs"] mod codec;` at their own crate root - +//! this file borrows that module via `super::codec` rather than redeclaring it, since `rustc` +//! rejects loading the same file as two distinct modules in one crate. +#![allow(dead_code)] Review Comment: **`#![allow(dead_code)]` is hiding seven dead helpers, and one of them is load-bearing.** Referenced nowhere outside their own definitions: `FLEXIBLE_FROM_VERSION` (`:45`), `build_fetch_v2_default_max_bytes_request` (`:220`), `build_produce_flexible_request_with_topic` (`:230`), `build_fetch_v3_no_isolation_request` (`:247`), `build_fetch_request_with_sections` (`:309`), `build_list_offsets_branch_request` (`:399`), `build_create_topics_request_with_sections` (`:435`). `build_fetch_request_with_sections` is 88 lines and is the only builder in the suite that produces a Fetch request with real topics, partitions, forgotten-topics and rack. Because it is dead, **no test sends a Fetch with a non-empty topics array except through a `.bin` fixture** - so `encode_fetch_response`'s per-partition echo loop (`src/protocol/responses.rs:134-152`) is covered only by the fixture-backed tests. Same shape for `encode_list_offsets_response` and `encode_create_topics_response`. `FLEXIBLE_FROM_VERSION` (`:45`) is also a third copy of the flexible-threshold table, after `header_tests.rs:39` and `kafka-protocol` itself. Fix: wire `build_fetch_request_with_sections` into a test - it would give the Fetch echo loop non-fixture coverage - and delete the rest, then drop `#![allow(dead_code)]` from this file so the next one is caught. Tradeoff: removing the allow from a `#[path]`-included module makes it warn in every test binary that includes the module without using every item, which is why the allow is there. Keeping the allow on `tcp.rs`/`codec.rs` (genuinely shared broadly) and deleting the dead items from `wire.rs` gets both. Also `build_metadata_flexible_request_for_version` (`:81`) and `extract_body_from_framed_message` (`fixtures.rs:86`) are `pub` but internal-only. ########## gateways/kafka/scripts/ci-wire-fixtures.sh: ########## @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# 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. + +# Generate or remove gitignored kafka-tool wire fixtures for CI. +# Run from the iggy workspace root. + +set -euo pipefail + +FIXTURES_DIR="gateways/kafka/tools/kafka-tool/kafka_messages" + +# API keys exercised by decode_validation_tests and handler_regression_tests. Review Comment: **This comment names two test suites that do not exist.** Neither `decode_validation_tests` nor `handler_regression_tests` is in this PR. The suites that actually consume these fixtures are `api_handler_tests.rs`, `server_e2e_tests.rs` and `version_firewall_tests.rs`, via `load_fixture_body_or_skip` / `fixture_exists`. This is the load-bearing instance of the drift: this line is the authoritative note explaining *why* `FIXTURE_API_KEYS=(0 1 2 19)` is that specific list. Someone who renames or deletes the real consumers and greps for `decode_validation_tests` to check whether this list is still needed finds nothing, and then either deletes a live entry or leaves a dead one. The identical stale comment is at `.github/actions/rust/pre-merge/action.yml:282`. Fix: `# API keys requested by api_handler_tests, version_firewall_tests, and server_e2e_tests.` Separately: `generate()` loops `cargo run` once per key, four times per job across three partition jobs - twelve invocations. `--api-key` already takes `ArgAction::Append` (`tools/kafka-tool/src/main.rs:66`) and `generate_cli_tests.rs` proves repeated flags work, so one invocation would do. ########## gateways/kafka/tests/header_tests.rs: ########## @@ -0,0 +1,282 @@ +// 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. + +//! `request_header_version` / `response_header_version` are thin wrappers around +//! `kafka_protocol::messages::ApiKey` (see `src/protocol/header.rs`); decoding/encoding the +//! header bytes themselves is `kafka_protocol::messages::RequestHeader`/`ResponseHeader`'s own +//! tested responsibility, not re-tested here. These tests cover the gateway-specific policy +//! layered on top: the unknown-API-key fallback and the `ApiVersions` response-header special case. + +use kafka_protocol::messages::ApiKey; + +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, +}; +use iggy_gateway_kafka::protocol::header::{request_header_version, response_header_version}; + +/// Flexible-encoding threshold per API key (mirrors `protocol/header.rs`; cross-checked against +/// the independent `kafka-protocol` crate below rather than trusted on its own). +/// +/// Keys 4-7 (LeaderAndIsr/StopReplica/UpdateMetadata/ControlledShutdown) are inter-broker-only +/// APIs `kafka_protocol` 0.17 does not implement (`ApiKey::try_from` fails for them), so this +/// gateway's wrapper always falls back to header v1 for them - `i16::MAX`, not their legacy +/// threshold from the pre-migration hand-rolled table. +const API_KEY_FLEXIBLE_FROM: &[(i16, i16)] = &[ + (0, 9), + (1, 12), + (2, 6), + (3, 9), + (4, i16::MAX), + (5, i16::MAX), + (6, i16::MAX), + (7, i16::MAX), + (8, 8), + (9, 6), + (10, 3), + (11, 6), + (12, 4), + (13, 4), + (14, 4), + (15, 5), + (16, 3), + (17, i16::MAX), + (18, 3), + (19, 5), + (20, 4), + (21, 2), + (22, 2), + (23, 4), + (24, 3), + (25, 3), + (26, 3), + // WriteTxnMarkers' only valid versions are 1-2 (no v0 on the real wire) and both are + // flexible; `kafka_protocol` encodes this as an unconditional header v2, matching the `0` + // ("always flexible") arm below rather than a real threshold. + (27, 0), + (28, 3), + (29, 2), + (30, 2), + (31, 2), + (32, 4), + (33, 2), + (34, 2), + (35, 2), + (36, 2), + (37, 2), + (38, 2), + (39, 2), + (40, 2), + (41, 2), + (42, 2), + (43, 2), + (44, 1), + (45, 0), + (46, 0), + (47, i16::MAX), + (48, 1), + (49, 1), + (50, 0), + (51, 0), + (55, 0), + (56, 0), + (57, 0), + (60, 0), + (61, 0), + (64, 0), + (65, 0), + (66, 0), + (67, 0), + (68, 0), + (69, 0), + (71, 0), + (72, 0), + (74, 0), + (75, 0), + (76, 0), + (77, 0), + (78, 0), + (79, 0), + (80, 0), +]; + +#[test] +fn request_header_version_matches_independent_kafka_protocol_crate() { + // API_KEY_FLEXIBLE_FROM is hand-transcribed from header.rs's threshold table, so comparing + // request_header_version only against that same mirror can't catch a value wrong in both + // places (the same transcription mistake copied twice). Cross-check against the third-party + // `kafka-protocol` crate's own per-key header-version logic instead, over every version that + // crate considers actually valid for the key - outside that range a version never appeared + // on the real wire, so there's no independently-meaningful answer to compare against. + for &(api_key, _) in API_KEY_FLEXIBLE_FROM { + let Ok(external) = ApiKey::try_from(api_key) else { + continue; + }; + let range = external.valid_versions(); + for version in range.min..=range.max { + assert_eq!( Review Comment: **This assertion cannot fail.** `src/protocol/header.rs:30` is: ```rust ApiKey::try_from(api_key).map_or(1, |key| key.request_header_version(api_version)) ``` The test already filtered to `Ok(external)` at `:127-129`, so for every `api_key` that reaches this assertion the left-hand side expands to exactly the right-hand side. It asserts `X == X` - no input, no refactor, and no `kafka-protocol` upgrade can make it fail. The doc comment at `:119-124` claims the opposite, specifically: "comparing `request_header_version` only against that same mirror can't catch a value wrong in both places ... Cross-check against the third-party `kafka-protocol` crate's own per-key header-version logic instead." That is the one thing this test does not do. Same shape at `:169-188`: `expected_at_v0` is derived from `request_header_version(k, 0) >= 2`, i.e. from the same crate function the assertion then checks, so it verifies `kafka-protocol`'s internal consistency rather than the gateway's policy. The sibling `request_header_version_hits_every_api_key_match_arm` (`:141-166`) is the one carrying real value - it pins all 66 rows of the hand-transcribed `API_KEY_FLEXIBLE_FROM` table against the crate and would fail on an upgrade that moves a threshold. Fix: delete this assertion, or narrow it to the keys where `ApiKey::try_from` *fails* (4, 5, 6, 7, 17, 47) - which is exactly the set it currently `continue`s straight past, and the only place the `map_or(1, ...)` fallback is observable. Tradeoff: dropping it removes the tripwire that would fire if `header.rs` ever stopped delegating to `ApiKey`; if that is the intent, say so in the comment instead of claiming a cross-check. Note `API_KEY_FLEXIBLE_FROM`'s own doc at `:32-38` says it "mirrors `protocol/header.rs`". There is no table in `header.rs` to mirror - that line is stale too. ########## gateways/kafka/src/main.rs: ########## @@ -0,0 +1,172 @@ +// 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::fmt::Display; +use std::str::FromStr; +use std::time::Duration; + +use tokio::net::TcpListener; +use tokio::signal; +use tokio::sync::{Semaphore, broadcast}; + +use iggy_gateway_kafka::server::init_tracing; +use iggy_gateway_kafka::{GatewayConfig, KafkaGateway}; + +#[tokio::main] +async fn main() -> Result<(), Box<dyn std::error::Error>> { + init_tracing(); + + let config = load_config()?; + + let listener = TcpListener::bind(&config.bind_addr) Review Comment: **Accept backlog is 128 while `max_connections` is 1024.** `tokio::net::TcpListener::bind` delegates to mio, which hardcodes the backlog: `mio-1.2.2/src/net/tcp/listener.rs:87-93` calls `listen(fd, 128)` on every non-Horizon/Haiku target. So the semaphore admits 1024 concurrent connections while the kernel accept queue holds 128. A burst of 1024 simultaneous connects - a consumer-group rebalance, a deployment rollout, a load-balancer failover - overflows the queue. With `net.ipv4.tcp_abort_on_overflow=0` (the default) the SYNs are silently dropped and clients retry on the exponential SYN backoff: 1s, 3s, 7s. That reads as a network fault, not as a server limit. Repo precedent for the fix: `core/message_bus/src/socket_opts.rs:63` builds the listener with `socket2` and calls `socket.listen(libc::SOMAXCONN)` (4096 on Linux >= 5.4). `socket2` is already a dependency of this crate. Tradeoff: a deeper backlog lets more SYN-flood state accumulate in the kernel. Also `tests/common/server.rs` binds its own listeners, so the tuned path stays untested unless the bind is factored into a shared helper. ########## gateways/kafka/tests/server_integration_tests.rs: ########## @@ -0,0 +1,179 @@ +// 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. + +#[path = "common/codec.rs"] +mod codec; + +use std::time::Duration; + +use bytes::{Buf, BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; + +use codec::Encoder; +use iggy_gateway_kafka::server::read_frame; + +async fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let client = tokio::spawn(async move { TcpStream::connect(addr).await.unwrap() }); + let (server, _) = listener.accept().await.unwrap(); + let client = client.await.unwrap(); + (client, server) +} + +/// Raw length-prefixed write (no Kafka response header) - mirrors `server::write_frame`. +async fn write_length_prefixed( + stream: &mut TcpStream, + payload: &[u8], + write_timeout: Duration, +) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { + let len = payload.len(); + assert!(i32::try_from(len).is_ok()); + let mut frame = BytesMut::with_capacity(4 + len); + frame.put_i32(i32::try_from(len).expect("len fits i32")); + frame.extend_from_slice(payload); + tokio::time::timeout(write_timeout, stream.write_all(&frame)) + .await + .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "write timeout"))??; + Ok(()) +} + +#[tokio::test] +async fn read_frame_reads_valid_payload() { + let (mut client, mut server) = tcp_pair().await; + + let mut enc = Encoder::with_capacity(64); + enc.write_i16(18); + enc.write_i16(3); + enc.write_i32(123); + enc.write_nullable_string(Some("test-client")).unwrap(); + let payload = enc.freeze(); + + let mut frame = BytesMut::with_capacity(4 + payload.len()); + frame.extend_from_slice( + &i32::try_from(payload.len()) + .expect("test payload fits i32") + .to_be_bytes(), + ); + frame.extend_from_slice(&payload); + client.write_all(&frame).await.unwrap(); + + let parsed = read_frame( + &mut server, + 4096, + Duration::from_secs(5), + Duration::from_secs(1), + ) + .await + .unwrap(); + assert_eq!(parsed, payload); +} + +#[tokio::test] +async fn write_frame_writes_length_prefixed_payload() { Review Comment: **These two tests exercise a test-local helper, not production code.** The helper at `:40-54` has a doc comment saying it "mirrors `server::write_frame`". There is no `write_frame` in `src/server.rs`. The production writer is `send_response` (`src/server.rs:391`), which prefixes a Kafka response header - correlation id, plus a tagged-fields byte at header v1 - that this helper does not write. So `write_frame_writes_length_prefixed_payload` and `write_frame_length_prefix_is_big_endian` (`:137`) assert properties of 15 lines of test code. Make `send_response` write its length prefix little-endian and both still pass. The only real coverage of that framing is `send_response_writes_header_and_body` in `src/server.rs:527`. Fix: delete both tests and the helper, or point them at `send_response`. Tradeoff: this file then contains only `read_frame` tests and should be renamed accordingly - it already duplicates two of `src/server.rs`'s own `read_frame` unit tests. ########## gateways/kafka/tests/server_e2e_tests.rs: ########## @@ -0,0 +1,466 @@ +// 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/codec.rs"] +mod codec; +#[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_NOT_LEADER_OR_FOLLOWER, +}; + +use 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 request = wire::build_api_versions_flexible_request("iggy-test", "0.1.0"); + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 3, 42_002, &request).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_closes_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + // Unknown api key (8, OffsetCommit) has no response schema this gateway can encode, so the + // server closes the connection without a (misparseable) response body. + let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]); + stream.write_all(&frame1).await.unwrap(); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "unknown api key must close the connection without a 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); + } +} + +// Negative-frame-length-closes-connection coverage lives in listener_robustness_tests.rs +// (uses a timeout-guarded read helper, so a regression fails fast instead of hanging). + +#[tokio::test] +async fn e2e_oversized_frame_is_rejected() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let mut frame = BytesMut::new(); + frame.put_i32(10_000_000); // exceeds default 8 MiB cap Review Comment: **`unwrap_or(0)` maps every failure mode onto the success value.** ```rust let n = stream.read(&mut buf).await.unwrap_or(0); assert_eq!(n, 0, "server should close after oversized frame"); ``` Any I/O error becomes `0`, which is the asserted value. And there is no timeout, so if the server accepted the 10 MB declared frame and waited for a body, this blocks until nextest's `slow-timeout = { period = "60s", terminate-after = 5 }` kills it at 300 s. Four lines above, `:143-144` explains that the negative-frame-length test was moved to `listener_robustness_tests.rs` precisely because it "uses a timeout-guarded read helper, so a regression fails fast instead of hanging" - and then this test does the unguarded read. It is a strictly weaker duplicate of `tests/listener_robustness_tests.rs:118-144`, which uses `ByteRead::Closed` with a 2 s bound. Fix: delete this test. Tradeoff: the deleted one uses the *default* 8 MiB cap while the survivor uses a 64-byte override, so if the default should stay covered, change the survivor's config rather than keeping both. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,558 @@ +// 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 bytes::{Buf, Bytes}; +use kafka_protocol::messages::api_versions_response::ApiVersion; +use kafka_protocol::messages::metadata_response::{MetadataResponseBroker, MetadataResponseTopic}; +use kafka_protocol::messages::{ + ApiVersionsRequest, ApiVersionsResponse, BrokerId, CreateTopicsRequest, FetchRequest, + ListOffsetsRequest, MetadataRequest, MetadataResponse, ProduceRequest, TopicName, +}; +use kafka_protocol::protocol::{Decodable, StrBytes}; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::responses::{ + encode_create_topics_error_response, encode_create_topics_response, + encode_fetch_error_response, encode_fetch_response, encode_list_offsets_error_response, + encode_list_offsets_response, encode_message, encode_produce_error_response, + encode_produce_response, +}; + +pub const API_KEY_PRODUCE: i16 = 0; +pub const API_KEY_FETCH: i16 = 1; +pub const API_KEY_LIST_OFFSETS: i16 = 2; +pub const API_KEY_METADATA: i16 = 3; +pub const API_KEY_API_VERSIONS: i16 = 18; +pub const API_KEY_CREATE_TOPICS: i16 = 19; + +pub const DEFAULT_KAFKA_PORT: u16 = 9093; + +pub const ERROR_NONE: i16 = 0; +pub const ERROR_UNKNOWN_TOPIC_OR_PARTITION: i16 = 3; +/// Retriable; Produce stub uses this until the Iggy bridge persists records. +pub const ERROR_NOT_LEADER_OR_FOLLOWER: i16 = 6; +pub const ERROR_UNSUPPORTED_VERSION: i16 = 35; +pub const ERROR_INVALID_PARTITIONS: i16 = 37; +pub const ERROR_INVALID_REPLICATION_FACTOR: i16 = 38; +/// `CreateTopics` stub: do not claim topics were created (no controller / no Iggy bridge). +pub const ERROR_NOT_CONTROLLER: i16 = 41; +pub const ERROR_INVALID_REQUEST: i16 = 42; + +/// Result of handling one Kafka request body. +#[derive(Debug)] +pub enum HandleOutcome { + /// Write this response body (with a response header). + Respond(Bytes), + /// Produce with `acks=0`: write nothing, keep the connection open. + NoResponse, + /// No parseable response exists for this request; close the TCP connection. + Close, +} + +impl HandleOutcome { + /// Return the response body, or panic with `msg` if the outcome is not [`Self::Respond`]. + /// + /// # Panics + /// + /// Panics when the outcome is [`Self::NoResponse`] or [`Self::Close`]. + #[must_use] + pub fn expect_response(self, msg: &str) -> Bytes { + match self { + Self::Respond(body) => body, + Self::NoResponse => panic!("{msg}: got NoResponse"), + Self::Close => panic!("{msg}: got Close"), + } + } + + #[must_use] + pub const fn is_no_response(&self) -> bool { + matches!(self, Self::NoResponse) + } + + #[must_use] + pub const fn is_close(&self) -> bool { + matches!(self, Self::Close) + } +} + +#[derive(Debug, Clone)] +pub struct BrokerAdvertise { + pub host: String, + pub port: i32, +} + +impl Default for BrokerAdvertise { + fn default() -> Self { + Self { + host: "127.0.0.1".to_string(), + port: i32::from(DEFAULT_KAFKA_PORT), + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct ApiVersionRange { + pub api_key: i16, + pub min_version: i16, + pub max_version: i16, +} + +static SUPPORTED_RANGES: &[ApiVersionRange] = &[ + ApiVersionRange { + api_key: API_KEY_PRODUCE, + min_version: 3, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_FETCH, + min_version: 4, + max_version: 12, + }, + ApiVersionRange { + api_key: API_KEY_LIST_OFFSETS, + min_version: 1, + max_version: 6, + }, + ApiVersionRange { + api_key: API_KEY_METADATA, + min_version: 0, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_API_VERSIONS, + min_version: 0, + max_version: 3, + }, + ApiVersionRange { + api_key: API_KEY_CREATE_TOPICS, + min_version: 2, + max_version: 5, + }, +]; + +#[must_use] +pub fn supported_api_ranges() -> &'static [ApiVersionRange] { + SUPPORTED_RANGES +} + +/// Handles one decoded request frame and returns how the connection should proceed. +pub fn handle_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> HandleOutcome { + if api_key == API_KEY_PRODUCE { + return handle_produce_request(api_version, body); + } + handle_other_request(api_key, api_version, body, broker) +} + +/// Decode `T` from the whole request body and reject unconsumed trailing bytes. +/// +/// `kafka_protocol`'s `Decodable` stops once it has read the fields its schema defines; it does +/// not know (or care) whether the caller handed it an exact-length body, so the trailing-bytes +/// check has to live here. +fn decode_exhaustive<T: Decodable>(version: i16, mut body: Bytes) -> Result<T> { + let value = + T::decode(&mut body, version).map_err(|e| KafkaProtocolError::Malformed(e.to_string()))?; + if body.has_remaining() { + return Err(KafkaProtocolError::Malformed( + "unexpected trailing bytes in request body".to_string(), + )); + } + Ok(value) +} + +/// Turn an encode [`Result`] into a [`HandleOutcome`], closing the connection when encoding +/// fails rather than propagating - there is no parseable response to send in that case. +fn respond_or_close(result: Result<Bytes>, api_name: &str) -> HandleOutcome { + match result { + Ok(body) => HandleOutcome::Respond(body), + Err(error) => { + tracing::warn!(%error, "failed to encode {api_name} response; closing connection"); + HandleOutcome::Close + } + } +} + +/// Produce is the only request the wire protocol allows to go unanswered +/// (`acks=0`), so it gets its own path that may return [`HandleOutcome::NoResponse`]. +/// +/// The firewall check runs AFTER decoding the request, not before: `ApiVersions` advertises +/// Produce min=0 (see [`advertised_min_version`]) while the firewall's real floor is 3, so a +/// spec-compliant client can legitimately send Produce v0-2 with `acks=0`. Rejecting those +/// versions before reading `acks` would send an error response the client never expects, +/// desyncing the next correlation id it reads. +fn handle_produce_request(api_version: i16, body: Bytes) -> HandleOutcome { + // Above the encoder max there is no response parseable at the client's version, so close + // rather than decode (same policy the other APIs apply). + if api_version > supported_max_version(API_KEY_PRODUCE).unwrap_or(i16::MAX) { + return HandleOutcome::Close; + } + // `kafka_protocol`'s ProduceRequest/ProduceResponse schemas only go back to v3, so v0-2 + // (still advertised as the min in ApiVersions per KAFKA-18659) can be neither decoded nor + // encoded by the crate - there is no parseable response at these versions regardless of + // body content. `acks` is always the first i16 on the wire there (`transactional_id` was + // added in v3), so it's peeked by hand: acks=0 must keep the connection open per the wire + // protocol's fire-and-forget rule even though no response can ever be encoded for it. + if api_version < 3 { + let acks = match body.get(0..2) { + Some(&[hi, lo]) => Some(i16::from_be_bytes([hi, lo])), + _ => None, + }; + return match acks { + Some(0) | None => HandleOutcome::NoResponse, + Some(_) => unsupported_version_response(API_KEY_PRODUCE, api_version, |v| { + encode_produce_error_response(v, ERROR_UNSUPPORTED_VERSION) + }), + }; + } + match decode_exhaustive::<ProduceRequest>(api_version, body) { + // acks=0 is fire-and-forget: the client isn't reading a response, so + // sending one desyncs the next correlation id it expects. + Ok(req) if req.acks == 0 => HandleOutcome::NoResponse, + Ok(req) => { + if !is_supported_version(API_KEY_PRODUCE, api_version) { + return unsupported_version_response(API_KEY_PRODUCE, api_version, |v| { + encode_produce_error_response(v, ERROR_UNSUPPORTED_VERSION) + }); + } + respond_or_close(encode_produce_response(api_version, &req), "Produce") + } + Err(error) => { + // `kafka_protocol` decodes the whole request in one shot; a failure anywhere gives + // no partial-field access, so `acks` is unknowable here (unlike the pre-migration + // field-by-field decoder, which could still know `acks` on a later-field failure). + // Responding risks desyncing an acks=0 fire-and-forget client's correlation stream, + // so every Produce decode failure now stays silent - a behavior change from the + // hand-rolled decoder, which answered with INVALID_REQUEST when `acks` was known and + // nonzero. + tracing::warn!(%error, "failed to decode Produce request (no response)"); Review Comment: **Per-request `warn!` on a connection that stays open, through a synchronous stdout writer.** Also at `api.rs:359` and `server.rs:191`. The writer is configured at `server.rs:483-490`: `tracing_subscriber::fmt()`'s default is `fn() -> io::Stdout`, and `io::Stdout` is a `LineWriter` behind a `ReentrantLock`. That is one global lock acquisition and one `write(2)` per log line, serialized across every worker in the runtime. A client looping malformed Produce bodies on one connection is never disconnected (this arm returns `NoResponse`), so there is no rate limit and no backpressure. Each iteration costs an `anyhow::Error` -> `String` allocation, an event format, the global stdout lock, and a `write(2)`. At scale the `write(2)` becomes the throughput ceiling for the whole gateway, including well-behaved connections. Same shape at `server.rs:191` under a connect flood: one unbounded `warn!` per rejected connection. The repo already solved this. `core/server_common/src/log/logger.rs:257` uses `tracing_appender::non_blocking(io::stdout())` with the comment "Use the non_blocking appender to avoid blocking the threads", and `tracing-appender = "0.2.5"` is already a workspace dependency. Also worth demoting these two decode-failure sites to `debug!` - they are attacker-controlled, not operator-actionable. Tradeoff: `non_blocking` returns a `WorkerGuard` that must outlive `main`, so `init_tracing()` stops returning `()` and `main.rs:31` has to hold it. It also introduces a bounded queue that drops events under load, which makes the shutdown-path `warn!`s at `server.rs:239-242` lossy. ########## gateways/kafka/tests/server_e2e_tests.rs: ########## @@ -0,0 +1,466 @@ +// 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/codec.rs"] +mod codec; +#[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_NOT_LEADER_OR_FOLLOWER, +}; + +use 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 request = wire::build_api_versions_flexible_request("iggy-test", "0.1.0"); + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 3, 42_002, &request).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_closes_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + // Unknown api key (8, OffsetCommit) has no response schema this gateway can encode, so the + // server closes the connection without a (misparseable) response body. + let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]); + stream.write_all(&frame1).await.unwrap(); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "unknown api key must close the connection without a 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); + } +} + +// Negative-frame-length-closes-connection coverage lives in listener_robustness_tests.rs +// (uses a timeout-guarded read helper, so a regression fails fast instead of hanging). + +#[tokio::test] +async fn e2e_oversized_frame_is_rejected() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let mut frame = BytesMut::new(); + frame.put_i32(10_000_000); // exceeds default 8 MiB cap + frame.resize(4 + 100, 0); + stream.write_all(&frame).await.unwrap(); + + let mut buf = [0u8; 1]; + let n = stream.read(&mut buf).await.unwrap_or(0); + assert_eq!(n, 0, "server should close after oversized frame"); +} + +// ── Produce acks=0 (broker must stay silent) ──────────────────────────────── + +#[tokio::test] +async fn e2e_produce_v3_acks_zero_sends_no_response() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let body = build_produce_v3_body(0, 0); + let frame = build_request_frame(API_KEY_PRODUCE, 3, 42, Some("review-test"), &body); + stream + .write_all(&frame) + .await + .expect("write produce acks=0"); + + let response = + read_response_frame_with_timeout(&mut stream, 8 * 1024 * 1024, Duration::from_millis(500)) + .await; + + assert!( + response.is_none(), + "Produce with acks=0 must not receive a response frame (Kafka spec); got {} bytes", + response.as_ref().map_or(0, Bytes::len) + ); +} + +#[tokio::test] +async fn e2e_produce_v3_acks_zero_malformed_topics_sends_no_response() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + // acks=0, claims one topic, no topic bytes - decode fails after acks is read. + let body = build_produce_v3_body(0, 1); + let frame = build_request_frame(API_KEY_PRODUCE, 3, 99, Some("review-test"), &body); + stream + .write_all(&frame) + .await + .expect("write produce acks=0 malformed"); + + let response = + read_response_frame_with_timeout(&mut stream, 8 * 1024 * 1024, Duration::from_millis(500)) + .await; + + assert!( + response.is_none(), + "Produce with acks=0 must stay silent even when the body is malformed; got {} bytes", + response.as_ref().map_or(0, Bytes::len) + ); +} + +#[tokio::test] +async fn e2e_produce_v3_acks_one_still_returns_response() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let body = build_produce_v3_body(1, 0); + let frame = build_request_frame(API_KEY_PRODUCE, 3, 43, Some("review-test"), &body); + stream + .write_all(&frame) + .await + .expect("write produce acks=1"); + + let response = + read_response_frame_with_timeout(&mut stream, 8 * 1024 * 1024, Duration::from_secs(2)) + .await + .expect("Produce with acks=1 should receive a response"); + + let (corr, resp_body) = parse_response_payload(API_KEY_PRODUCE, 3, response); + assert_eq!(corr, 43); + assert!(!resp_body.is_empty()); +} + +// ── ListOffsets v0 (no encodable representation in kafka_protocol) ───────── + +#[tokio::test] +async fn e2e_list_offsets_v0_closes_connection() { + // `kafka_protocol` has no encoder for ListOffsets v0's legacy `old_style_offsets` shape + // (it predates the schema the crate generates from - see `responses::encode_list_offsets_error_response`), + // so a v0 request - already below the firewall's min=1 - now closes instead of getting the + // pre-migration downgraded response. + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let request_body = build_list_offsets_v0_request_with_topic_t(); + let frame = build_request_frame( + API_KEY_LIST_OFFSETS, + 0, + 7, + Some("review-test"), + &request_body, + ); + stream + .write_all(&frame) + .await + .expect("write list offsets v0"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "ListOffsets v0 has no encodable response shape and must close" + ); +} + +// ── Metadata topic name echo (must not hardcode a placeholder topic name) ── + +#[tokio::test] +async fn e2e_metadata_v1_response_contains_requested_topic_name() { + let (addr, _shutdown) = spawn_test_server().await; + let topic = "orders"; + let request_body = build_metadata_legacy_request(&[topic]); + let frame = build_request_frame(API_KEY_METADATA, 1, 9, Some("review-test"), &request_body); + + let mut stream = TcpStream::connect(addr).await.expect("connect"); + stream.write_all(&frame).await.expect("write metadata v1"); + + let payload = + read_response_frame_with_timeout(&mut stream, 8 * 1024 * 1024, Duration::from_secs(2)) + .await + .expect("metadata response"); + + let full_response = { + let mut framed = BytesMut::with_capacity(4 + payload.len()); + framed.put_i32(i32::try_from(payload.len()).expect("metadata response fits i32")); + framed.extend_from_slice(&payload); + framed.freeze() + }; + + assert!( + full_response + .windows(topic.len()) + .any(|window| window == topic.as_bytes()), + "metadata response must contain requested topic name {topic:?}; \ Review Comment: **Byte-substring scan proves less than the test name claims.** `full_response.windows(topic.len()).any(|w| w == b"orders")` passes for implementations that are wrong in ways the name rules out: - the gateway puts "orders" in `cluster_id` instead of the topic slot - it echoes `"ordersXYZ"` - it returns five topics, one of which happens to be "orders" The second assertion - "metadata response must not substitute unknown-topic for requested names" - checks for a string no version of this code has ever emitted, so it cannot fail either. `tests/api_handler_tests.rs:759` already does this structurally. Fix: promote `read_metadata_v1_topics` (`api_handler_tests.rs:726`) into `tests/common/` and assert the decoded name list equals `["orders"]`. Tradeoff: a decoder helper in `tests/common/` gets compiled into every suite via `#[path]`, so it needs its own `#![allow(dead_code)]` or should live in `codec.rs`. ########## gateways/kafka/tools/kafka-tool/tests/generate_cli_tests.rs: ########## @@ -0,0 +1,47 @@ +// 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::process::Command; +use std::time::{SystemTime, UNIX_EPOCH}; + +#[test] +fn generate_accepts_repeated_api_key_flags() { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock") + .as_nanos(); + let output = std::env::temp_dir().join(format!("kafka-gen-test-{nanos}")); + + let status = Command::new(env!("CARGO_BIN_EXE_kafka-message-gen")) + .arg("generate") Review Comment: **The only automated check on the fixture generator asserts nothing about what it generated.** `assert!(status.success())` and nothing else. But `cmd_generate` (`src/main.rs:630`) warns and continues on a `build_framed` failure and returns `Ok(())`, then `:634` prints `Generated 0 messages` and exits 0. So the generator can emit zero files and this test passes. Everything CI's fixture coverage depends on rests on this one assertion. Fix: assert the expected files exist - `output.join("000_Produce_v3.bin").is_file()`, `output.join("001_Fetch_v4.bin").is_file()` - and make `cmd_generate` return an error when a *requested* `--api-key` produced nothing. Tradeoff: hard-failing on any skip would break bare `generate`, since the full 4.1 registry includes messages `kafka-protocol` 0.17 cannot build - so gate the error on "zero files for a requested key", not "any skip". Minor: the `remove_dir_all` at `:46` does not run when an assert fails, so each failure leaks a temp dir. ########## gateways/kafka/docs/TEST_SUITE.md: ########## @@ -0,0 +1,59 @@ +# Kafka gateway — automated regression test suite + +Regression tests live under [`tests/`](../tests/). Run from the workspace root: + +```bash +cargo test -p iggy-gateway-kafka +``` + +## 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. Every fixture-dependent +suite goes through `tests/common/fixtures.rs::load_fixture_body_or_skip`, which skips with a +regeneration hint when a fixture is missing, and panics instead when `KAFKA_FIXTURES_REQUIRED=1` +is set (CI sets this) so a broken generation step can't leave a suite green with zero assertions. + +--- + +## Test files + +An exact per-file test count and a full test-name-to-scenario matrix used to live here; both +drifted out of sync with the actual suites more than once as tests were added and consolidated. +Rather than re-derive a snapshot that will drift again, this only lists what each file is for — +`cargo test -p iggy-gateway-kafka -- --list` gives the exact current test names. + +| File | Suite focus | Depends on fixtures | +| ------ | ------------- | --------------------- | +| [`codec_tests.rs`](../tests/codec_tests.rs) | Primitive encode/decode round-trips, varint, compact strings, tagged fields | No | Review Comment: **Three catalogued suites do not exist, and one that does is missing.** Broken relative links: this line -> `../tests/codec_tests.rs`, `:35` -> `../tests/decode_safety_tests.rs`, `:40` -> `../tests/decode_validation_tests.rs`. None of the three files is in this PR. `tests/fixtures_canary_tests.rs` exists and is absent from the table. The "Depends on fixtures: Yes" row points at the nonexistent `decode_validation_tests.rs`, while the three suites that actually depend on fixtures - `api_handler_tests`, `version_firewall_tests`, `server_e2e_tests` - are marked "Partial". Same dangling names elsewhere: `README.md:35`, `MANUAL_TESTING.md:56` and `:258`, `TEST_SUITE.md:11`/`:55`/`:56`, `SCOPE.md:18`. And dangling source paths: `SCOPE.md:15`/`:16` (`src/protocol/requests.rs`), `:80`/`:102` (`codec.rs`) - both deleted in the `kafka-protocol` refactor - plus `kafka_api_keys_reference.md:276` (`encode_error_only_response`, also deleted). Worth noting `:27-30` of this file argues that the previous exact matrix "drifted out of sync with the actual suites more than once" and that this table avoids that. It has drifted. Fix: regenerate the table from `ls tests/*.rs`. Consider a CI check that every path in this table resolves - it has now drifted twice. ########## gateways/kafka/tests/common/tcp.rs: ########## @@ -0,0 +1,251 @@ +// 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. + +//! TCP round-trip helpers - compiled into each integration test binary via `#[path]`. +//! +//! Callers must also declare `#[path = "common/codec.rs"] mod codec;` at their own crate root - +//! this file borrows that module via `super::codec` rather than redeclaring it, since `rustc` +//! rejects loading the same file as two distinct modules in one crate. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::time::Duration; + +use bytes::{BufMut, Bytes, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time; + +use iggy_gateway_kafka::protocol::header::{request_header_version, response_header_version}; + +use super::codec::{self, Decoder}; + +/// Build a complete length-prefixed Kafka request frame (header + body). +pub fn build_request_frame( + api_key: i16, + api_version: i16, + correlation_id: i32, + client_id: Option<&str>, + body: &[u8], +) -> Bytes { + let hdr_ver = request_header_version(api_key, api_version); + let mut enc = codec::Encoder::with_capacity(64 + body.len()); + enc.write_i16(api_key); + enc.write_i16(api_version); + enc.write_i32(correlation_id); + // client_id is the legacy NULLABLE_STRING at every header version, even v2 - only the + // trailing tagged-fields section is new for the "flexible" header. Kafka's RequestHeader + // schema never made client_id itself a compact string. + enc.write_nullable_string(client_id) + .expect("test client_id fits i16"); + if hdr_ver >= 2 { + enc.write_empty_tagged_fields(); + } + enc.write_bytes(body); + + let payload = enc.freeze(); + let payload_len = i32::try_from(payload.len()).expect("test payload fits i32"); + let mut frame = BytesMut::with_capacity(4 + payload.len()); + frame.put_i32(payload_len); + frame.extend_from_slice(&payload); + frame.freeze() +} + +/// Parse correlation id and response body from a raw response payload (no length prefix). +pub fn parse_response_payload(api_key: i16, api_version: i16, payload: Bytes) -> (i32, Bytes) { + let resp_hdr_ver = response_header_version(api_key, api_version); + let mut d = Decoder::new(payload); + let correlation_id = d.read_i32().expect("correlation_id"); + if resp_hdr_ver >= 1 { + d.read_tagged_fields().expect("response tagged fields"); + } + let body = d.read_bytes(d.remaining()).expect("response body"); + (correlation_id, body) +} + +/// Generous ceiling for the "default" response read. A server regression that drops a +/// response then becomes a bounded test failure instead of an indefinite hang. +const DEFAULT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Read one length-prefixed response frame from the stream. +/// +/// Bounded by [`DEFAULT_RESPONSE_TIMEOUT`] so a dropped response fails fast instead of hanging. +pub async fn read_response_frame(stream: &mut TcpStream, max_size: usize) -> Bytes { + time::timeout( + DEFAULT_RESPONSE_TIMEOUT, + read_response_frame_raw(stream, max_size), + ) + .await + .unwrap_or_else(|_| { + panic!( + "no response frame within {DEFAULT_RESPONSE_TIMEOUT:?} (server dropped the response?)" + ) + }) +} + +async fn read_response_frame_raw(stream: &mut TcpStream, max_size: usize) -> Bytes { + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .await + .expect("response length prefix"); + let frame_len_i32 = i32::from_be_bytes(len_buf); + assert!(frame_len_i32 > 0, "response frame length must be positive"); + let frame_len = usize::try_from(frame_len_i32).expect("positive i32 frame length fits usize"); + assert!( + frame_len <= max_size, + "response frame too large: {frame_len}" + ); + let mut buf = vec![0u8; frame_len]; + stream.read_exact(&mut buf).await.expect("response body"); + Bytes::from(buf) +} + +/// Minimal Produce v3 body: nullable `transactional_id`, acks, timeout, empty topics array. +pub fn build_produce_v3_body(acks: i16, topics_count: i32) -> Bytes { + let mut body = BytesMut::new(); + body.put_i16(-1); // null transactional_id + body.put_i16(acks); + body.put_i32(1_000); // timeout_ms + body.put_i32(topics_count); + body.freeze() +} + +/// Minimal Produce v0-v2 body (no `transactional_id`): acks, timeout, empty topics array. +pub fn build_produce_v2_body(acks: i16, topics_count: i32) -> Bytes { + let mut body = BytesMut::new(); + body.put_i16(acks); + body.put_i32(1_000); // timeout_ms + body.put_i32(topics_count); + body.freeze() +} + +/// Minimal flexible Produce body (v9+): null compact `transactional_id`, acks, timeout, +/// compact topics array, empty tagged fields. +pub fn build_produce_flexible_body(acks: i16, topics_count: u32) -> Bytes { + let mut body = BytesMut::new(); + body.put_u8(0); // null transactional_id (compact string, varint 0) + body.put_i16(acks); + body.put_i32(1_000); // timeout_ms + body.put_u8(u8::try_from(topics_count + 1).expect("small topic count")); // compact array len + body.put_u8(0); // empty tagged fields + body.freeze() +} + +/// `ListOffsets` v0 request body for topic "t", partition 0. +pub fn build_list_offsets_v0_request_with_topic_t() -> Bytes { + let mut body = BytesMut::new(); + body.put_i32(-1); // replica_id + body.put_i32(1); // topics array length + body.put_i16(1); // topic name length + body.put_u8(b't'); + body.put_i32(1); // partitions array length + body.put_i32(0); // partition index + body.put_i64(-1); // timestamp + body.put_i32(1); // max_num_offsets + body.freeze() +} + +/// Legacy Metadata request body listing topic names (non-flexible, v0–v3 fields only). +/// +/// Prefer version-aware wire helpers when targeting Metadata v4+, which also require +/// `allow_auto_topic_creation` (and later authorized-ops flags). +pub fn build_metadata_legacy_request(topic_names: &[&str]) -> Bytes { + let mut body = BytesMut::new(); + body.put_i32(i32::try_from(topic_names.len()).expect("topic name count fits i32")); + for name in topic_names { + let name_bytes = name.as_bytes(); + let len = i16::try_from(name_bytes.len()).expect("topic name fits i16"); + body.put_i16(len); + body.extend_from_slice(name_bytes); + } + body.freeze() +} + +/// Read one length-prefixed response frame, returning `None` on timeout. +pub async fn read_response_frame_with_timeout( + stream: &mut TcpStream, + max_size: usize, + timeout: Duration, +) -> Option<Bytes> { + time::timeout(timeout, read_response_frame_raw(stream, max_size)) + .await + .ok() +} + +/// Concatenate multiple length-prefixed frames (for pipelining tests). +pub fn concat_frames(frames: &[Bytes]) -> Bytes { + let total: usize = frames.iter().map(Bytes::len).sum(); + let mut out = BytesMut::with_capacity(total); + for frame in frames { + out.extend_from_slice(frame); + } + out.freeze() +} + +/// Outcome of a single-byte read, used by connection-close assertions. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ByteRead { + /// A byte was read from the stream. + Byte(u8), + /// The server closed the connection (clean EOF or reset). + Closed, + /// No byte arrived within the timeout; the connection is still open. + Timeout, +} + +/// Read one byte, distinguishing a server-side close from an idle timeout so that +/// "server must close the connection" tests can assert [`ByteRead::Closed`] explicitly +/// instead of passing on a mere stall. +pub async fn read_byte_with_timeout(stream: &mut TcpStream, timeout: Duration) -> ByteRead { + let mut buf = [0u8; 1]; + match time::timeout(timeout, stream.read(&mut buf)).await { + // 0 bytes = clean EOF; a reset / broken pipe is still a closed connection here. + Ok(Ok(0) | Err(_)) => ByteRead::Closed, + Ok(Ok(_)) => ByteRead::Byte(buf[0]), + Err(_) => ByteRead::Timeout, + } +} + +/// Scan a response body for a big-endian `i16` error code at any 2-byte-aligned +/// offset. Used by corrupt-body / unsupported-version tests that assert an error +/// code is present somewhere in the response without fully decoding its shape. +pub fn scan_for_error_code(body: &Bytes, code: i16) -> bool { Review Comment: **Unaligned scan, a doc comment that contradicts the code, and an `||` that defeats both callers.** The doc says the code is matched "at any 2-byte-aligned offset". `body.windows(2)` steps by one byte, so it also matches unaligned pairs - the low half of any `i32`/`i64` field that happens to equal the code, a partition index, a throttle time, or padding. The bigger problem is at the call sites. `version_firewall_tests.rs:720-730` and `:733-741` do: ```rust assert!(scan_for_error_code(&resp, ERROR_INVALID_REQUEST) || scan_for_error_code(&resp, ERROR_UNSUPPORTED_VERSION)); ``` Both tests are named `..._returns_invalid_request_error`. Change `handle_versioned_request`'s decode-error arm (`src/protocol/api.rs:360`) to `encode_err(api_version, ERROR_UNSUPPORTED_VERSION)` and both still pass. `listener_robustness_tests.rs:553-559` has the same helper dependency. The shared path is genuinely covered elsewhere - `api_handler_tests.rs:216` and `:234` decode structurally and assert `== 42` - so these two tests claim a precision they do not have rather than hiding a live bug. Fix: correct the doc to say "any byte offset", drop the `||`, and replace the two `version_firewall_tests` callers with structural decodes at v1/v2 - the non-flexible shapes the precise tests do not cover. Tradeoff: structural decodes at v1/v2 hard-code those field layouts in the test and break if `kafka-protocol`'s schema for those versions changes. That is the right direction: a loud break beats a silent pass. ########## .github/actions/rust/pre-merge/action.yml: ########## @@ -279,6 +279,32 @@ runs: compile_duration=$((compile_end - compile_start)) echo "::notice::Tests compiled in ${compile_duration}s ($(date -ud @${compile_duration} +'%M:%S'))" + # decode_validation_tests need gitignored wire fixtures. Generate when Review Comment: **This comment names a test suite that does not exist.** `decode_validation_tests` is not in this PR. The fixture consumers are `api_handler_tests.rs`, `server_e2e_tests.rs` and `version_firewall_tests.rs`. Same stale name at `gateways/kafka/scripts/ci-wire-fixtures.sh:26`. It matters more here than in a doc: the `export KAFKA_FIXTURES_REQUIRED=1` below turns any gap in this block's scope into a hard CI failure rather than a skip, so a future maintainer needs to be able to find out what the block protects. Grepping the named suite returns nothing, which makes the whole block read as safe to delete. Fix: name the three real suites. ########## gateways/kafka/docs/kafka_api_keys_reference.md: ########## @@ -0,0 +1,304 @@ +# Kafka Protocol API Key Reference — Kafka 4.0.0 + +> **Source**: [`ApiKeys.java` @ Kafka 4.0.0](https://github.com/apache/kafka/blob/4.0.0/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java) +> and the canonical [protocol message schemas](https://github.com/apache/kafka/tree/4.0.0/clients/src/main/resources/common/message). +> +> Generated for: **Iggy Kafka Bridge Gateway** — `gateways/kafka/` +> Branch: `feat(gateways)/kafka_to_iggy_listener` + +--- + +## Legend + +| Symbol | Meaning | +| -------- | --------- | +| 🔴 Bridge | Core data path — must be fully implemented and forwarded to Iggy | +| 🟠 Required Stub | Client state-machine API — must return a well-formed response or clients will stall/crash | +| 🟡 Optional Stub | Admin/observability — can safely return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` | +| ❌ Reject | Internal broker / KRaft only — return `INVALID_REQUEST` with a well-formed frame; **do not close the connection** | + +> **Header.rs ✓** = The API key is already present in `request_header_version()` / `response_header_version()` with the correct flexible-encoding threshold. Review Comment: **The `Header.rs ✓` column, and the "Current Implementation Gaps" section, describe an implementation that no longer exists.** This line defines the column as "already present in `request_header_version()` / `response_header_version()` with the correct flexible-encoding threshold". After the `kafka-protocol` migration, `src/protocol/header.rs:29-40` has no table - it is two `ApiKey::try_from(...).map_or(...)` wrappers. So every mark in all 14 tables is meaningless. Specific claims that no longer hold: - `:69` "`header.rs` already handles this correctly via the `api_key == 18` guard" - no such guard exists. - `:287-293` "still fall through to the `_ => i16::MAX` (non-flexible) arm" - there is no match and no arm; the fallback is `map_or(1, ...)`. - `:276` "falls through to `encode_error_only_response` (2-byte error frame)" - that function was deleted; unknown keys now `Close`. - `:227-230` marks keys 4-7 as present, while `tests/header_tests.rs:35-38` documents that `ApiKey::try_from` *fails* for exactly those keys in kafka-protocol 0.17, so the wrapper returns header v1. The reference and the test state opposite facts. Internal arithmetic also drifted: `:253` counts "6" and lists seven names; `:257` says "83 ... Key IDs 0-88 with gap at 73" (0-88 minus one gap is 88). And `:7` records "Branch: `feat(gateways)/kafka_to_iggy_listener`" - a shipped doc naming the PR branch, which cuts against CLAUDE.md principle 8. Fix: delete the `Header.rs ✓` column and the "Current Implementation Gaps in `api.rs`" section - both were snapshots of deleted code - and replace with one sentence: header framing comes from `kafka_protocol::messages::ApiKey`, so any key that crate knows is framed correctly and any key it does not gets header v1. Tradeoff: the doc loses its "what's left to do" value, so the remaining scope should move into `SCOPE.md`'s TODO section, which already has that job. ########## 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 (or iggy-gateway-kafka subdir) +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` | Review Comment: **The "must pass before check-in" procedure expects the wrong result in six of nine Category A steps, and its commands hit the wrong port.** This row (A5, Produce v3) expects `ec=0`; `src/protocol/responses.rs:91` returns `ERROR_NOT_LEADER_OR_FOLLOWER` (6). `:73` (A8, CreateTopics v2) expects "Decode + stub ack, `ec=0`"; `responses.rs:297` returns `ERROR_NOT_CONTROLLER` (41) - contradicting `README.md:5` and `SCOPE.md:7` inside this same PR. `:117-118` (D1/D2, Produce v8/v9) expect `ec=0`. `:71-72` (A6/A7) expect `ec=0`, which holds only for the top-level field; per-partition is 6. Beyond Category A: - `:92` (B2) expects Metadata out-of-range to "show `error_code=35`"; `src/protocol/api.rs:317-327` closes the connection and sends nothing. - `:93` (B3) expects a "version-aware response with `error_code=35`"; above-max closes (`api.rs:392-400`) and below-min also closes today for all four APIs, which `api.rs:376-382` documents accurately. - `:134` (E3) expects name `unknown-topic`; `api.rs:464-471` echoes the requested name, and `tests/server_e2e_tests.rs:296-298` asserts exactly that it must not substitute `unknown-topic`. - `:164` (H1) expects a truncated Produce body to give `error_code=42` "or connection error"; `api.rs:237-247` returns `NoResponse` - the connection stays open and nothing comes back. `api.rs:242-244` documents this as a deliberate change from the pre-refactor decoder; the doc was not updated with it. - `:176`/`:179`/`:183`/`:193` describe error-code paths and a `header.rs` "lookup table" that no longer exist. - `:43` generates fixtures for keys 3 and 18 that no test consumes (`api_handler_tests.rs:303`/`:317` `continue` past them) and that `ci-wire-fixtures.sh:27` deliberately excludes. Every `send`/`verify` line also omits `--host`, and the tool defaults to `127.0.0.1:9092` (`tools/kafka-tool/src/main.rs:77`, `:90`) while the gateway binds 9093 (`api.rs:42`). So A6, A7, A8 and C1-C4 connection-refuse before any assertion runs. Fix: rewrite the expected-result column against the code, add `--host 127.0.0.1:9093` to every command, and drop the `___/103` count at `:244` (see the README comment). ########## gateways/kafka/README.md: ########## @@ -0,0 +1,63 @@ +# Kafka gateway (`iggy-gateway-kafka`) + +Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests, validates scoped API keys and versions, and returns stub responses. + +> **Stub warning:** no API persists or reads real data yet. Produce, Fetch, and ListOffsets return retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere instead of trusting a fake success. CreateTopics does **not** create topics; valid requests return `NOT_CONTROLLER` (41). Metadata still reports requested topics as unknown. Persistence lands with the Iggy bridge (see [docs/SCOPE.md](docs/SCOPE.md)). + +## Run + +```bash +cargo run -p iggy-gateway-kafka +``` + +Default bind: `127.0.0.1:9093`. Environment variables: + +| Variable | Default | Description | +| --- | --- | --- | +| `IGGY_KAFKA_BIND_ADDR` | `127.0.0.1:9093` | TCP address to listen on | +| `IGGY_KAFKA_ADVERTISED_HOST` | bind IP | Hostname/IP clients use to reach this broker (required when binding to `0.0.0.0`/`::`) | +| `IGGY_KAFKA_ADVERTISED_PORT` | bind port | Port advertised in Metadata responses | +| `IGGY_KAFKA_MAX_CONNECTIONS` | `1024` | Maximum concurrent connections before new ones are rejected | +| `IGGY_KAFKA_MAX_FRAME_SIZE` | `8388608` | Maximum accepted request frame size in bytes | +| `IGGY_KAFKA_IDLE_TIMEOUT_SECS` | `600` | Seconds a connection may sit idle before the next frame's length prefix arrives | +| `IGGY_KAFKA_READ_TIMEOUT_SECS` | `15` | Seconds allowed to read a frame body once its length prefix arrives | +| `IGGY_KAFKA_WRITE_TIMEOUT_SECS` | `10` | Seconds allowed to write a response frame | +| `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown waits for in-flight connections before abandoning them | + +## Test + +```bash +cargo test -p iggy-gateway-kafka +``` + +259 regression tests across 12 suites — see [docs/TEST_SUITE.md](docs/TEST_SUITE.md) for the full catalog. Review Comment: **Three different hard test counts ship in this PR, and none of them is correct.** - here: "259 regression tests across 12 suites" - `docs/MANUAL_TESTING.md:244`: "`cargo test -p iggy-gateway-kafka` - ___/103 passed" - PR description: "103 regression tests across 12 suites" Actual: 10 suite files under `gateways/kafka/tests/` holding 157 `#[test]`/`#[tokio::test]` functions, plus 25 in-crate unit tests (`src/main.rs` 3, `src/protocol/api.rs` 6, `src/server.rs` 16) = **182** for `cargo test -p iggy-gateway-kafka`. 259 looks transplanted: it is exactly `sum(max - min + 1)` over the tool's `API_REGISTRY`, i.e. the versioned-message count printed by `tools/kafka-tool/src/main.rs:588-597`. `MANUAL_TESTING.md:244` is the practical problem - it asks a tester to fill in `___/103` on a checklist that will read 182, so the checklist cannot be completed truthfully. Fix: drop the counts. `docs/TEST_SUITE.md:27-30` already makes this argument for the per-file matrix ("both drifted out of sync ... more than once"); the same rule applies to the totals. -- 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]
