slbotbm commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3740510915
########## gateways/kafka/src/protocol/codec.rs: ########## @@ -0,0 +1,408 @@ +// 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. + +//! Low-level Kafka primitive encoders/decoders (ported wire codec). + +#![allow(clippy::missing_const_for_fn, clippy::bool_to_int_with_if)] +#![allow( + clippy::missing_errors_doc, + clippy::cast_sign_loss, + clippy::must_use_candidate, + clippy::missing_panics_doc, + clippy::cast_possible_truncation, + clippy::cast_lossless +)] + +use bytes::{Buf, BufMut, Bytes, BytesMut}; + +use crate::error::{KafkaProtocolError, Result}; + +/// Upper bound for Kafka array/collection element counts decoded from the wire. +/// Matches typical broker limits and prevents OOM from adversarial length prefixes. +pub const MAX_COLLECTION_LEN: usize = 65_536; + +/// Initial `Vec::with_capacity` hint for wire-decoded array counts. +/// +/// A count up to `MAX_COLLECTION_LEN` is still validated as a count, but pre-reserving on it +/// directly lets a few-byte frame declaring a huge count force a multi-megabyte allocation +/// before any element bytes are checked present. Clamp the reservation; genuine large arrays +/// still grow correctly via `Vec::push`'s amortized doubling once real elements are decoded. +pub const PREALLOC_HINT: usize = 128; + +pub struct Decoder { + bytes: Bytes, +} + +impl Decoder { + pub fn new(bytes: Bytes) -> Self { + Self { bytes } + } + + pub fn remaining(&self) -> usize { + self.bytes.remaining() + } + + pub fn read_u8(&mut self) -> Result<u8> { + self.ensure(1)?; + Ok(self.bytes.get_u8()) + } + + pub fn read_i8(&mut self) -> Result<i8> { + self.ensure(1)?; + Ok(self.bytes.get_i8()) + } + + pub fn read_i16(&mut self) -> Result<i16> { + self.ensure(2)?; + Ok(self.bytes.get_i16()) + } + + pub fn read_i32(&mut self) -> Result<i32> { + self.ensure(4)?; + Ok(self.bytes.get_i32()) + } + + pub fn read_i64(&mut self) -> Result<i64> { + self.ensure(8)?; + Ok(self.bytes.get_i64()) + } + + pub fn read_bool(&mut self) -> Result<bool> { + Ok(self.read_i8()? != 0) + } + + /// Unsigned varint (Kafka uses this for compact array lengths and tagged-field counts). + /// Value is encoded with 7 bits per byte, LSB first; the high bit of each byte signals + /// that more bytes follow. + pub fn read_varint(&mut self) -> Result<u64> { + let mut result: u64 = 0; + let mut shift = 0u32; + loop { + let byte = self.read_u8()?; + if shift == 63 && byte & 0x7E != 0 { + return Err(KafkaProtocolError::InvalidVarint); + } + result |= ((byte & 0x7F) as u64) << shift; + if byte & 0x80 == 0 { + return Ok(result); + } + shift += 7; + if shift >= 64 { + return Err(KafkaProtocolError::InvalidVarint); + } + } + } + + /// Legacy array length: signed i32 count (must be non-negative). + pub fn read_i32_array_count(&mut self) -> Result<usize> { + let n = self.read_i32()?; + if n < 0 { + return Err(KafkaProtocolError::InvalidArrayLength(n)); + } + // Safe: n is in [0, i32::MAX]; i32::MAX (2_147_483_647) fits in usize + // on all 32-bit and 64-bit platforms this crate targets. + let count = n as usize; + if count > MAX_COLLECTION_LEN { + return Err(KafkaProtocolError::CollectionTooLarge { + count, + max: MAX_COLLECTION_LEN, + }); + } + Ok(count) + } + + /// Legacy nullable array length: signed i32 count, where -1 is the spec-defined null + /// (`all items`) sentinel - used by Metadata's `topics` field to mean "all topics" - treated + /// as empty (0 elements) rather than an error, mirroring how `read_compact_array_count` + /// treats a null compact array. + pub fn read_i32_array_count_nullable(&mut self) -> Result<usize> { + let n = self.read_i32()?; + if n == -1 { + return Ok(0); + } + if n < 0 { + return Err(KafkaProtocolError::InvalidArrayLength(n)); + } + // Safe: n is in [0, i32::MAX]; i32::MAX (2_147_483_647) fits in usize + // on all 32-bit and 64-bit platforms this crate targets. + let count = n as usize; + if count > MAX_COLLECTION_LEN { + return Err(KafkaProtocolError::CollectionTooLarge { + count, + max: MAX_COLLECTION_LEN, + }); + } + Ok(count) + } + + /// Compact array length: unsigned varint holding `element_count + 1`. + /// Per the Kafka spec, varint=0 encodes a null (absent) array; treat as empty (0 elements) + /// so optional fields like `forgotten_topics` are skipped rather than rejected. + pub fn read_compact_array_count(&mut self) -> Result<usize> { + let n = self.read_varint()?; + if n == 0 { + return Ok(0); + } + let count = usize::try_from(n - 1).map_err(|_| KafkaProtocolError::CollectionTooLarge { + count: MAX_COLLECTION_LEN + 1, + max: MAX_COLLECTION_LEN, + })?; + if count > MAX_COLLECTION_LEN { + return Err(KafkaProtocolError::CollectionTooLarge { + count, + max: MAX_COLLECTION_LEN, + }); + } + Ok(count) + } Review Comment: This maps compact-array prefix `0` to an empty array unconditionally. Prefix `0` is Kafka's null encoding and is invalid for non-nullable arrays, but the same helper is used for required fields throughout `requests.rs`, including Produce topic data, Fetch topics and partitions, ListOffsets topics, and CreateTopics topics. For example, a flexible Produce request with null `topic_data` is accepted as an empty request, and a CreateTopics v5 request with null `topics` produces an empty response instead of `INVALID_REQUEST`. Split the decoder into nullable and non-nullable compact-array helpers and reject `0` for required arrays. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,576 @@ +// 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::Bytes; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::codec::{Decoder, Encoder, PREALLOC_HINT}; +use crate::protocol::requests::{ + ProduceDecodeResult, decode_create_topics_request, decode_fetch_request, + decode_list_offsets_request, decode_produce_request, +}; +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_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; + +const MAX_SUPPORTED_METADATA_VERSION: i16 = 9; + +/// Sentinel for `topic_authorized_operations` / `cluster_authorized_operations` when ACLs are not supported. +const AUTHORIZED_OPS_UNKNOWN: i32 = i32::MIN; + +/// Result of handling one Kafka request body. +#[derive(Debug)] +pub enum HandleOutcome { + /// Write this response body (with a response header). + Respond(Bytes), + /// Write this response body (with a response header), then close the TCP connection. + RespondAndClose(Bytes), + /// Produce with `acks=0`: write nothing, keep the connection open. + NoResponse, + /// Client cannot parse an error at this request wire version; close the TCP connection. + Close, +} + +impl HandleOutcome { + /// Collapse to `Some(body)` for a normal response, or `None` for [`HandleOutcome::NoResponse`]. + /// + /// # Panics + /// + /// Panics on [`HandleOutcome::Close`] - match on `Close` explicitly, or use + /// [`Self::expect_response`] in tests that require a body. + #[must_use] + pub fn into_optional_response(self) -> Option<Bytes> { + match self { + Self::Respond(body) | Self::RespondAndClose(body) => Some(body), + Self::NoResponse => None, + Self::Close => panic!("HandleOutcome::Close has no response body"), + } + } + + /// 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) | Self::RespondAndClose(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) +} + +/// 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 `acks`, 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 { + match decode_produce_request(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. + ProduceDecodeResult::Ok(req) if req.acks == 0 => HandleOutcome::NoResponse, + ProduceDecodeResult::Ok(req) => { + if !is_supported_version(API_KEY_PRODUCE, api_version) { + return HandleOutcome::Respond(encode_produce_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )); + } + HandleOutcome::Respond(encode_produce_response(api_version, &req)) + } + ProduceDecodeResult::Err { + acks: Some(0), + error, + } => { + tracing::warn!( + "Failed to decode Produce request with acks=0 (no response): {:?}", + error + ); + HandleOutcome::NoResponse + } + ProduceDecodeResult::Err { error, .. } => { + tracing::warn!("Failed to decode Produce request: {:?}", error); + let code = if is_supported_version(API_KEY_PRODUCE, api_version) { + ERROR_INVALID_REQUEST + } else { + ERROR_UNSUPPORTED_VERSION + }; + HandleOutcome::Respond(encode_produce_error_response(api_version, code)) + } + } +} + +fn handle_other_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> HandleOutcome { + match api_key { + API_KEY_API_VERSIONS => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_api_versions_response(api_version, ERROR_NONE)) + } else { + // KIP-511: reply with v0 when the requested version is not understood. + HandleOutcome::Respond(encode_api_versions_response(0, ERROR_UNSUPPORTED_VERSION)) + } + } + API_KEY_METADATA => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_metadata_response( + api_version, + api_version, + body, + broker, + ERROR_NONE, + )) + } else { + // Clamping the response to MAX_SUPPORTED_METADATA_VERSION leaves a body the + // client parses at its own (unsupported) version, so UNSUPPORTED_VERSION never + // survives. Clients that skip ApiVersions get a naked close instead. + tracing::warn!( + api_version, + max_supported = MAX_SUPPORTED_METADATA_VERSION, + "Metadata version unsupported; closing connection" + ); + HandleOutcome::Close + } + } + API_KEY_FETCH => { + if is_supported_version(api_key, api_version) { + match decode_fetch_request(api_version, body) { + Ok(req) => HandleOutcome::Respond(encode_fetch_response(api_version, &req)), + Err(e) => { + tracing::warn!("Failed to decode Fetch request: {:?}", e); + HandleOutcome::Respond(encode_fetch_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_fetch_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + API_KEY_LIST_OFFSETS => { + if is_supported_version(api_key, api_version) { + match decode_list_offsets_request(api_version, body) { + Ok(req) => { + HandleOutcome::Respond(encode_list_offsets_response(api_version, &req)) + } + Err(e) => { + tracing::warn!("Failed to decode ListOffsets request: {:?}", e); + HandleOutcome::Respond(encode_list_offsets_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_list_offsets_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + API_KEY_CREATE_TOPICS => { + if is_supported_version(api_key, api_version) { + match decode_create_topics_request(api_version, body) { + Ok(req) => { + HandleOutcome::Respond(encode_create_topics_response(api_version, &req)) + } + Err(e) => { + tracing::warn!("Failed to decode CreateTopics request: {:?}", e); + HandleOutcome::Respond(encode_create_topics_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_create_topics_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + _ => HandleOutcome::RespondAndClose(encode_error_only_response(ERROR_UNSUPPORTED_VERSION)), + } +} + +#[must_use] +pub fn is_supported_version(api_key: i16, api_version: i16) -> bool { + SUPPORTED_RANGES + .iter() + .find(|r| r.api_key == api_key) + .is_some_and(|r| api_version >= r.min_version && api_version <= r.max_version) +} + +/// Min version advertised in `ApiVersions` (may differ from the firewall min). +/// +/// Produce must advertise min=0 per KAFKA-18659 / `PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` +/// even though this gateway only accepts Produce v3+. +#[must_use] +pub const fn advertised_min_version(api_key: i16, firewall_min: i16) -> i16 { + if api_key == API_KEY_PRODUCE { + 0 + } else { + firewall_min + } +} + +fn encode_api_versions_response(api_version: i16, error_code: i16) -> Bytes { + let flexible = api_version >= 3; + let ranges = SUPPORTED_RANGES; + let mut e = Encoder::with_capacity(128); + + e.write_i16(error_code); + + if flexible { + e.write_varint((ranges.len() + 1) as u64); + for r in ranges { + e.write_i16(r.api_key); + e.write_i16(advertised_min_version(r.api_key, r.min_version)); + e.write_i16(r.max_version); + e.write_empty_tagged_fields(); + } + } else { + e.write_i32(i32::try_from(ranges.len()).expect("supported range table is small")); + for r in ranges { + e.write_i16(r.api_key); + e.write_i16(advertised_min_version(r.api_key, r.min_version)); + e.write_i16(r.max_version); + } + } + + if api_version >= 1 { + e.write_i32(0); + } + + if flexible { + e.write_empty_tagged_fields(); + } + + e.freeze() +} + +fn encode_metadata_response( + response_version: i16, + decode_version: i16, + body: Bytes, + broker: &BrokerAdvertise, + top_level_error_code: i16, +) -> Bytes { + let flexible = response_version >= 9; + // Empty body = all-topics request; 0 topics is correct for this stub. + // Non-empty body that fails to decode = malformed request; return 0 topics. + // Kafka Metadata response has no top-level error code field: errors are per-topic only. + // 0 topics is spec-correct and unambiguous for a decode failure. + let (topics, effective_error) = if body.is_empty() { + (Vec::new(), top_level_error_code) + } else { Review Comment: This considers an empty body an all-topics request, although every Metadata version requires at least an array count. Additionally, `gateways/kafka/src/protocol/api.rs:493` only decodes topics and ignores required fields such as `AllowAutoTopicCreation`, authorization flags, and top-level tagged fields. For example, a v9 body containing only `0x00` is accepted despite missing the rest of the schema. The complete version-specific body should be decoded and validated. For versions without a parseable top-level error, malformed requests should close the connection rather than receive a successful zero-topic response. ########## gateways/kafka/src/protocol/requests.rs: ########## @@ -0,0 +1,526 @@ +// 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 request decoders for critical API keys + +#![allow(clippy::too_many_lines, clippy::doc_markdown)] +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::codec::{Decoder, PREALLOC_HINT}; +use bytes::Bytes; + +/// Produce Request (API Key 0) +#[derive(Debug, Clone)] +pub struct ProduceRequest { + pub transactional_id: Option<String>, + pub acks: i16, + pub timeout_ms: i32, + pub topics: Vec<ProduceTopicData>, +} + +#[derive(Debug, Clone)] +pub struct ProduceTopicData { + pub topic: String, + pub partitions: Vec<ProducePartitionData>, +} + +#[derive(Debug, Clone)] +pub struct ProducePartitionData { + pub partition: i32, + pub records: Option<Bytes>, // Raw RecordBatch bytes +} + +/// Outcome of decoding a Produce request body. +/// +/// Carries `acks` on failure so callers can honor fire-and-forget (`acks=0`) silence +/// even when later fields are malformed. +#[derive(Debug)] +pub enum ProduceDecodeResult { + Ok(ProduceRequest), + Err { + acks: Option<i16>, + error: KafkaProtocolError, + }, +} + +impl ProduceDecodeResult { + /// Collapse to `Result` for tests and callers that only need a successful `ProduceRequest`. + /// + /// # Errors + /// + /// Returns an error if the byte stream is malformed or if the API version is unsupported. + pub fn into_request(self) -> Result<ProduceRequest> { + match self { + Self::Ok(req) => Ok(req), + Self::Err { error, .. } => Err(error), + } + } +} + +macro_rules! produce_decode { + ($acks:expr, $expr:expr) => { + match $expr { + Ok(value) => value, + Err(error) => { + return ProduceDecodeResult::Err { acks: $acks, error }; + } + } + }; +} + +pub fn decode_produce_request(version: i16, body: Bytes) -> ProduceDecodeResult { + let mut d = Decoder::new(body); + let flexible = version >= 9; + + // transactional_id (v3+) + let transactional_id = if version >= 3 { + produce_decode!( + None, + if flexible { + d.read_compact_nullable_string() + } else { + d.read_nullable_string() + } + ) + } else { + None + }; + + let acks = produce_decode!(None, d.read_i16()); + let acks_read = Some(acks); + + let timeout_ms = produce_decode!(acks_read, d.read_i32()); + + // topics array + let topics_count = produce_decode!( + acks_read, + if flexible { + d.read_compact_array_count() + } else { + d.read_i32_array_count() + } + ); + + let mut topics = Vec::with_capacity(topics_count.min(PREALLOC_HINT)); + for _ in 0..topics_count { + let topic = produce_decode!( + acks_read, + if flexible { + d.read_compact_nullable_string() + } else { + d.read_nullable_string() + } + .and_then(|name| name.ok_or(KafkaProtocolError::NullTopicName)) + ); + + let partitions_count = produce_decode!( + acks_read, + if flexible { + d.read_compact_array_count() + } else { + d.read_i32_array_count() + } + ); + + let mut partitions = Vec::with_capacity(partitions_count.min(PREALLOC_HINT)); + for _ in 0..partitions_count { + let partition = produce_decode!(acks_read, d.read_i32()); + let records = produce_decode!( + acks_read, + if flexible { + d.read_compact_nullable_bytes() + } else { + d.read_nullable_bytes() + } + ); + partitions.push(ProducePartitionData { partition, records }); + if flexible { + produce_decode!(acks_read, d.read_tagged_fields()); + } + } + + topics.push(ProduceTopicData { topic, partitions }); + if flexible { + produce_decode!(acks_read, d.read_tagged_fields()); + } + } + + if flexible { + produce_decode!(acks_read, d.read_tagged_fields()); + } + + ProduceDecodeResult::Ok(ProduceRequest { + transactional_id, + acks, + timeout_ms, + topics, + }) +} + +/// Fetch Request (API Key 1) +#[derive(Debug, Clone)] +pub struct FetchRequest { + pub max_wait_ms: i32, + pub min_bytes: i32, + pub max_bytes: i32, + pub isolation_level: i8, + pub topics: Vec<FetchTopic>, +} + +#[derive(Debug, Clone)] +pub struct FetchTopic { + pub topic: String, + pub partitions: Vec<FetchPartition>, +} + +#[derive(Debug, Clone)] +pub struct FetchPartition { + pub partition: i32, + pub fetch_offset: i64, + pub partition_max_bytes: i32, +} +/// Decodes a raw byte stream into a `FetchRequest`. +/// +/// # Errors +/// +/// Returns an error if the byte stream is malformed or if the API version is unsupported. +pub fn decode_fetch_request(version: i16, body: Bytes) -> Result<FetchRequest> { + let mut d = Decoder::new(body); + let flexible = version >= 12; + + let _replica_id = d.read_i32()?; + let max_wait_ms = d.read_i32()?; + let min_bytes = d.read_i32()?; + + let max_bytes = if version >= 3 { + d.read_i32()? + } else { + 52_428_800 // default 50MB + }; + + let isolation_level = if version >= 4 { d.read_i8()? } else { 0 }; + + // session_id and session_epoch (v7+) - read and discard (stub path) + if version >= 7 { + d.read_i32()?; // session_id + d.read_i32()?; // session_epoch + } + + // topics array + let topics_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + + let mut topics = Vec::with_capacity(topics_count.min(PREALLOC_HINT)); + for _ in 0..topics_count { + let topic = if flexible { + d.read_compact_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + } else { + d.read_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + }; + + let partitions_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + + let mut partitions = Vec::with_capacity(partitions_count.min(PREALLOC_HINT)); + for _ in 0..partitions_count { + let partition = d.read_i32()?; + + if version >= 9 { + d.read_i32()?; // current_leader_epoch + } + + let fetch_offset = d.read_i64()?; + + if version >= 12 { + d.read_i32()?; // last_fetched_epoch + } + + if version >= 5 { + d.read_i64()?; // log_start_offset + } + + let partition_max_bytes = d.read_i32()?; + + partitions.push(FetchPartition { + partition, + fetch_offset, + partition_max_bytes, + }); + + if flexible { + d.read_tagged_fields()?; + } + } + + topics.push(FetchTopic { topic, partitions }); + if flexible { + d.read_tagged_fields()?; + } + } + + // forgotten_topics_data (v7+) - skip + if version >= 7 { + let forgotten_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + for _ in 0..forgotten_count { + if flexible { + d.read_compact_nullable_string()?; + let partitions_count = d.read_compact_array_count()?; + for _ in 0..partitions_count { + d.read_i32()?; + } + d.read_tagged_fields()?; + } else { + d.read_nullable_string()?; + let partitions_count = d.read_i32_array_count()?; + for _ in 0..partitions_count { + d.read_i32()?; + } + } + } + } + + // rack_id (v11+) + if version >= 11 { + if flexible { + d.read_compact_nullable_string()?; + } else { + d.read_nullable_string()?; + } + } + + if flexible { + d.read_tagged_fields()?; + } + + Ok(FetchRequest { + max_wait_ms, + min_bytes, + max_bytes, + isolation_level, + topics, + }) +} + +/// ListOffsets Request (API Key 2) +#[derive(Debug, Clone)] +pub struct ListOffsetsRequest { + pub isolation_level: i8, + pub topics: Vec<ListOffsetsTopic>, +} + +#[derive(Debug, Clone)] +pub struct ListOffsetsTopic { + pub topic: String, + pub partitions: Vec<ListOffsetsPartition>, +} + +#[derive(Debug, Clone)] +pub struct ListOffsetsPartition { + pub partition: i32, + pub timestamp: i64, // -2 = earliest, -1 = latest +} +/// Decodes a raw byte stream into a `ListOffsetsRequest`. +/// +/// # Errors +/// +/// Returns an error if the byte stream is malformed or if the API version is unsupported. +pub fn decode_list_offsets_request(version: i16, body: Bytes) -> Result<ListOffsetsRequest> { + let mut d = Decoder::new(body); + let flexible = version >= 6; + + let _replica_id = d.read_i32()?; + + let isolation_level = if version >= 2 { d.read_i8()? } else { 0 }; + + let topics_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + + let mut topics = Vec::with_capacity(topics_count.min(PREALLOC_HINT)); + for _ in 0..topics_count { + let topic = if flexible { + d.read_compact_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + } else { + d.read_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + }; + + let partitions_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + + let mut partitions = Vec::with_capacity(partitions_count.min(PREALLOC_HINT)); + for _ in 0..partitions_count { + let partition = d.read_i32()?; + + if version >= 4 { + d.read_i32()?; // current_leader_epoch + } + + let timestamp = d.read_i64()?; + + if version == 0 { + d.read_i32()?; // max_num_offsets (deprecated) + } + + partitions.push(ListOffsetsPartition { + partition, + timestamp, + }); + + if flexible { + d.read_tagged_fields()?; + } + } + + topics.push(ListOffsetsTopic { topic, partitions }); + if flexible { + d.read_tagged_fields()?; + } + } + + if flexible { + d.read_tagged_fields()?; + } + + Ok(ListOffsetsRequest { + isolation_level, + topics, + }) +} + +/// `CreateTopics` Request (API Key 19) +#[derive(Debug, Clone)] +pub struct CreateTopicsRequest { + pub topics: Vec<CreatableTopic>, + pub timeout_ms: i32, + pub validate_only: bool, +} + +#[derive(Debug, Clone)] +pub struct CreatableTopic { + pub name: String, + pub num_partitions: i32, + pub replication_factor: i16, +} + +/// Decodes a raw byte stream into a `CreateTopicsRequest`. +/// +/// # Errors +/// +/// Returns an error if the byte stream is malformed or if the API version is unsupported. +pub fn decode_create_topics_request(version: i16, body: Bytes) -> Result<CreateTopicsRequest> { + let mut d = Decoder::new(body); + let flexible = version >= 5; + + let topics_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + + let mut topics = Vec::with_capacity(topics_count.min(PREALLOC_HINT)); + for _ in 0..topics_count { + let name = if flexible { + d.read_compact_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + } else { + d.read_nullable_string()? + .ok_or(KafkaProtocolError::NullTopicName)? + }; + + let num_partitions = d.read_i32()?; + let replication_factor = d.read_i16()?; + + // assignments (COMPACT_ARRAY or ARRAY) - skip + let assignments_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + for _ in 0..assignments_count { + d.read_i32()?; // partition_index + let replicas_count = if flexible { + d.read_compact_array_count()? + } else { + d.read_i32_array_count()? + }; + for _ in 0..replicas_count { + d.read_i32()?; // broker_id + } + if flexible { + d.read_tagged_fields()?; + } + } Review Comment: This parses and discards manual assignments, so `CreatableTopic` does not record whether assignments were supplied. `gateways/kafka/src/protocol/responses.rs:336` then rejects `num_partitions = -1` and `replication_factor = -1` for all versions below v4. Those sentinel values are valid on v2 and v3 when a manual partition assignment is present. KIP-464 only extended their validity to requests without assignments in v4. Preserve whether assignments were provided and use it when validating the two fields. ########## gateways/kafka/tools/kafka-tool/src/main.rs: ########## @@ -0,0 +1,844 @@ +// 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 anyhow::{Context, Result}; +use bytes::{BufMut, Bytes, BytesMut}; +use clap::{Parser, Subcommand}; +use iggy_gateway_kafka::protocol::api::supported_api_ranges; +use kafka_protocol::messages::add_partitions_to_txn_request::*; +use kafka_protocol::messages::create_topics_request::*; +use kafka_protocol::messages::delete_records_request::*; +use kafka_protocol::messages::delete_topics_request::*; +use kafka_protocol::messages::describe_configs_request::*; +use kafka_protocol::messages::fetch_request::*; +use kafka_protocol::messages::join_group_request::*; +use kafka_protocol::messages::list_offsets_request::*; +use kafka_protocol::messages::offset_commit_request::*; +use kafka_protocol::messages::produce_request::*; +use kafka_protocol::messages::txn_offset_commit_request::*; +use kafka_protocol::messages::*; +use kafka_protocol::protocol::{Encodable, StrBytes}; +use kafka_protocol::records::{ + Compression, Record, RecordBatchEncoder, RecordEncodeOptions, TimestampType, +}; +use std::path::PathBuf; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tracing::{info, warn}; + +mod response; + +#[derive(Parser)] +#[command( + name = "kafka-message-gen", + about = "Generate Kafka wire protocol binary messages for all API keys and versions", + long_about = "Generates correctly-framed Kafka protocol requests from Kafka 4.1.0 schemas.\n\ +Each output .bin file is TCP-ready: [len:i32][api_key:i16][api_version:i16][correlation_id:i32][client_id][payload]" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// List all supported API keys with name and version range + List, + /// Generate binary .bin files for all API keys and versions + Generate { + #[arg(short, long, default_value = "kafka_messages")] + output: PathBuf, + /// Filter to API key(s), repeatable: --api-key 0 --api-key 1 + #[arg(long, action = clap::ArgAction::Append)] + api_key: Vec<i16>, + /// Filter to a single version + #[arg(long)] + version: Option<i16>, + /// Print hex dump to stdout + #[arg(long)] + hex: bool, + }, + /// Send messages to a live Kafka-compatible server and show responses + Send { + #[arg(long, default_value = "127.0.0.1:9092")] + host: String, + #[arg(long)] + api_key: Option<i16>, + #[arg(long)] + version: Option<i16>, + #[arg(long, default_value = "5000")] + timeout_ms: u64, + /// Compact one-line output (default is verbose decoded response) + #[arg(long)] + quiet: bool, + }, + Verify { + #[arg(long, default_value = "127.0.0.1:9092")] + host: String, + /// Limit to these API keys (repeatable). Defaults to all gateway-scoped keys. + #[arg(long, action = clap::ArgAction::Append)] + api_key: Vec<i16>, + /// Limit to a single protocol version + #[arg(long)] + version: Option<i16>, + #[arg(long, default_value = "5000")] + timeout_ms: u64, + /// Stop on the first failure + #[arg(long)] + fail_fast: bool, + /// Use the full Kafka 4.1 registry (for real brokers), not the Iggy gateway scope + #[arg(long)] + all_apis: bool, + /// Compact one-line output (default is verbose decoded response) + #[arg(long)] + quiet: bool, + }, +} + +// ── API Registry ───────────────────────────────────────────────────────────── +// Source: validVersions in apache/kafka trunk JSON schema files, Kafka 4.1.0 +// Format: (api_key, name, min_version, max_version) +const API_REGISTRY: &[(i16, &str, i16, i16)] = &[ + (0, "Produce", 3, 13), + (1, "Fetch", 4, 18), + (2, "ListOffsets", 1, 11), + (3, "Metadata", 0, 13), + (8, "OffsetCommit", 2, 10), + (9, "OffsetFetch", 1, 10), + (10, "FindCoordinator", 0, 6), + (11, "JoinGroup", 0, 9), + (12, "Heartbeat", 0, 4), + (13, "LeaveGroup", 0, 5), + (14, "SyncGroup", 0, 5), + (15, "DescribeGroups", 0, 6), + (16, "ListGroups", 0, 5), + (17, "SaslHandshake", 0, 1), + (18, "ApiVersions", 0, 5), + (19, "CreateTopics", 2, 7), + (20, "DeleteTopics", 1, 6), + (21, "DeleteRecords", 0, 2), + (22, "InitProducerId", 0, 6), + (23, "OffsetForLeaderEpoch", 2, 4), + (24, "AddPartitionsToTxn", 0, 5), + (25, "AddOffsetsToTxn", 0, 4), + (26, "EndTxn", 0, 5), + (27, "WriteTxnMarkers", 1, 2), + (28, "TxnOffsetCommit", 0, 5), + (29, "DescribeAcls", 1, 3), + (30, "CreateAcls", 1, 3), + (31, "DeleteAcls", 1, 3), + (32, "DescribeConfigs", 1, 4), + (33, "AlterConfigs", 0, 2), + (34, "AlterReplicaLogDirs", 1, 2), + (35, "DescribeLogDirs", 1, 5), + (36, "SaslAuthenticate", 0, 2), + (37, "CreatePartitions", 0, 3), + (38, "CreateDelegationToken", 1, 3), + (39, "RenewDelegationToken", 1, 2), + (40, "ExpireDelegationToken", 1, 2), + (41, "DescribeDelegationToken", 1, 3), + (42, "DeleteGroups", 0, 2), + (43, "ElectLeaders", 0, 2), + (44, "IncrementalAlterConfigs", 0, 1), + (45, "AlterPartitionReassignments", 0, 1), + (46, "ListPartitionReassignments", 0, 1), + (47, "OffsetDelete", 0, 0), + (48, "DescribeClientQuotas", 0, 1), + (49, "AlterClientQuotas", 0, 1), + (50, "DescribeUserScramCredentials", 0, 0), + (51, "AlterUserScramCredentials", 0, 0), + (55, "DescribeQuorum", 2, 3), + (56, "AlterPartition", 2, 3), + (57, "UpdateFeatures", 0, 2), + (60, "DescribeCluster", 0, 2), + (61, "DescribeProducers", 0, 0), + (64, "UnregisterBroker", 0, 0), + (65, "DescribeTransactions", 0, 0), + (66, "ListTransactions", 0, 1), + (67, "AllocateProducerIds", 0, 0), + (68, "ConsumerGroupHeartbeat", 0, 1), + (69, "ConsumerGroupDescribe", 0, 1), + (71, "GetTelemetrySubscriptions", 0, 0), + (72, "PushTelemetry", 0, 0), + (74, "AssignReplicasToDirs", 0, 0), + (75, "DescribeTopicPartitions", 0, 0), + (76, "ListClientMetricsResources", 0, 0), +]; + +/// Gateway verify scope from `iggy_gateway_kafka::supported_api_ranges()`. +/// Names come from `API_REGISTRY` (Kafka catalog); version bounds come from the gateway. +fn gateway_verify_registry() -> Vec<(i16, &'static str, i16, i16)> { + supported_api_ranges() + .iter() + .map(|range| { + let name = API_REGISTRY + .iter() + .find(|(key, _, _, _)| *key == range.api_key) + .map_or("Unknown", |(_, name, _, _)| *name); + (range.api_key, name, range.min_version, range.max_version) + }) + .collect() +} + +// ── Request framing ─────────────────────────────────────────────────────────── +// Wire format (Kafka protocol spec): +// [total_length: i32] big-endian, excludes self +// [api_key: i16] +// [api_version: i16] +// [correlation_id: i32] +// header v1: [client_id: NULLABLE_STRING] +// header v2: [client_id: COMPACT_NULLABLE_STRING] [request_header_tagged_fields] +// [payload: bytes] + +fn write_unsigned_varint(buf: &mut BytesMut, mut value: u64) { + loop { + let mut byte = (value & 0x7F) as u8; + value >>= 7; + if value != 0 { + byte |= 0x80; + } + buf.put_u8(byte); + if value == 0 { + break; + } + } +} + +fn write_compact_nullable_string(buf: &mut BytesMut, value: Option<&str>) { + match value { + None => write_unsigned_varint(buf, 0), + Some(s) => { + write_unsigned_varint(buf, (s.len() + 1) as u64); + buf.put_slice(s.as_bytes()); + } + } +} + +fn frame_request( + api_key: i16, + api_version: i16, + correlation_id: i32, + client_id: &str, + payload: &[u8], + flexible: bool, +) -> Bytes { + let mut header = BytesMut::new(); + header.put_i16(api_key); + header.put_i16(api_version); + header.put_i32(correlation_id); + if flexible { + write_compact_nullable_string(&mut header, Some(client_id)); + header.put_u8(0); // empty request-header tagged fields + } else { + header.put_i16(i16::try_from(client_id.len()).expect("client_id fits i16")); + header.put_slice(client_id.as_bytes()); + } + + let blen = header.len() + payload.len(); + let mut buf = BytesMut::with_capacity(4 + blen); + buf.put_i32(i32::try_from(blen).expect("frame fits i32")); + buf.put_slice(&header); + buf.put_slice(payload); + buf.freeze() +} + +// ── Payload builders ────────────────────────────────────────────────────────── +// Build the API-specific encoded body for a given api_key and version. +// All required fields contain realistic non-zero values. +// Returns raw bytes WITHOUT the framing header. +fn build_payload(api_key: i16, version: i16) -> Result<Bytes> { + let mut buf = BytesMut::new(); + match api_key { + 18 => { + let mut r = ApiVersionsRequest::default(); + if version >= 3 { + r.client_software_name = StrBytes::from_static_str("kafka-message-gen"); + r.client_software_version = StrBytes::from_static_str("0.1.0"); + } + r.encode(&mut buf, version).context("ApiVersions")?; + } + 3 => { + let mut r = MetadataRequest::default(); + if version >= 1 { + r.topics = None; + } + if version >= 4 { + r.allow_auto_topic_creation = true; + } + if version >= 8 { + r.include_cluster_authorized_operations = false; + r.include_topic_authorized_operations = false; + } + r.encode(&mut buf, version).context("Metadata")?; + } + 0 => { + let rec = Record { + transactional: false, + control: false, + partition_leader_epoch: 0, + producer_id: -1, + producer_epoch: -1, + timestamp_type: TimestampType::Creation, + offset: 0, + sequence: 0, + timestamp: 1_700_000_000_000, + key: Some(Bytes::from_static(b"test-key")), + value: Some(Bytes::from_static(b"test-value")), + headers: indexmap::IndexMap::new(), + }; + let mut rb = BytesMut::new(); + RecordBatchEncoder::encode( + &mut rb, + [rec].iter(), + &RecordEncodeOptions { + version: 2, + compression: Compression::None, + }, + ) + .context("RecordBatch encode")?; + let pd = TopicProduceData::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partition_data(vec![ + PartitionProduceData::default() + .with_index(0) + .with_records(Some(rb.freeze())), + ]); + let mut r = ProduceRequest::default() + .with_acks(-1) + .with_timeout_ms(5000) + .with_topic_data(vec![pd]); + if version >= 3 { + r.transactional_id = None; + } + r.encode(&mut buf, version).context("Produce")?; + } + 1 => { + let fp = FetchPartition::default() + .with_partition(0) + .with_fetch_offset(0) + .with_partition_max_bytes(1_048_576); + let ft = FetchTopic::default() + .with_topic(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![fp]); + let mut r = FetchRequest::default() + .with_replica_id(BrokerId(-1)) + .with_max_wait_ms(500) + .with_min_bytes(1) + .with_topics(vec![ft]); + if version >= 3 { + r.max_bytes = 52_428_800; + } + if version >= 4 { + r.isolation_level = 0; + } + if version >= 7 { + r.session_id = 0; + r.session_epoch = -1; + } + r.encode(&mut buf, version).context("Fetch")?; + } + 2 => { + let p = ListOffsetsPartition::default() + .with_partition_index(0) + .with_timestamp(-1); + let t = ListOffsetsTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + ListOffsetsRequest::default() + .with_replica_id(BrokerId(-1)) + .with_isolation_level(0) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("ListOffsets")?; + } + 8 => { + let p = OffsetCommitRequestPartition::default() + .with_partition_index(0) + .with_committed_offset(42) + .with_committed_metadata(Some(StrBytes::from_static_str(""))); + let t = OffsetCommitRequestTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + OffsetCommitRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("OffsetCommit")?; + } + 9 => { + OffsetFetchRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .encode(&mut buf, version) + .context("OffsetFetch")?; + } + 10 => { + FindCoordinatorRequest::default() + .with_key(StrBytes::from_static_str("test-group")) + .with_key_type(0) + .encode(&mut buf, version) + .context("FindCoordinator")?; + } + 11 => { + let p = JoinGroupRequestProtocol::default() + .with_name(StrBytes::from_static_str("range")) + .with_metadata(Bytes::from_static(b"\x00\x00\x00\x01\x00\x0atest-topic")); + JoinGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_session_timeout_ms(30_000) + .with_rebalance_timeout_ms(300_000) + .with_member_id(StrBytes::from_static_str("")) + .with_protocol_type(StrBytes::from_static_str("consumer")) + .with_protocols(vec![p]) + .encode(&mut buf, version) + .context("JoinGroup")?; + } + 12 => { + HeartbeatRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_generation_id(1) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .encode(&mut buf, version) + .context("Heartbeat")?; + } + 13 => { + LeaveGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .encode(&mut buf, version) + .context("LeaveGroup")?; + } + 14 => { + SyncGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_generation_id(1) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .with_protocol_type(Some(StrBytes::from_static_str("consumer"))) + .with_protocol_name(Some(StrBytes::from_static_str("range"))) + .encode(&mut buf, version) + .context("SyncGroup")?; + } + 15 => { + DescribeGroupsRequest::default() + .with_groups(vec![GroupId::from(StrBytes::from_static_str("test-group"))]) + .with_include_authorized_operations(false) + .encode(&mut buf, version) + .context("DescribeGroups")?; + } + 16 => { + ListGroupsRequest::default() + .encode(&mut buf, version) + .context("ListGroups")?; + } + 17 => { + SaslHandshakeRequest::default() + .with_mechanism(StrBytes::from_static_str("PLAIN")) + .encode(&mut buf, version) + .context("SaslHandshake")?; + } + 19 => { + let t = CreatableTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str( + "iggy-test-topic", + ))) + .with_num_partitions(1) + .with_replication_factor(1); + CreateTopicsRequest::default() + .with_topics(vec![t]) + .with_timeout_ms(30_000) + .with_validate_only(false) + .encode(&mut buf, version) + .context("CreateTopics")?; + } + 20 => { + let r = if version >= 6 { + DeleteTopicsRequest::default() + .with_topics(vec![DeleteTopicState::default().with_name(Some( + TopicName::from(StrBytes::from_static_str("iggy-test-topic")), + ))]) + .with_timeout_ms(30_000) + } else { + DeleteTopicsRequest::default() + .with_topic_names(vec![TopicName::from(StrBytes::from_static_str( + "iggy-test-topic", + ))]) + .with_timeout_ms(30_000) + }; + r.encode(&mut buf, version).context("DeleteTopics")?; + } + 21 => { + let p = DeleteRecordsPartition::default() + .with_partition_index(0) + .with_offset(0); + let t = DeleteRecordsTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + DeleteRecordsRequest::default() + .with_topics(vec![t]) + .with_timeout_ms(30_000) + .encode(&mut buf, version) + .context("DeleteRecords")?; + } + 22 => { + InitProducerIdRequest::default() + .with_transactional_id(None) + .with_transaction_timeout_ms(60_000) + .encode(&mut buf, version) + .context("InitProducerId")?; + } + 24 => { + let t = AddPartitionsToTxnTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![0i32]); + AddPartitionsToTxnRequest::default() + .with_v3_and_below_transactional_id(TransactionalId(StrBytes::from_static_str( + "test-txn", + ))) + .with_v3_and_below_producer_id(ProducerId(100)) + .with_v3_and_below_producer_epoch(1) + .with_v3_and_below_topics(vec![t]) + .encode(&mut buf, version) + .context("AddPartitionsToTxn")?; + } + 25 => { + AddOffsetsToTxnRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .encode(&mut buf, version) + .context("AddOffsetsToTxn")?; + } + 26 => { + EndTxnRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_committed(true) + .encode(&mut buf, version) + .context("EndTxn")?; + } + 28 => { + let p = TxnOffsetCommitRequestPartition::default() + .with_partition_index(0) + .with_committed_offset(42) + .with_committed_metadata(Some(StrBytes::from_static_str(""))); + let t = TxnOffsetCommitRequestTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + TxnOffsetCommitRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("TxnOffsetCommit")?; + } + 32 => { + let r = DescribeConfigsResource::default() + .with_resource_type(2) + .with_resource_name(StrBytes::from_static_str("test-topic")); + DescribeConfigsRequest::default() + .with_resources(vec![r]) + .encode(&mut buf, version) + .context("DescribeConfigs")?; + } + 36 => { + SaslAuthenticateRequest::default() + .with_auth_bytes(Bytes::from_static(b"\x00iggy\x00secret")) + .encode(&mut buf, version) + .context("SaslAuthenticate")?; + } + other => { + warn!("api_key={other}: no explicit builder — empty payload (framing test)"); + } + } + Ok(buf.freeze()) +} + +// Build a complete framed Kafka request message ready for TCP transmission. +fn build_framed(api_key: i16, version: i16, corr: i32) -> Result<Bytes> { + let payload = build_payload(api_key, version)?; + // Header version comes from kafka-protocol's own per-request `HeaderVersion` impl (via + // `ApiKey::request_header_version`), not the gateway's table under test - otherwise a bug + // in the gateway's flexible-version threshold would mis-frame the fixture identically and + // the tests would still pass. + let api = + ApiKey::try_from(api_key).map_err(|()| anyhow::anyhow!("unknown api_key={api_key}"))?; + let flexible = api.request_header_version(version) >= 2; + Ok(frame_request( + api_key, + version, + corr, + "kafka-message-gen", + &payload, + flexible, + )) +} + +// ── Commands ────────────────────────────────────────────────────────────────── + +fn cmd_list() { + println!( + "{:<6} {:<42} {:<10} {:<10} {:<8}", + "Key", "Name", "MinVer", "MaxVer", "Count" + ); + println!("{}", "─".repeat(78)); + for &(k, n, min, max) in API_REGISTRY { + println!( + "{:<6} {:<42} {:<10} {:<10} {:<8}", + k, + n, + min, + max, + max - min + 1 + ); + } + let total: i16 = API_REGISTRY + .iter() + .map(|&(_, _, min, max)| max - min + 1) + .sum(); + println!("{}", "─".repeat(78)); + println!( + "Total: {} API keys | {} versioned messages", + API_REGISTRY.len(), + total + ); +} + +async fn cmd_generate( + out: PathBuf, + filter_keys: Vec<i16>, + fv: Option<i16>, + hex_dump: bool, +) -> Result<()> { + tokio::fs::create_dir_all(&out).await?; + let (mut n, mut corr) = (0usize, 1i32); + for &(ak, name, min, max) in API_REGISTRY { + if !filter_keys.is_empty() && !filter_keys.contains(&ak) { + continue; + } + for v in min..=max { + if fv.is_some_and(|fv| fv != v) { + continue; + } + match build_framed(ak, v, corr) { + Ok(msg) => { + let fname = format!("{:03}_{}_v{}.bin", ak, name, v); + tokio::fs::write(out.join(&fname), &msg).await?; + if hex_dump { + println!("── {} v{} ({} bytes) ──", name, v, msg.len()); + println!("{}", hex::encode(&msg)); + println!(); + } else { + info!(" {} ({} bytes)", fname, msg.len()); + } + n += 1; + corr += 1; + } + Err(e) => warn!("SKIP {} v{}: {e}", name, v), + } + } + } + println!("\n✓ Generated {n} messages → {}/", out.display()); + println!( + " Quick test: cat {}/018_ApiVersions_v3.bin | nc 127.0.0.1 9092 | xxd", + out.display() + ); + Ok(()) +} + +async fn connect(host: &str) -> Result<TcpStream> { + TcpStream::connect(host) + .await + .with_context(|| format!("Cannot connect to {host}")) +} + +async fn read_kafka_response(stream: &mut TcpStream) -> std::io::Result<Vec<u8>> { + let mut lb = [0u8; 4]; + stream.read_exact(&mut lb).await?; + let frame_len = i32::from_be_bytes(lb); + if frame_len <= 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("invalid response frame length: {frame_len}"), + )); + } + let mut body = vec![ + 0u8; + usize::try_from(frame_len).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "response frame length does not fit usize", + ) + })? + ]; + stream.read_exact(&mut body).await?; + Ok(body) +} Review Comment: accepts any positive signed 32-bit response length and immediately allocates a `Vec` of that size. A misconfigured or malicious endpoint can send a four-byte length prefix near `i32::MAX` and make `send` or `verify` attempt a roughly 2 GiB allocation before the read timeout can help. Apply a configured maximum response size before allocation, and preferably read incrementally as the gateway server does for request frames. ########## gateways/kafka/src/server.rs: ########## @@ -0,0 +1,786 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use bytes::{BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{Semaphore, broadcast}; +use tokio::time::{timeout, timeout_at}; +use tokio_util::task::TaskTracker; +use tracing::{debug, error, info, warn}; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::api::{ + BrokerAdvertise, DEFAULT_KAFKA_PORT, ERROR_INVALID_REQUEST, HandleOutcome, + encode_error_only_response, handle_request, +}; +use crate::protocol::codec::Decoder; +use crate::protocol::header::{ + RequestHeader, ResponseHeader, request_header_version, response_header_version, +}; +use std::io; + +const READ_CHUNK: usize = 65536; + +#[derive(Debug, Clone)] +pub struct ServerConfig { + pub bind_addr: String, + /// Hostname or IP advertised in Metadata (`IGGY_KAFKA_ADVERTISED_HOST`). Required when + /// `bind_addr` uses a wildcard address (`0.0.0.0` / `::`). + pub advertised_host: Option<String>, + /// Port advertised in Metadata (`IGGY_KAFKA_ADVERTISED_PORT`). Defaults to the bind port. + pub advertised_port: Option<u16>, + pub max_frame_size: usize, + /// Maximum concurrent connections accepted before new ones are rejected. + pub max_connections: usize, + /// Bound on how long an accepted connection may sit idle before sending the next + /// frame's length prefix. Kafka brokers default `connections.max.idle.ms` to 10 minutes; + /// match that so well-behaved idle clients aren't dropped. + pub idle_timeout: Duration, + pub read_timeout: Duration, + pub write_timeout: Duration, + /// Cap on how long graceful shutdown waits for in-flight connections to finish. Without + /// this, a connection idling inside `idle_timeout` (10 minutes by default) would otherwise + /// hold shutdown open past typical orchestrator grace periods (e.g. Kubernetes' default + /// 30s `terminationGracePeriodSeconds`). + pub shutdown_drain_timeout: Duration, +} + +impl Default for ServerConfig { + fn default() -> Self { + Self { + bind_addr: format!("127.0.0.1:{DEFAULT_KAFKA_PORT}"), + advertised_host: None, + advertised_port: None, + max_frame_size: 8 * 1024 * 1024, + max_connections: 1024, + idle_timeout: Duration::from_mins(10), + read_timeout: Duration::from_secs(15), + write_timeout: Duration::from_secs(10), + shutdown_drain_timeout: Duration::from_secs(25), + } + } +} + +impl BrokerAdvertise { + /// Resolve the broker endpoint advertised in Metadata. + /// + /// `local_addr` is the address the listener is actually bound to (from `listener.local_addr()`). + /// + /// # Errors + /// + /// Returns `InvalidConfig` when `advertised_host` is empty or the listener binds to a wildcard + /// without an explicit advertised host. + pub fn from_server_config(config: &ServerConfig, local_addr: SocketAddr) -> Result<Self> { + let port = config + .advertised_port + .map_or_else(|| i32::from(local_addr.port()), i32::from); + + let host = if let Some(ref advertised) = config.advertised_host { + let trimmed = advertised.trim(); + if trimmed.is_empty() { + return Err(KafkaProtocolError::InvalidConfig( + "IGGY_KAFKA_ADVERTISED_HOST must not be empty".into(), + )); + } + if trimmed.len() > i16::MAX as usize { + return Err(KafkaProtocolError::InvalidConfig( + "IGGY_KAFKA_ADVERTISED_HOST exceeds Kafka nullable string limit (32767 bytes)" + .into(), + )); + } + trimmed.to_string() + } else if local_addr.ip().is_unspecified() { + return Err(KafkaProtocolError::InvalidConfig( + "binding to a wildcard address (0.0.0.0 or ::) requires \ + IGGY_KAFKA_ADVERTISED_HOST to be set to a reachable hostname or IP for \ + Metadata broker advertisement" + .into(), + )); + } else { + local_addr.ip().to_string() + }; + + Ok(Self { host, port }) + } +} + +pub struct KafkaServer { + config: Arc<ServerConfig>, +} + +impl KafkaServer { + #[must_use] + pub fn new(config: ServerConfig) -> Self { + Self { + config: Arc::new(config), + } + } + + /// Accept Kafka wire connections until `shutdown` fires, then drain in-flight tasks. + /// + /// `listener` must already be bound by the caller. This lets tests and `main` bind + /// the port before spawning the task, eliminating the TOCTOU race of bind-drop-rebind. + /// + /// # Errors + /// + /// Returns an error on invalid config or a non-transient `accept()` error. + pub async fn run( + self, + listener: TcpListener, + mut shutdown: broadcast::Receiver<()>, + ) -> Result<()> { + let local_addr = listener.local_addr()?; + let broker = Arc::new(BrokerAdvertise::from_server_config( + &self.config, + local_addr, + )?); + info!( + "kafka listener bound on {} (advertised as {}:{})", + local_addr, broker.host, broker.port + ); + + let tracker = TaskTracker::new(); + let conn_limiter = Arc::new(Semaphore::new(self.config.max_connections)); + + let drain_timeout = self.config.shutdown_drain_timeout; + + loop { + tokio::select! { + result = shutdown.recv() => { + match result { + Ok(()) => { + info!("kafka listener shutdown requested"); + drain(&tracker, drain_timeout).await; + break; + } + // Capacity-1 channel: lagged means a signal was sent before we polled - treat as shutdown. + Err(broadcast::error::RecvError::Lagged(_)) => { + info!("kafka listener shutdown requested (lagged)"); + drain(&tracker, drain_timeout).await; + break; + } + Err(broadcast::error::RecvError::Closed) => { + drain(&tracker, drain_timeout).await; + break; + } + } + } + accept_result = listener.accept() => { + match accept_result { + Ok((stream, peer)) => { + let Ok(permit) = Arc::clone(&conn_limiter).try_acquire_owned() else { + warn!(%peer, max_connections = self.config.max_connections, "connection limit reached, rejecting"); + continue; + }; + if let Err(e) = stream.set_nodelay(true) { + warn!(%peer, "TCP_NODELAY failed: {e}"); + } + if let Err(e) = enable_tcp_keepalive(&stream) { + warn!(%peer, "TCP_KEEPALIVE failed: {e}"); + } + let cfg = Arc::clone(&self.config); + let broker = Arc::clone(&broker); + tracker.spawn(async move { + let _permit = permit; + if let Err(err) = handle_connection(stream, cfg, peer, broker).await { + warn!(%peer, "connection closed with error: {err}"); + } + }); + } + Err(e) if is_transient_accept_error(&e) => { + // Brief backoff on fd exhaustion to avoid busy-spinning. + if matches!(e.raw_os_error(), Some(23 | 24)) { + tokio::time::sleep(Duration::from_millis(10)).await; + } + warn!(%e, "transient accept error, continuing"); + } + Err(e) => { + drain(&tracker, drain_timeout).await; + return Err(e.into()); + } + } + } + + } + } + Ok(()) + } +} + +/// Close the tracker to new spawns and wait for in-flight connections to finish, but not past +/// `deadline` - an idle connection can otherwise hold shutdown open for up to `idle_timeout` +/// (10 minutes by default), past typical orchestrator grace periods. +async fn drain(tracker: &TaskTracker, deadline: Duration) { + tracker.close(); + if timeout(deadline, tracker.wait()).await.is_err() { + warn!( + ?deadline, + "shutdown drain deadline exceeded; abandoning in-flight connections" + ); + } +} Review Comment: This closes the `TaskTracker` and times out `tracker.wait()`, but neither action cancels tracked tasks. `KafkaServer::run` can therefore return `Ok(())` while existing connections continue processing for up to the idle timeout or longer. The test at `gateways/kafka/src/server.rs:600` only checks that `run` returns, not that the held socket closes. Use a cancellation token or explicitly abort connection tasks after the drain deadline, and assert that clients receive EOF during shutdown. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,576 @@ +// 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::Bytes; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::codec::{Decoder, Encoder, PREALLOC_HINT}; +use crate::protocol::requests::{ + ProduceDecodeResult, decode_create_topics_request, decode_fetch_request, + decode_list_offsets_request, decode_produce_request, +}; +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_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; + +const MAX_SUPPORTED_METADATA_VERSION: i16 = 9; + +/// Sentinel for `topic_authorized_operations` / `cluster_authorized_operations` when ACLs are not supported. +const AUTHORIZED_OPS_UNKNOWN: i32 = i32::MIN; + +/// Result of handling one Kafka request body. +#[derive(Debug)] +pub enum HandleOutcome { + /// Write this response body (with a response header). + Respond(Bytes), + /// Write this response body (with a response header), then close the TCP connection. + RespondAndClose(Bytes), + /// Produce with `acks=0`: write nothing, keep the connection open. + NoResponse, + /// Client cannot parse an error at this request wire version; close the TCP connection. + Close, +} + +impl HandleOutcome { + /// Collapse to `Some(body)` for a normal response, or `None` for [`HandleOutcome::NoResponse`]. + /// + /// # Panics + /// + /// Panics on [`HandleOutcome::Close`] - match on `Close` explicitly, or use + /// [`Self::expect_response`] in tests that require a body. + #[must_use] + pub fn into_optional_response(self) -> Option<Bytes> { + match self { + Self::Respond(body) | Self::RespondAndClose(body) => Some(body), + Self::NoResponse => None, + Self::Close => panic!("HandleOutcome::Close has no response body"), + } + } + + /// 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) | Self::RespondAndClose(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) +} + +/// 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 `acks`, 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 { + match decode_produce_request(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. + ProduceDecodeResult::Ok(req) if req.acks == 0 => HandleOutcome::NoResponse, + ProduceDecodeResult::Ok(req) => { + if !is_supported_version(API_KEY_PRODUCE, api_version) { + return HandleOutcome::Respond(encode_produce_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )); + } + HandleOutcome::Respond(encode_produce_response(api_version, &req)) + } + ProduceDecodeResult::Err { + acks: Some(0), + error, + } => { + tracing::warn!( + "Failed to decode Produce request with acks=0 (no response): {:?}", + error + ); + HandleOutcome::NoResponse + } + ProduceDecodeResult::Err { error, .. } => { + tracing::warn!("Failed to decode Produce request: {:?}", error); + let code = if is_supported_version(API_KEY_PRODUCE, api_version) { + ERROR_INVALID_REQUEST + } else { + ERROR_UNSUPPORTED_VERSION + }; + HandleOutcome::Respond(encode_produce_error_response(api_version, code)) + } + } +} + +fn handle_other_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> HandleOutcome { + match api_key { + API_KEY_API_VERSIONS => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_api_versions_response(api_version, ERROR_NONE)) + } else { + // KIP-511: reply with v0 when the requested version is not understood. + HandleOutcome::Respond(encode_api_versions_response(0, ERROR_UNSUPPORTED_VERSION)) + } + } + API_KEY_METADATA => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_metadata_response( + api_version, + api_version, + body, + broker, + ERROR_NONE, + )) + } else { + // Clamping the response to MAX_SUPPORTED_METADATA_VERSION leaves a body the + // client parses at its own (unsupported) version, so UNSUPPORTED_VERSION never + // survives. Clients that skip ApiVersions get a naked close instead. + tracing::warn!( + api_version, + max_supported = MAX_SUPPORTED_METADATA_VERSION, + "Metadata version unsupported; closing connection" + ); + HandleOutcome::Close + } + } + API_KEY_FETCH => { + if is_supported_version(api_key, api_version) { + match decode_fetch_request(api_version, body) { + Ok(req) => HandleOutcome::Respond(encode_fetch_response(api_version, &req)), + Err(e) => { + tracing::warn!("Failed to decode Fetch request: {:?}", e); + HandleOutcome::Respond(encode_fetch_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_fetch_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + API_KEY_LIST_OFFSETS => { + if is_supported_version(api_key, api_version) { + match decode_list_offsets_request(api_version, body) { + Ok(req) => { + HandleOutcome::Respond(encode_list_offsets_response(api_version, &req)) + } + Err(e) => { + tracing::warn!("Failed to decode ListOffsets request: {:?}", e); + HandleOutcome::Respond(encode_list_offsets_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_list_offsets_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + API_KEY_CREATE_TOPICS => { + if is_supported_version(api_key, api_version) { + match decode_create_topics_request(api_version, body) { + Ok(req) => { + HandleOutcome::Respond(encode_create_topics_response(api_version, &req)) + } + Err(e) => { + tracing::warn!("Failed to decode CreateTopics request: {:?}", e); + HandleOutcome::Respond(encode_create_topics_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } + } else { + HandleOutcome::Respond(encode_create_topics_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )) + } + } + _ => HandleOutcome::RespondAndClose(encode_error_only_response(ERROR_UNSUPPORTED_VERSION)), + } +} + +#[must_use] +pub fn is_supported_version(api_key: i16, api_version: i16) -> bool { + SUPPORTED_RANGES + .iter() + .find(|r| r.api_key == api_key) + .is_some_and(|r| api_version >= r.min_version && api_version <= r.max_version) +} + +/// Min version advertised in `ApiVersions` (may differ from the firewall min). +/// +/// Produce must advertise min=0 per KAFKA-18659 / `PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` +/// even though this gateway only accepts Produce v3+. +#[must_use] +pub const fn advertised_min_version(api_key: i16, firewall_min: i16) -> i16 { + if api_key == API_KEY_PRODUCE { + 0 + } else { + firewall_min + } +} + +fn encode_api_versions_response(api_version: i16, error_code: i16) -> Bytes { + let flexible = api_version >= 3; + let ranges = SUPPORTED_RANGES; + let mut e = Encoder::with_capacity(128); + + e.write_i16(error_code); + + if flexible { + e.write_varint((ranges.len() + 1) as u64); + for r in ranges { + e.write_i16(r.api_key); + e.write_i16(advertised_min_version(r.api_key, r.min_version)); + e.write_i16(r.max_version); + e.write_empty_tagged_fields(); + } + } else { + e.write_i32(i32::try_from(ranges.len()).expect("supported range table is small")); + for r in ranges { + e.write_i16(r.api_key); + e.write_i16(advertised_min_version(r.api_key, r.min_version)); + e.write_i16(r.max_version); + } + } + + if api_version >= 1 { + e.write_i32(0); Review Comment: This passes the requested version into the response encoder. CreateTopics v7 requires a `TopicId` between `Name` and `ErrorCode`, but `gateways/kafka/src/protocol/responses.rs:375` omits it. Produce v13 similarly replaces the topic name with a UUID while `gateways/kafka/src/protocol/responses.rs:67` still writes a string. Clients cannot parse the intended `UNSUPPORTED_VERSION` response. These versions should close the connection or receive explicitly version-correct responses. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,576 @@ +// 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::Bytes; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::codec::{Decoder, Encoder, PREALLOC_HINT}; +use crate::protocol::requests::{ + ProduceDecodeResult, decode_create_topics_request, decode_fetch_request, + decode_list_offsets_request, decode_produce_request, +}; +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_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; + +const MAX_SUPPORTED_METADATA_VERSION: i16 = 9; + +/// Sentinel for `topic_authorized_operations` / `cluster_authorized_operations` when ACLs are not supported. +const AUTHORIZED_OPS_UNKNOWN: i32 = i32::MIN; + +/// Result of handling one Kafka request body. +#[derive(Debug)] +pub enum HandleOutcome { + /// Write this response body (with a response header). + Respond(Bytes), + /// Write this response body (with a response header), then close the TCP connection. + RespondAndClose(Bytes), + /// Produce with `acks=0`: write nothing, keep the connection open. + NoResponse, + /// Client cannot parse an error at this request wire version; close the TCP connection. + Close, +} + +impl HandleOutcome { + /// Collapse to `Some(body)` for a normal response, or `None` for [`HandleOutcome::NoResponse`]. + /// + /// # Panics + /// + /// Panics on [`HandleOutcome::Close`] - match on `Close` explicitly, or use + /// [`Self::expect_response`] in tests that require a body. + #[must_use] + pub fn into_optional_response(self) -> Option<Bytes> { + match self { + Self::Respond(body) | Self::RespondAndClose(body) => Some(body), + Self::NoResponse => None, + Self::Close => panic!("HandleOutcome::Close has no response body"), + } + } + + /// 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) | Self::RespondAndClose(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) +} + +/// 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 `acks`, 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 { + match decode_produce_request(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. + ProduceDecodeResult::Ok(req) if req.acks == 0 => HandleOutcome::NoResponse, + ProduceDecodeResult::Ok(req) => { + if !is_supported_version(API_KEY_PRODUCE, api_version) { + return HandleOutcome::Respond(encode_produce_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )); + } + HandleOutcome::Respond(encode_produce_response(api_version, &req)) + } + ProduceDecodeResult::Err { + acks: Some(0), + error, + } => { + tracing::warn!( + "Failed to decode Produce request with acks=0 (no response): {:?}", + error + ); + HandleOutcome::NoResponse + } + ProduceDecodeResult::Err { error, .. } => { + tracing::warn!("Failed to decode Produce request: {:?}", error); + let code = if is_supported_version(API_KEY_PRODUCE, api_version) { + ERROR_INVALID_REQUEST + } else { + ERROR_UNSUPPORTED_VERSION + }; + HandleOutcome::Respond(encode_produce_error_response(api_version, code)) + } + } +} + +fn handle_other_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> HandleOutcome { + match api_key { + API_KEY_API_VERSIONS => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_api_versions_response(api_version, ERROR_NONE)) + } else { + // KIP-511: reply with v0 when the requested version is not understood. + HandleOutcome::Respond(encode_api_versions_response(0, ERROR_UNSUPPORTED_VERSION)) + } + } + API_KEY_METADATA => { + if is_supported_version(api_key, api_version) { + HandleOutcome::Respond(encode_metadata_response( + api_version, + api_version, + body, + broker, + ERROR_NONE, Review Comment: This returns success based solely on the requested version. V3 requires `ClientSoftwareName`, `ClientSoftwareVersion`, and tagged fields, but an empty body receives a normal response. Malformed flexible encodings therefore bypass the advertised decode and validation layer. The request body should be decoded according to its version before constructing the response. -- 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]
