You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/09/19 08:47:55 UTC
(iggy) 01/01: refactor(gateways): prepare the Kafka gateway for parallel handler work
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-gateway-prep
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 07366c46f1af803f429e42b1e420d9e23b794d3d
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Sep 19 13:47:42 2026 +0530
refactor(gateways): prepare the Kafka gateway for parallel handler work
Handlers go async, each API key gets a file, and the bridge connects behind a default-off flag.
---
.config/nextest.toml | 21 +-
gateways/kafka/README.md | 1 +
gateways/kafka/src/bridge/iggy_bridge.rs | 529 --------------------
.../mod.rs => bridge/iggy_bridge/fetch.rs} | 8 +-
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 187 ++++++++
gateways/kafka/src/bridge/iggy_bridge/offsets.rs | 176 +++++++
.../mod.rs => bridge/iggy_bridge/produce.rs} | 8 +-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 220 +++++++++
gateways/kafka/src/main.rs | 73 ++-
gateways/kafka/src/protocol/api.rs | 532 +++------------------
.../kafka/src/protocol/handlers/api_versions.rs | 108 +++++
.../kafka/src/protocol/handlers/create_topics.rs | 131 +++++
gateways/kafka/src/protocol/handlers/fetch.rs | 119 +++++
.../kafka/src/protocol/handlers/list_offsets.rs | 116 +++++
gateways/kafka/src/protocol/handlers/metadata.rs | 172 +++++++
gateways/kafka/src/protocol/handlers/mod.rs | 181 +++++++
gateways/kafka/src/protocol/handlers/produce.rs | 151 ++++++
gateways/kafka/src/protocol/mod.rs | 2 +-
gateways/kafka/src/protocol/responses.rs | 318 ------------
gateways/kafka/src/server.rs | 41 +-
gateways/kafka/tests/api_handler_tests.rs | 200 +++++---
.../kafka/tests/bridge_iggy_integration_tests.rs | 398 +--------------
gateways/kafka/tests/broker_advertise_tests.rs | 5 +-
gateways/kafka/tests/common/iggy_server.rs | 411 ++++++++++++++++
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 15 +-
gateways/kafka/tests/listener_robustness_tests.rs | 1 +
gateways/kafka/tests/response_negative_tests.rs | 42 +-
gateways/kafka/tests/server_e2e_tests.rs | 2 +-
gateways/kafka/tests/version_firewall_tests.rs | 148 +++---
29 files changed, 2368 insertions(+), 1948 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 1ff8ee2c5..06bea38e7 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -43,17 +43,18 @@ test-group = "elasticsearch"
# from PortGuard's own flock, so no two ever share a port). #[serial] on them is a dead letter
# under nextest - nextest runs every test as its own process, and #[serial]'s mutex only excludes
# within one process (nextest does not support #[serial] macro, same gap the workaround above
-# documents). Without this override, nextest's default scheduler runs several of these processes
-# at once. The test file's own harness sets IGGY_SHARDING_PIN_CORES=false for every spawned
-# server, so pinned-core contention itself isn't the remaining risk; cpu_allocation = "numa:auto"
-# (config.toml) still is - nothing sets IGGY_SHARDING_CPU_ALLOCATION, so every spawned server still
-# sizes its shard pool to the whole box, and several such servers running at once compete for it,
-# the same reason core/integration's own harness (harness/handle/server.rs) overrides both
-# settings for its spawned test servers. Keep this override even if that gets fixed too: it's also
-# what keeps this crate's own real-server tests from adding to contention against unrelated
-# packages' tests in the same nextest run.
+# documents). Without this override, nextest's default scheduler runs as many of these processes
+# at once as it has threads.
+#
+# The harness (tests/common/iggy_server.rs) sets IGGY_SHARDING_PIN_CORES=false and
+# IGGY_SHARDING_CPU_ALLOCATION=0..4 on every server it spawns, the same pairing core/integration's
+# own harness (harness/handle/server.rs) uses, so a spawned server no longer sizes its shard pool
+# to the whole box. core/integration spawns a server per test under no group cap at all on the
+# strength of that. The cap here is kept anyway, at the width of one server's core allocation, so
+# that this crate's real-server tests bound their own shard-thread footprint rather than
+# competing for the box against unrelated packages' tests in the same run.
[test-groups.kafka_bridge]
-max-threads = 1
+max-threads = 4
[[profile.default.overrides]]
filter = 'binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)'
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index f20292390..13ad30652 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -23,6 +23,7 @@ Default bind: `127.0.0.1:9093`. Environment variables:
| `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 |
+| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. A failed connection is fatal, not a downgrade to stubs. |
## Test
diff --git a/gateways/kafka/src/bridge/iggy_bridge.rs b/gateways/kafka/src/bridge/iggy_bridge.rs
deleted file mode 100644
index 41c314d25..000000000
--- a/gateways/kafka/src/bridge/iggy_bridge.rs
+++ /dev/null
@@ -1,529 +0,0 @@
-// 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::future::Future;
-use std::time::Duration;
-
-use iggy::prelude::{
- AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder, IggyError,
- StreamClient, TopicClient, TopicCreateOptions,
-};
-use tracing::{debug, info};
-
-use crate::bridge::config::IggyBridgeConfig;
-use crate::bridge::error::BridgeError;
-use crate::bridge::topic_map::validate_kafka_topic_name;
-
-/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
-///
-/// Not the SDK's own default (`TcpClientReconnectionConfig::default()` is `max_retries: None` -
-/// unlimited, one dial per second, forever). A Kafka client already retries at the wire-protocol
-/// level once a handler maps a bridge failure to a retriable error code; the bridge blocking a
-/// request task inside an unbounded internal reconnect loop would just add a second, invisible
-/// retry layer underneath that one instead of surfacing the failure so the mapped code can be
-/// sent.
-///
-/// This bounds the *count*, not the *wall-clock time*, of that inner retry loop - see
-/// [`REQUEST_TIMEOUT`] for the latter.
-const RECONNECTION_RETRIES: u32 = 3;
-
-/// Wall-clock ceiling [`with_request_timeout`] applies uniformly: to the initial `client.connect()`
-/// in [`IggyBridge::connect`] (including every attempt [`RECONNECTION_RETRIES`] makes internally),
-/// and to every call made after it succeeds (`get_stream`, `create_stream`, `get_topic`,
-/// `create_topic`, `shutdown`).
-///
-/// One constant, not two separately-named ones with the same value: both call sites bound the
-/// identical underlying hazard. `TcpClient::establish_bounded` only applies its own
-/// `FAILOVER_DIAL_TIMEOUT` (2s) when at least two failover candidates are configured
-/// (`tcp_client.rs`) - a bridge always configures exactly one address, so that guard never engages
-/// at either site, and the plain `TcpStream::connect` underneath has no deadline of its own.
-/// Against a firewall that drops SYN packets instead of refusing them, each dial attempt pays the
-/// kernel's own SYN-retry timeout (minutes, not seconds) rather than the `reconnection_interval`
-/// between attempts - a closed port (instant RST) never exercises this path, so the failure mode
-/// only shows up in production. 15s comfortably covers a slow-but-alive server's handshake (well
-/// above p99 login latency) while still failing well short of the pathological multi-minute case.
-///
-/// Known limitation, not closed by this constant: [`with_request_timeout`] cancels only the
-/// *caller's* wait, not the SDK's own work. `TcpClient::send_raw_vsr_attempt` (`tcp_client.rs`)
-/// writes, flushes and reads inside a detached `tokio::spawn` specifically so that dropping the
-/// awaiting future - exactly what this timeout does on expiry - cannot abort it mid-flight (that
-/// function's own "SAFETY: we run code holding the `stream` lock in a task so we can't be
-/// cancelled while holding the lock"). A call that times out here can leave that detached task
-/// still holding `IggyClient`'s single stream mutex for up to `RESPONSE_READ_TIMEOUT` (30s,
-/// `tcp_client.rs`) longer, queuing every other bridge call behind it. Raising this constant does
-/// not close the gap either: a mid-call reconnect (`send_raw_with_response`, `tcp_client.rs`)
-/// replays on a fresh `RESPONSE_READ_TIMEOUT` budget, and the reconnect dial itself goes through
-/// the same undead-lined `TcpStream::connect` this constant exists to bound - genuinely unbounded,
-/// so no finite value here can guarantee catching it. That same unboundedness is why this timeout
-/// cannot simply be dropped for post-connect calls either: [`IggyBridge`] holds one `IggyClient`
-/// with no pooling, so an unbounded reconnect dial with nothing here to stop it would wedge every
-/// later call on this bridge, not just the one that triggered it. Tolerable only because nothing
-/// calls this bridge from a live Kafka handler yet - closing it needs either a cooperatively
-/// cancellable SDK call or a deadline on the SDK's own reconnect dial, neither of which this
-/// bridge can add from the outside. Must be resolved before #3535/#3536 share this client across
-/// concurrent connections.
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
-
-/// Wraps a single Iggy client call in [`REQUEST_TIMEOUT`]. See that constant's doc for why every
-/// bridge method needs this, not just [`IggyBridge::connect`] - and for the cancellation gap this
-/// wrapper does not close.
-///
-/// On expiry, maps to [`BridgeError::Timeout`], not `IggyError::CannotEstablishConnection`: the
-/// SDK's write/read run in a detached task this timeout cannot abort (see [`REQUEST_TIMEOUT`]'s
-/// doc), so by the time this fires the request may already be on the wire, or already applied
-/// server-side - a genuinely unknown outcome, not a known connection failure. `BridgeError`'s own
-/// mapping already treats an unknown outcome (`IggyError::TransientNotCommitted`) as distinct from
-/// a known-safe-to-retry one for exactly this reason; a caller-side timeout is the same shape and
-/// must not borrow the known-safe code.
-async fn with_request_timeout<T>(
- op: impl Future<Output = Result<T, IggyError>>,
-) -> Result<T, BridgeError> {
- tokio::time::timeout(REQUEST_TIMEOUT, op)
- .await
- .map_err(|_elapsed| BridgeError::Timeout)?
- .map_err(BridgeError::Iggy)
-}
-
-/// Owns one connected `IggyClient` and resolves Kafka topics against it.
-///
-/// Produce/Fetch handler wiring is a separate, later change (`#3535`/`#3536`) - this type is the
-/// shared plumbing those handlers will call into, exercised standalone here via its own tests and
-/// an integration test against a real `iggy-server`.
-///
-/// One `IggyClient`, shared across every Kafka connection this gateway serves - and the SDK's TCP
-/// transport is lockstep, one request in flight at a time (`tcp_client.rs`: "the connection is
-/// lockstep", its stream mutex held across write, flush, and read). Every concurrent Kafka
-/// connection ends up serialized behind whichever single Iggy request is in flight; the Kafka
-/// side's own connection limit does nothing to relieve this. No pooling exists yet - `close`
-/// already takes `self` by value, which anticipates an eventual `Arc`-shared bridge, but nothing
-/// wires that up today. Documented here and in the README rather than silently discovered under
-/// load once `#3535`/`#3536` land.
-pub struct IggyBridge {
- client: IggyClient,
- config: IggyBridgeConfig,
-}
-
-impl IggyBridge {
- /// Connects to Iggy using `config` and authenticates.
- ///
- /// Builds the client through the SDK's fluent TCP builder rather than hand-assembling an
- /// `iggy://user:pass@host` connection string: that string format splits on `@` then `:`, so a
- /// password containing either character (`p@ss:word`) would be misparsed into a garbled
- /// address instead of failing with a diagnosable config error. The fluent builder passes
- /// `username`/`password` as already-separated fields, sidestepping the ambiguity entirely.
- ///
- /// # Errors
- ///
- /// Returns [`BridgeError::InvalidConfig`] if `config.address` is empty. Returns
- /// [`BridgeError::Timeout`] if connecting takes longer than `REQUEST_TIMEOUT` (an
- /// unreachable-and-silently-dropping address, not just a refused one, is covered - see that
- /// constant's doc). Returns [`BridgeError::Iggy`] if the address is malformed, the TCP
- /// connection fails, or authentication is rejected - this is the boundary
- /// [`BridgeError::to_kafka_error_code`] exists for: a handler calling this must map the error
- /// to a wire response, never panic or unwrap, since an unreachable Iggy backend is an
- /// expected runtime condition, not a bug.
- pub async fn connect(config: IggyBridgeConfig) -> Result<Self, BridgeError> {
- if config.address.trim().is_empty() {
- return Err(BridgeError::InvalidConfig(
- "Iggy address must not be empty".to_string(),
- ));
- }
-
- let credentials =
- Credentials::UsernamePassword(config.username.clone(), config.password.clone());
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(config.address.clone())
- .with_auto_sign_in(AutoLogin::Enabled(credentials))
- .with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
- .build()
- .map_err(BridgeError::Iggy)?;
- with_request_timeout(client.connect()).await?;
- info!("Iggy bridge connected to {}", config.address);
-
- Ok(Self { client, config })
- }
-
- /// Tears down the underlying Iggy client, including its background heartbeat task.
- ///
- /// Not `IggyClient::disconnect`: that only tears down the transport
- /// (`TcpClient::disconnect_transport`) and never touches `heartbeat_handle` - only
- /// `IggyClient`'s own `Drop` aborts that task (`client.rs`). A `disconnect`ed-but-not-dropped
- /// bridge would keep heartbeating on a schedule, hit `NotConnected` (itself in the SDK's
- /// retriable set), and reconnect plus re-authenticate using the credentials `connect`
- /// configured, so the "closed" client silently comes back. `shutdown` sets
- /// `ClientState::Shutdown`, which the heartbeat loop's next `ping` observes as
- /// `IggyError::ClientShutdown` and self-terminates on, and which `sign_in_credentials` never
- /// dials past.
- ///
- /// Takes `self` by value: `shutdown` is terminal (no reconnect is coming back from it), so
- /// nothing legitimate is left to call on this bridge afterward. This does mean a bridge shared
- /// via `Arc` cannot call this directly (`Arc::try_unwrap` first) - not a concern before
- /// `#3535`/`#3536` wire an owning caller in.
- ///
- /// # Errors
- ///
- /// Returns [`BridgeError::Timeout`] if it takes longer than `REQUEST_TIMEOUT`. Returns
- /// [`BridgeError::Iggy`] if the underlying client reports a shutdown failure (e.g. the socket
- /// was already in a state that rejects a clean shutdown).
- pub async fn close(self) -> Result<(), BridgeError> {
- with_request_timeout(self.client.shutdown()).await
- }
-
- /// Ensures the Iggy stream and topic backing `kafka_topic` exist, creating either or both if
- /// missing. Resolves `kafka_topic` through the configured [`TopicMapping`](crate::bridge::topic_map::TopicMapping).
- ///
- /// Idempotent when repeated with the *same* `partition_count`: a `get` before each `create`
- /// means calling this twice for the same topic is a no-op the second time, and a
- /// `NameAlreadyExists` race from a concurrent caller creating the same stream/topic between
- /// this call's `get` and `create` is treated as success, not an error - the desired end state
- /// (it exists) is what idempotency actually promises, not that this call was the one that
- /// created it. A *different* `partition_count` against an already-existing topic is not
- /// idempotent - see [`BridgeError::PartitionCountMismatch`].
- ///
- /// Ensures the stream before the topic, so a topic-creation failure (a different
- /// `partition_count`, a transient error) can leave a stream that now exists with no topic in
- /// it yet - a retry heals this (idempotent on the stream half too), and no rollback is
- /// attempted: `TopicMapping::resolve` sends every *unmapped* Kafka topic to the same
- /// `default_stream`, so deleting a stream on a topic-creation failure risks deleting another
- /// topic's data that happens to share it, and this call has no way to tell whether it was the
- /// one that created the stream in the first place.
- ///
- /// `partition_count` is `u32`, so it cannot carry Kafka's own `CreateTopics` sentinel
- /// (`num_partitions == -1`, KIP-464 "use the broker default" - `protocol/responses.rs`
- /// already accepts that sentinel at the wire-validation layer). Resolving *what* the broker
- /// default should be for this bridge is a decision for whichever of `#3535`/`#3536` first
- /// calls this with a real Kafka request in hand, not one to invent here ahead of that need.
- ///
- /// No caching: every call pays a `get_stream` and a `get_topic` (two round trips once both
- /// already exist), even for a topic this same bridge already confirmed a moment ago. A cache
- /// keyed on `kafka_topic` would remove that cost, but would also have to answer "how does a
- /// cache entry ever get invalidated" - the topic being deleted and recreated with a different
- /// partition count out from under a stale cache entry is exactly
- /// `ensure_topic_targets_the_streams_live_incarnation_after_a_delete_and_recreate`'s own
- /// scenario, and a naive cache breaks that guarantee to save two round trips. Whatever wires
- /// this into `#3535`/`#3536` should call it once per topic and remember that it did, rather
- /// than once per Produce/Fetch - the cost belongs at the call site's discretion, not hidden
- /// (and potentially made wrong) inside this method.
- ///
- /// # Errors
- ///
- /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
- /// topic-naming rules. Returns [`BridgeError::Timeout`] if a call takes longer than
- /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] for connectivity/auth failures. Returns
- /// [`BridgeError::PartitionCountMismatch`] if the topic already exists with a different
- /// partition count than `partition_count`.
- pub async fn ensure_stream_and_topic(
- &self,
- kafka_topic: &str,
- partition_count: u32,
- ) -> Result<(), BridgeError> {
- validate_kafka_topic_name("kafka_topic", kafka_topic)?;
- let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
- let stream_id = self.ensure_stream(stream_name).await?;
- self.ensure_topic(&stream_id, topic_name, kafka_topic, partition_count)
- .await?;
- Ok(())
- }
-
- /// Ensures the stream named `stream_name` exists, creating it if missing.
- ///
- /// `Identifier::named` - never `Identifier::try_from`/`FromStr` - because the latter parses
- /// an all-digit string as a numeric Iggy ID rather than a name. A stream or topic named e.g.
- /// `"42"` would otherwise resolve against the wrong resource on every call after the first:
- /// the first `ensure_stream_and_topic("42", ...)` creates a stream *named* `"42"`, but a
- /// second call would look it up *by ID* `42` instead, almost certainly finding nothing and
- /// breaking the "idempotent on repeated calls" guarantee.
- ///
- /// Returns the same *named* `Identifier` it was given, not the numeric id the SDK hands back
- /// from `get`/`create` - streams are backed by a recycled slab (`core/metadata`'s
- /// `stm/stream.rs`: freed keys are reused by the next created stream), so a numeric id
- /// captured here could point at a *different* stream by the time `ensure_topic` uses it, if
- /// this stream is deleted and recreated in between. The name has no such window.
- async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> {
- let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
- if with_request_timeout(self.client.get_stream(&identifier))
- .await?
- .is_some()
- {
- debug!("Iggy stream '{stream_name}' already exists");
- return Ok(identifier);
- }
-
- match with_request_timeout(self.client.create_stream(stream_name)).await {
- Ok(_created) => {
- info!("created Iggy stream '{stream_name}'");
- Ok(identifier)
- }
- Err(BridgeError::Iggy(IggyError::StreamNameAlreadyExists(_))) => {
- // Lost a create race - the name now exists regardless of who won it.
- Ok(identifier)
- }
- Err(err) => Err(err),
- }
- }
-
- /// Looks up (or creates) the topic named `topic_name` under `stream_id`.
- ///
- /// `Identifier::named`, not `Identifier::try_from` - the same numeric-name ambiguity
- /// [`Self::ensure_stream`]'s doc comment describes for stream names applies to topic names.
- ///
- /// `partition_count == 0` is accepted here as defense in depth, not the primary guard: the
- /// server allows it by design (`rewrite.rs`), and the `CreateTopics` stub already rejects it
- /// at the wire level (`protocol/responses.rs`) before any bridge call would be reachable.
- async fn ensure_topic(
- &self,
- stream_id: &Identifier,
- topic_name: &str,
- kafka_topic: &str,
- partition_count: u32,
- ) -> Result<(), BridgeError> {
- let identifier = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
- if let Some(existing) =
- with_request_timeout(self.client.get_topic(stream_id, &identifier)).await?
- {
- debug!("Iggy topic '{topic_name}' already exists");
- // ensure_topic's contract is "the topic has partition_count partitions afterward" -
- // a mismatch here means that's false. Returning Ok(()) anyway (even with a warn!)
- // would let two concurrent callers requesting different counts for the same topic
- // both believe they succeeded; growing partitions on the caller's behalf is also a
- // bigger decision (CreatePartitions has its own semantics) than this method should
- // make silently. Erring is the only response that keeps the postcondition honest.
- if existing.partitions_count != partition_count {
- return Err(BridgeError::PartitionCountMismatch {
- // The Kafka-side name a caller actually asked about, not `topic_name` - see
- // the identical note on `PartitionOutOfRange` in `high_watermark`.
- topic: kafka_topic.to_string(),
- existing: existing.partitions_count,
- requested: partition_count,
- });
- }
- return Ok(());
- }
-
- // message_expiry left at TopicCreateOptions::default() (None -> ServerDefault) means
- // never-expire (segment_cleaner.rs treats ServerDefault the same as NeverExpire), not
- // Kafka's own 7-day default - deliberate for now (imposing a retention policy is a product
- // decision this bridge shouldn't make unasked), but a real surprise for anyone repointing
- // a Kafka app that assumes bounded retention. Flagged in the README; revisit once there's
- // a way to configure it (env var, topic-mapping field) rather than hardcoding a number.
- let options = TopicCreateOptions {
- partitions_count: Some(partition_count),
- ..TopicCreateOptions::default()
- };
- match with_request_timeout(self.client.create_topic(stream_id, topic_name, &options)).await
- {
- Ok(created) => {
- info!("created Iggy topic '{topic_name}' with {partition_count} partitions");
- // Cheap: TopicDetails is already in hand, no extra round trip. `partitions_count`
- // is a hard argument to create_topic (Some(partition_count), never None), so the
- // server has no "resolve at admission" substitution to fall back on here - but
- // checking anyway, the same way the other two branches check their own
- // postcondition, means a future server-side clamp/cap fails loudly here instead
- // of this method silently reporting success under a broken contract.
- if created.partitions_count != partition_count {
- return Err(BridgeError::PartitionCountMismatch {
- topic: kafka_topic.to_string(),
- existing: created.partitions_count,
- requested: partition_count,
- });
- }
- Ok(())
- }
- // Lost a create race - re-verify by name rather than trusting the race outcome alone.
- // The winner may have created it with a different partition count than this call
- // requested, so this needs the same mismatch check the existing-topic branch above
- // makes - skipping it here would let two concurrent ensure_topic(N) / ensure_topic(M)
- // calls for the same topic both return Ok(()).
- //
- // Untested: reaching this arm needs a real concurrent second caller mid-race, and
- // `IggyBridge` holds a concrete `IggyClient` with no seam for a fake that returns
- // `TopicNameAlreadyExists` on demand. A test that spins up two real concurrent callers
- // would hit it only sometimes - flaky, and passing wouldn't prove this arm ran. Left
- // as a known gap until something introduces a client seam.
- Err(BridgeError::Iggy(IggyError::TopicNameAlreadyExists(_, _))) => {
- let existing = with_request_timeout(self.client.get_topic(stream_id, &identifier))
- .await?
- .ok_or_else(|| {
- BridgeError::Iggy(IggyError::TopicNameNotFound(
- topic_name.to_string(),
- stream_id.to_string(),
- ))
- })?;
- if existing.partitions_count != partition_count {
- return Err(BridgeError::PartitionCountMismatch {
- topic: kafka_topic.to_string(),
- existing: existing.partitions_count,
- requested: partition_count,
- });
- }
- Ok(())
- }
- Err(err) => Err(err),
- }
- }
-
- /// Returns the high watermark (one past the highest *committed* offset - see
- /// `Partition::offset_frontier`'s own definition) for every partition in `partitions` of the
- /// Iggy topic `kafka_topic` maps to, in one round trip.
- ///
- /// Takes `kafka_topic`, not raw Iggy stream/topic names, and resolves it through the same
- /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping) `ensure_stream_and_topic`
- /// uses - a caller (a future `ListOffsets` handler) only ever has the Kafka-side name, and a
- /// topic with a mapping override would silently query the wrong Iggy resource if this took
- /// Iggy-space names directly instead.
- ///
- /// One call to `get_topic`, not one per partition: a single Kafka `ListOffsets` request asks
- /// about many partitions of one topic at once (`ListOffsetsRequest.json`'s
- /// `Topics[] -> Partitions[]`), and `get_topic`'s decode already rebuilds and sorts the whole
- /// partition vector regardless of how many of them the caller wants
- /// (`wire_conversions.rs`) - a naive one-call-per-partition wrapper around a
- /// single-partition method would turn one Kafka request into N round trips and N times the
- /// decode work. Returns results in the same order as `partitions`.
- ///
- /// `i64`, not `u64`: the `ListOffsets` response field this exists to fill is `int64`
- /// (`ListOffsetsResponse.json`), so a future handler needs no cast at the wire boundary - the
- /// lossy conversion happens once, here (`Partition::current_offset` is `u64`; converted via
- /// `i64::try_from`, saturating to `i64::MAX` on the practically-unreachable overflow case
- /// rather than panicking or wrapping).
- ///
- /// `Partition::current_offset` is the offset of the *last written* message, not "next offset
- /// to produce" - confirmed against a live server (3 produced messages read back
- /// `current_offset == 2`). An empty partition has no last-written offset at all, so this
- /// needs a dedicated empty case rather than inferring it from `current_offset == 0` (also a
- /// fresh partition's default value, indistinguishable from "one message at offset 0").
- ///
- /// That empty case is `messages_count == 0 && current_offset == 0`, not `messages_count == 0`
- /// alone: retention cleanup decrements `messages_count`
- /// (`iggy_partition.rs::decrement_messages_count`) without rewinding `current_offset`, so a
- /// fully-purged but previously-produced-to partition would otherwise read as empty. This is
- /// still not exact: a partition whose *only* message, at offset 0, gets trimmed by the same
- /// retention pass reports `(messages_count, current_offset) == (0, 0)` too, so its watermark
- /// would read `1` then drop back to `0` - a real Kafka watermark never moves backward. Narrow
- /// (needs retention to fully empty a partition that only ever held one message) and not
- /// fixable client-side (`PartitionResponse` carries no frontier field to disambiguate it), but
- /// worth having on the record rather than only implicitly true of the check below.
- ///
- /// Not the same value as `Partition::mint_frontier` (the offset the *next* mint will take):
- /// after a crash-recovery reservation, the append point can sit above the committed frontier
- /// by the reservation's lease block, and neither value is on the wire to tell them apart. This
- /// is still correct for `ListOffsets` LATEST (defined in terms of the committed offset), but a
- /// future Produce handler must not use it to predict a base offset for a write in flight.
- ///
- /// Known remaining gap, not fixable client-side: `(0, 0)` is also what a stats-registry MISS
- /// reports (`responses.rs`'s `PartitionResponse` builder), indistinguishable on the wire from
- /// a genuinely empty partition. Narrow and self-healing - `partition_reconciler.rs`'s
- /// `settle_partition_stats` opens this only on the teardown-for-rebuild path, closing once the
- /// rebuild completes; deletes never open it. The real answer (`Partition::offset_frontier`)
- /// stays server-side and isn't in this response. The check matches the server's own
- /// `PartitionState::store_offset_range_error` condition, so it's not an invented heuristic - a
- /// future `ListOffsets` (`#3537`) built on this inherits the same blind spot.
- ///
- /// # Errors
- ///
- /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
- /// topic-naming rules. Returns [`BridgeError::Timeout`] if the call takes longer than
- /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] if the mapped stream doesn't exist (see
- /// the note above on which resource is actually missing). These are call-level failures -
- /// nothing about any individual partition could be resolved. A single out-of-range
- /// `partition` does not fail the whole call: it is reported per-partition (see the return
- /// type), so the partitions that did resolve are never discarded to report one that didn't.
- pub async fn high_watermarks(
- &self,
- kafka_topic: &str,
- partitions: &[u32],
- ) -> Result<Vec<(u32, Result<i64, BridgeError>)>, BridgeError> {
- validate_kafka_topic_name("kafka_topic", kafka_topic)?;
- let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
- let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
- let topic_id = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
- let details = with_request_timeout(self.client.get_topic(&stream_id, &topic_id))
- .await?
- .ok_or_else(|| {
- // A missing *stream* also makes get_topic return Ok(None) (responses.rs), so this
- // reports TopicNameNotFound even when the stream is what's actually gone - both
- // map to the same Kafka wire code either way, so only the log text is affected.
- BridgeError::Iggy(IggyError::TopicNameNotFound(
- topic_name.to_string(),
- stream_name.to_string(),
- ))
- })?;
-
- Ok(partitions
- .iter()
- .map(|&partition| {
- // `TryFrom<GetTopicResponse> for TopicDetails` (wire_conversions.rs) sorts
- // `partitions` by `id` on every decode, so a binary search is correct here, not
- // just faster than a linear scan - for a 1000-partition topic, the difference is
- // O(log n) vs O(n) probes per requested partition.
- let watermark = details
- .partitions
- .binary_search_by_key(&partition, |p| p.id)
- .map(|index| &details.partitions[index])
- .map_err(|_| BridgeError::PartitionOutOfRange {
- // The Kafka-side name a caller (a future ListOffsets handler) actually
- // asked about, not `topic_name` - a mapping override would otherwise
- // quote the wrong (Iggy-side) name back at a Kafka client that never
- // heard of it.
- topic: kafka_topic.to_string(),
- partition,
- partitions_count: details.partitions_count,
- })
- .map(|partition_details| {
- if partition_details.messages_count == 0
- && partition_details.current_offset == 0
- {
- 0
- } else {
- partition_details.current_offset.saturating_add(1)
- }
- })
- .map(|watermark| i64::try_from(watermark).unwrap_or(i64::MAX));
- (partition, watermark)
- })
- .collect())
- }
-
- /// Convenience wrapper around [`Self::high_watermarks`] for a single partition. See that
- /// method's doc for the full semantics - callers wanting more than one partition of the same
- /// topic should call it directly instead of this in a loop, to get its one-round-trip batching.
- ///
- /// # Errors
- ///
- /// See [`Self::high_watermarks`].
- ///
- /// # Panics
- ///
- /// Never in practice: [`Self::high_watermarks`] returns exactly one result per requested
- /// partition on `Ok`, and this always requests exactly one.
- pub async fn high_watermark(
- &self,
- kafka_topic: &str,
- partition: u32,
- ) -> Result<i64, BridgeError> {
- let (_, watermark) = self
- .high_watermarks(kafka_topic, &[partition])
- .await?
- .into_iter()
- .next()
- .expect(
- "high_watermarks returns exactly one result per requested partition, \
- and exactly one was requested",
- );
- watermark
- }
-}
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
similarity index 83%
copy from gateways/kafka/src/protocol/mod.rs
copy to gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index d7fd8a214..3a239e707 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-pub mod api;
-pub mod bounds_guard;
-pub mod header;
-pub mod responses;
+//! Fetch-side bridge calls.
+//!
+//! Empty until the Fetch handler lands. It exists so that work and the Produce
+//! work above never edit the same file.
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
new file mode 100644
index 000000000..e11079141
--- /dev/null
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -0,0 +1,187 @@
+// 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::future::Future;
+use std::time::Duration;
+
+use iggy::prelude::{AutoLogin, Client, Credentials, IggyClient, IggyClientBuilder, IggyError};
+use tracing::info;
+
+use crate::bridge::config::IggyBridgeConfig;
+use crate::bridge::error::BridgeError;
+
+mod fetch;
+mod offsets;
+mod produce;
+mod topics;
+
+/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
+///
+/// Not the SDK's own default (`TcpClientReconnectionConfig::default()` is `max_retries: None` -
+/// unlimited, one dial per second, forever). A Kafka client already retries at the wire-protocol
+/// level once a handler maps a bridge failure to a retriable error code; the bridge blocking a
+/// request task inside an unbounded internal reconnect loop would just add a second, invisible
+/// retry layer underneath that one instead of surfacing the failure so the mapped code can be
+/// sent.
+///
+/// This bounds the *count*, not the *wall-clock time*, of that inner retry loop - see
+/// [`REQUEST_TIMEOUT`] for the latter.
+const RECONNECTION_RETRIES: u32 = 3;
+
+/// Wall-clock ceiling [`with_request_timeout`] applies uniformly: to the initial `client.connect()`
+/// in [`IggyBridge::connect`] (including every attempt [`RECONNECTION_RETRIES`] makes internally),
+/// and to every call made after it succeeds (`get_stream`, `create_stream`, `get_topic`,
+/// `create_topic`, `shutdown`).
+///
+/// One constant, not two separately-named ones with the same value: both call sites bound the
+/// identical underlying hazard. `TcpClient::establish_bounded` only applies its own
+/// `FAILOVER_DIAL_TIMEOUT` (2s) when at least two failover candidates are configured
+/// (`tcp_client.rs`) - a bridge always configures exactly one address, so that guard never engages
+/// at either site, and the plain `TcpStream::connect` underneath has no deadline of its own.
+/// Against a firewall that drops SYN packets instead of refusing them, each dial attempt pays the
+/// kernel's own SYN-retry timeout (minutes, not seconds) rather than the `reconnection_interval`
+/// between attempts - a closed port (instant RST) never exercises this path, so the failure mode
+/// only shows up in production. 15s comfortably covers a slow-but-alive server's handshake (well
+/// above p99 login latency) while still failing well short of the pathological multi-minute case.
+///
+/// Known limitation, not closed by this constant: [`with_request_timeout`] cancels only the
+/// *caller's* wait, not the SDK's own work. `TcpClient::send_raw_vsr_attempt` (`tcp_client.rs`)
+/// writes, flushes and reads inside a detached `tokio::spawn` specifically so that dropping the
+/// awaiting future - exactly what this timeout does on expiry - cannot abort it mid-flight (that
+/// function's own "SAFETY: we run code holding the `stream` lock in a task so we can't be
+/// cancelled while holding the lock"). A call that times out here can leave that detached task
+/// still holding `IggyClient`'s single stream mutex for up to `RESPONSE_READ_TIMEOUT` (30s,
+/// `tcp_client.rs`) longer, queuing every other bridge call behind it. Raising this constant does
+/// not close the gap either: a mid-call reconnect (`send_raw_with_response`, `tcp_client.rs`)
+/// replays on a fresh `RESPONSE_READ_TIMEOUT` budget, and the reconnect dial itself goes through
+/// the same undead-lined `TcpStream::connect` this constant exists to bound - genuinely unbounded,
+/// so no finite value here can guarantee catching it. That same unboundedness is why this timeout
+/// cannot simply be dropped for post-connect calls either: [`IggyBridge`] holds one `IggyClient`
+/// with no pooling, so an unbounded reconnect dial with nothing here to stop it would wedge every
+/// later call on this bridge, not just the one that triggered it. Tolerable only because nothing
+/// calls this bridge from a live Kafka handler yet - closing it needs either a cooperatively
+/// cancellable SDK call or a deadline on the SDK's own reconnect dial, neither of which this
+/// bridge can add from the outside. Must be resolved before #3535/#3536 share this client across
+/// concurrent connections.
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Wraps a single Iggy client call in [`REQUEST_TIMEOUT`]. See that constant's doc for why every
+/// bridge method needs this, not just [`IggyBridge::connect`] - and for the cancellation gap this
+/// wrapper does not close.
+///
+/// On expiry, maps to [`BridgeError::Timeout`], not `IggyError::CannotEstablishConnection`: the
+/// SDK's write/read run in a detached task this timeout cannot abort (see [`REQUEST_TIMEOUT`]'s
+/// doc), so by the time this fires the request may already be on the wire, or already applied
+/// server-side - a genuinely unknown outcome, not a known connection failure. `BridgeError`'s own
+/// mapping already treats an unknown outcome (`IggyError::TransientNotCommitted`) as distinct from
+/// a known-safe-to-retry one for exactly this reason; a caller-side timeout is the same shape and
+/// must not borrow the known-safe code.
+async fn with_request_timeout<T>(
+ op: impl Future<Output = Result<T, IggyError>>,
+) -> Result<T, BridgeError> {
+ tokio::time::timeout(REQUEST_TIMEOUT, op)
+ .await
+ .map_err(|_elapsed| BridgeError::Timeout)?
+ .map_err(BridgeError::Iggy)
+}
+
+/// Owns one connected `IggyClient` and resolves Kafka topics against it.
+///
+/// Produce/Fetch handler wiring is a separate, later change (`#3535`/`#3536`) - this type is the
+/// shared plumbing those handlers will call into, exercised standalone here via its own tests and
+/// an integration test against a real `iggy-server`.
+///
+/// One `IggyClient`, shared across every Kafka connection this gateway serves - and the SDK's TCP
+/// transport is lockstep, one request in flight at a time (`tcp_client.rs`: "the connection is
+/// lockstep", its stream mutex held across write, flush, and read). Every concurrent Kafka
+/// connection ends up serialized behind whichever single Iggy request is in flight; the Kafka
+/// side's own connection limit does nothing to relieve this. No pooling exists yet - `close`
+/// already takes `self` by value, which anticipates an eventual `Arc`-shared bridge, but nothing
+/// wires that up today. Documented here and in the README rather than silently discovered under
+/// load once `#3535`/`#3536` land.
+pub struct IggyBridge {
+ client: IggyClient,
+ config: IggyBridgeConfig,
+}
+
+impl IggyBridge {
+ /// Connects to Iggy using `config` and authenticates.
+ ///
+ /// Builds the client through the SDK's fluent TCP builder rather than hand-assembling an
+ /// `iggy://user:pass@host` connection string: that string format splits on `@` then `:`, so a
+ /// password containing either character (`p@ss:word`) would be misparsed into a garbled
+ /// address instead of failing with a diagnosable config error. The fluent builder passes
+ /// `username`/`password` as already-separated fields, sidestepping the ambiguity entirely.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidConfig`] if `config.address` is empty. Returns
+ /// [`BridgeError::Timeout`] if connecting takes longer than `REQUEST_TIMEOUT` (an
+ /// unreachable-and-silently-dropping address, not just a refused one, is covered - see that
+ /// constant's doc). Returns [`BridgeError::Iggy`] if the address is malformed, the TCP
+ /// connection fails, or authentication is rejected - this is the boundary
+ /// [`BridgeError::to_kafka_error_code`] exists for: a handler calling this must map the error
+ /// to a wire response, never panic or unwrap, since an unreachable Iggy backend is an
+ /// expected runtime condition, not a bug.
+ pub async fn connect(config: IggyBridgeConfig) -> Result<Self, BridgeError> {
+ if config.address.trim().is_empty() {
+ return Err(BridgeError::InvalidConfig(
+ "Iggy address must not be empty".to_string(),
+ ));
+ }
+
+ let credentials =
+ Credentials::UsernamePassword(config.username.clone(), config.password.clone());
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address(config.address.clone())
+ .with_auto_sign_in(AutoLogin::Enabled(credentials))
+ .with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
+ .build()
+ .map_err(BridgeError::Iggy)?;
+ with_request_timeout(client.connect()).await?;
+ info!("Iggy bridge connected to {}", config.address);
+
+ Ok(Self { client, config })
+ }
+
+ /// Tears down the underlying Iggy client, including its background heartbeat task.
+ ///
+ /// Not `IggyClient::disconnect`: that only tears down the transport
+ /// (`TcpClient::disconnect_transport`) and never touches `heartbeat_handle` - only
+ /// `IggyClient`'s own `Drop` aborts that task (`client.rs`). A `disconnect`ed-but-not-dropped
+ /// bridge would keep heartbeating on a schedule, hit `NotConnected` (itself in the SDK's
+ /// retriable set), and reconnect plus re-authenticate using the credentials `connect`
+ /// configured, so the "closed" client silently comes back. `shutdown` sets
+ /// `ClientState::Shutdown`, which the heartbeat loop's next `ping` observes as
+ /// `IggyError::ClientShutdown` and self-terminates on, and which `sign_in_credentials` never
+ /// dials past.
+ ///
+ /// Takes `self` by value: `shutdown` is terminal (no reconnect is coming back from it), so
+ /// nothing legitimate is left to call on this bridge afterward. This does mean a bridge shared
+ /// via `Arc` cannot call this directly (`Arc::try_unwrap` first) - not a concern before
+ /// `#3535`/`#3536` wire an owning caller in.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::Timeout`] if it takes longer than `REQUEST_TIMEOUT`. Returns
+ /// [`BridgeError::Iggy`] if the underlying client reports a shutdown failure (e.g. the socket
+ /// was already in a state that rejects a clean shutdown).
+ pub async fn close(self) -> Result<(), BridgeError> {
+ with_request_timeout(self.client.shutdown()).await
+ }
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs
new file mode 100644
index 000000000..b199f7f65
--- /dev/null
+++ b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs
@@ -0,0 +1,176 @@
+// 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.
+
+//! Offset and high-watermark lookups.
+
+use iggy::prelude::{Identifier, IggyError, TopicClient};
+
+use super::{IggyBridge, with_request_timeout};
+use crate::bridge::error::BridgeError;
+use crate::bridge::topic_map::validate_kafka_topic_name;
+
+impl IggyBridge {
+ /// Returns the high watermark (one past the highest *committed* offset - see
+ /// `Partition::offset_frontier`'s own definition) for every partition in `partitions` of the
+ /// Iggy topic `kafka_topic` maps to, in one round trip.
+ ///
+ /// Takes `kafka_topic`, not raw Iggy stream/topic names, and resolves it through the same
+ /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping) `ensure_stream_and_topic`
+ /// uses - a caller (a future `ListOffsets` handler) only ever has the Kafka-side name, and a
+ /// topic with a mapping override would silently query the wrong Iggy resource if this took
+ /// Iggy-space names directly instead.
+ ///
+ /// One call to `get_topic`, not one per partition: a single Kafka `ListOffsets` request asks
+ /// about many partitions of one topic at once (`ListOffsetsRequest.json`'s
+ /// `Topics[] -> Partitions[]`), and `get_topic`'s decode already rebuilds and sorts the whole
+ /// partition vector regardless of how many of them the caller wants
+ /// (`wire_conversions.rs`) - a naive one-call-per-partition wrapper around a
+ /// single-partition method would turn one Kafka request into N round trips and N times the
+ /// decode work. Returns results in the same order as `partitions`.
+ ///
+ /// `i64`, not `u64`: the `ListOffsets` response field this exists to fill is `int64`
+ /// (`ListOffsetsResponse.json`), so a future handler needs no cast at the wire boundary - the
+ /// lossy conversion happens once, here (`Partition::current_offset` is `u64`; converted via
+ /// `i64::try_from`, saturating to `i64::MAX` on the practically-unreachable overflow case
+ /// rather than panicking or wrapping).
+ ///
+ /// `Partition::current_offset` is the offset of the *last written* message, not "next offset
+ /// to produce" - confirmed against a live server (3 produced messages read back
+ /// `current_offset == 2`). An empty partition has no last-written offset at all, so this
+ /// needs a dedicated empty case rather than inferring it from `current_offset == 0` (also a
+ /// fresh partition's default value, indistinguishable from "one message at offset 0").
+ ///
+ /// That empty case is `messages_count == 0 && current_offset == 0`, not `messages_count == 0`
+ /// alone: retention cleanup decrements `messages_count`
+ /// (`iggy_partition.rs::decrement_messages_count`) without rewinding `current_offset`, so a
+ /// fully-purged but previously-produced-to partition would otherwise read as empty. This is
+ /// still not exact: a partition whose *only* message, at offset 0, gets trimmed by the same
+ /// retention pass reports `(messages_count, current_offset) == (0, 0)` too, so its watermark
+ /// would read `1` then drop back to `0` - a real Kafka watermark never moves backward. Narrow
+ /// (needs retention to fully empty a partition that only ever held one message) and not
+ /// fixable client-side (`PartitionResponse` carries no frontier field to disambiguate it), but
+ /// worth having on the record rather than only implicitly true of the check below.
+ ///
+ /// Not the same value as `Partition::mint_frontier` (the offset the *next* mint will take):
+ /// after a crash-recovery reservation, the append point can sit above the committed frontier
+ /// by the reservation's lease block, and neither value is on the wire to tell them apart. This
+ /// is still correct for `ListOffsets` LATEST (defined in terms of the committed offset), but a
+ /// future Produce handler must not use it to predict a base offset for a write in flight.
+ ///
+ /// Known remaining gap, not fixable client-side: `(0, 0)` is also what a stats-registry MISS
+ /// reports (`responses.rs`'s `PartitionResponse` builder), indistinguishable on the wire from
+ /// a genuinely empty partition. Narrow and self-healing - `partition_reconciler.rs`'s
+ /// `settle_partition_stats` opens this only on the teardown-for-rebuild path, closing once the
+ /// rebuild completes; deletes never open it. The real answer (`Partition::offset_frontier`)
+ /// stays server-side and isn't in this response. The check matches the server's own
+ /// `PartitionState::store_offset_range_error` condition, so it's not an invented heuristic - a
+ /// future `ListOffsets` (`#3537`) built on this inherits the same blind spot.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
+ /// topic-naming rules. Returns [`BridgeError::Timeout`] if the call takes longer than
+ /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] if the mapped stream doesn't exist (see
+ /// the note above on which resource is actually missing). These are call-level failures -
+ /// nothing about any individual partition could be resolved. A single out-of-range
+ /// `partition` does not fail the whole call: it is reported per-partition (see the return
+ /// type), so the partitions that did resolve are never discarded to report one that didn't.
+ pub async fn high_watermarks(
+ &self,
+ kafka_topic: &str,
+ partitions: &[u32],
+ ) -> Result<Vec<(u32, Result<i64, BridgeError>)>, BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ let topic_id = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ let details = with_request_timeout(self.client.get_topic(&stream_id, &topic_id))
+ .await?
+ .ok_or_else(|| {
+ // A missing *stream* also makes get_topic return Ok(None) (responses.rs), so this
+ // reports TopicNameNotFound even when the stream is what's actually gone - both
+ // map to the same Kafka wire code either way, so only the log text is affected.
+ BridgeError::Iggy(IggyError::TopicNameNotFound(
+ topic_name.to_string(),
+ stream_name.to_string(),
+ ))
+ })?;
+
+ Ok(partitions
+ .iter()
+ .map(|&partition| {
+ // `TryFrom<GetTopicResponse> for TopicDetails` (wire_conversions.rs) sorts
+ // `partitions` by `id` on every decode, so a binary search is correct here, not
+ // just faster than a linear scan - for a 1000-partition topic, the difference is
+ // O(log n) vs O(n) probes per requested partition.
+ let watermark = details
+ .partitions
+ .binary_search_by_key(&partition, |p| p.id)
+ .map(|index| &details.partitions[index])
+ .map_err(|_| BridgeError::PartitionOutOfRange {
+ // The Kafka-side name a caller (a future ListOffsets handler) actually
+ // asked about, not `topic_name` - a mapping override would otherwise
+ // quote the wrong (Iggy-side) name back at a Kafka client that never
+ // heard of it.
+ topic: kafka_topic.to_string(),
+ partition,
+ partitions_count: details.partitions_count,
+ })
+ .map(|partition_details| {
+ if partition_details.messages_count == 0
+ && partition_details.current_offset == 0
+ {
+ 0
+ } else {
+ partition_details.current_offset.saturating_add(1)
+ }
+ })
+ .map(|watermark| i64::try_from(watermark).unwrap_or(i64::MAX));
+ (partition, watermark)
+ })
+ .collect())
+ }
+
+ /// Convenience wrapper around [`Self::high_watermarks`] for a single partition. See that
+ /// method's doc for the full semantics - callers wanting more than one partition of the same
+ /// topic should call it directly instead of this in a loop, to get its one-round-trip batching.
+ ///
+ /// # Errors
+ ///
+ /// See [`Self::high_watermarks`].
+ ///
+ /// # Panics
+ ///
+ /// Never in practice: [`Self::high_watermarks`] returns exactly one result per requested
+ /// partition on `Ok`, and this always requests exactly one.
+ pub async fn high_watermark(
+ &self,
+ kafka_topic: &str,
+ partition: u32,
+ ) -> Result<i64, BridgeError> {
+ let (_, watermark) = self
+ .high_watermarks(kafka_topic, &[partition])
+ .await?
+ .into_iter()
+ .next()
+ .expect(
+ "high_watermarks returns exactly one result per requested partition, \
+ and exactly one was requested",
+ );
+ watermark
+ }
+}
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
similarity index 83%
copy from gateways/kafka/src/protocol/mod.rs
copy to gateways/kafka/src/bridge/iggy_bridge/produce.rs
index d7fd8a214..55681ee95 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-pub mod api;
-pub mod bounds_guard;
-pub mod header;
-pub mod responses;
+//! Produce-side bridge calls.
+//!
+//! Empty until the Produce handler lands. It exists so that work and the Fetch
+//! work below never edit the same file.
diff --git a/gateways/kafka/src/bridge/iggy_bridge/topics.rs b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
new file mode 100644
index 000000000..395f6f27d
--- /dev/null
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@ -0,0 +1,220 @@
+// 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.
+
+//! Stream and topic provisioning.
+
+use iggy::prelude::{Identifier, IggyError, StreamClient, TopicClient, TopicCreateOptions};
+use tracing::{debug, info};
+
+use super::{IggyBridge, with_request_timeout};
+use crate::bridge::error::BridgeError;
+use crate::bridge::topic_map::validate_kafka_topic_name;
+
+impl IggyBridge {
+ /// Ensures the Iggy stream and topic backing `kafka_topic` exist, creating either or both if
+ /// missing. Resolves `kafka_topic` through the configured [`TopicMapping`](crate::bridge::topic_map::TopicMapping).
+ ///
+ /// Idempotent when repeated with the *same* `partition_count`: a `get` before each `create`
+ /// means calling this twice for the same topic is a no-op the second time, and a
+ /// `NameAlreadyExists` race from a concurrent caller creating the same stream/topic between
+ /// this call's `get` and `create` is treated as success, not an error - the desired end state
+ /// (it exists) is what idempotency actually promises, not that this call was the one that
+ /// created it. A *different* `partition_count` against an already-existing topic is not
+ /// idempotent - see [`BridgeError::PartitionCountMismatch`].
+ ///
+ /// Ensures the stream before the topic, so a topic-creation failure (a different
+ /// `partition_count`, a transient error) can leave a stream that now exists with no topic in
+ /// it yet - a retry heals this (idempotent on the stream half too), and no rollback is
+ /// attempted: `TopicMapping::resolve` sends every *unmapped* Kafka topic to the same
+ /// `default_stream`, so deleting a stream on a topic-creation failure risks deleting another
+ /// topic's data that happens to share it, and this call has no way to tell whether it was the
+ /// one that created the stream in the first place.
+ ///
+ /// `partition_count` is `u32`, so it cannot carry Kafka's own `CreateTopics` sentinel
+ /// (`num_partitions == -1`, KIP-464 "use the broker default" - `protocol/responses.rs`
+ /// already accepts that sentinel at the wire-validation layer). Resolving *what* the broker
+ /// default should be for this bridge is a decision for whichever of `#3535`/`#3536` first
+ /// calls this with a real Kafka request in hand, not one to invent here ahead of that need.
+ ///
+ /// No caching: every call pays a `get_stream` and a `get_topic` (two round trips once both
+ /// already exist), even for a topic this same bridge already confirmed a moment ago. A cache
+ /// keyed on `kafka_topic` would remove that cost, but would also have to answer "how does a
+ /// cache entry ever get invalidated" - the topic being deleted and recreated with a different
+ /// partition count out from under a stale cache entry is exactly
+ /// `ensure_topic_targets_the_streams_live_incarnation_after_a_delete_and_recreate`'s own
+ /// scenario, and a naive cache breaks that guarantee to save two round trips. Whatever wires
+ /// this into `#3535`/`#3536` should call it once per topic and remember that it did, rather
+ /// than once per Produce/Fetch - the cost belongs at the call site's discretion, not hidden
+ /// (and potentially made wrong) inside this method.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
+ /// topic-naming rules. Returns [`BridgeError::Timeout`] if a call takes longer than
+ /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] for connectivity/auth failures. Returns
+ /// [`BridgeError::PartitionCountMismatch`] if the topic already exists with a different
+ /// partition count than `partition_count`.
+ pub async fn ensure_stream_and_topic(
+ &self,
+ kafka_topic: &str,
+ partition_count: u32,
+ ) -> Result<(), BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id = self.ensure_stream(stream_name).await?;
+ self.ensure_topic(&stream_id, topic_name, kafka_topic, partition_count)
+ .await?;
+ Ok(())
+ }
+
+ /// Ensures the stream named `stream_name` exists, creating it if missing.
+ ///
+ /// `Identifier::named` - never `Identifier::try_from`/`FromStr` - because the latter parses
+ /// an all-digit string as a numeric Iggy ID rather than a name. A stream or topic named e.g.
+ /// `"42"` would otherwise resolve against the wrong resource on every call after the first:
+ /// the first `ensure_stream_and_topic("42", ...)` creates a stream *named* `"42"`, but a
+ /// second call would look it up *by ID* `42` instead, almost certainly finding nothing and
+ /// breaking the "idempotent on repeated calls" guarantee.
+ ///
+ /// Returns the same *named* `Identifier` it was given, not the numeric id the SDK hands back
+ /// from `get`/`create` - streams are backed by a recycled slab (`core/metadata`'s
+ /// `stm/stream.rs`: freed keys are reused by the next created stream), so a numeric id
+ /// captured here could point at a *different* stream by the time `ensure_topic` uses it, if
+ /// this stream is deleted and recreated in between. The name has no such window.
+ async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> {
+ let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ if with_request_timeout(self.client.get_stream(&identifier))
+ .await?
+ .is_some()
+ {
+ debug!("Iggy stream '{stream_name}' already exists");
+ return Ok(identifier);
+ }
+
+ match with_request_timeout(self.client.create_stream(stream_name)).await {
+ Ok(_created) => {
+ info!("created Iggy stream '{stream_name}'");
+ Ok(identifier)
+ }
+ Err(BridgeError::Iggy(IggyError::StreamNameAlreadyExists(_))) => {
+ // Lost a create race - the name now exists regardless of who won it.
+ Ok(identifier)
+ }
+ Err(err) => Err(err),
+ }
+ }
+
+ /// Looks up (or creates) the topic named `topic_name` under `stream_id`.
+ ///
+ /// `Identifier::named`, not `Identifier::try_from` - the same numeric-name ambiguity
+ /// [`Self::ensure_stream`]'s doc comment describes for stream names applies to topic names.
+ ///
+ /// `partition_count == 0` is accepted here as defense in depth, not the primary guard: the
+ /// server allows it by design (`rewrite.rs`), and the `CreateTopics` stub already rejects it
+ /// at the wire level (`protocol/responses.rs`) before any bridge call would be reachable.
+ async fn ensure_topic(
+ &self,
+ stream_id: &Identifier,
+ topic_name: &str,
+ kafka_topic: &str,
+ partition_count: u32,
+ ) -> Result<(), BridgeError> {
+ let identifier = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ if let Some(existing) =
+ with_request_timeout(self.client.get_topic(stream_id, &identifier)).await?
+ {
+ debug!("Iggy topic '{topic_name}' already exists");
+ // ensure_topic's contract is "the topic has partition_count partitions afterward" -
+ // a mismatch here means that's false. Returning Ok(()) anyway (even with a warn!)
+ // would let two concurrent callers requesting different counts for the same topic
+ // both believe they succeeded; growing partitions on the caller's behalf is also a
+ // bigger decision (CreatePartitions has its own semantics) than this method should
+ // make silently. Erring is the only response that keeps the postcondition honest.
+ if existing.partitions_count != partition_count {
+ return Err(BridgeError::PartitionCountMismatch {
+ // The Kafka-side name a caller actually asked about, not `topic_name` - see
+ // the identical note on `PartitionOutOfRange` in `high_watermark`.
+ topic: kafka_topic.to_string(),
+ existing: existing.partitions_count,
+ requested: partition_count,
+ });
+ }
+ return Ok(());
+ }
+
+ // message_expiry left at TopicCreateOptions::default() (None -> ServerDefault) means
+ // never-expire (segment_cleaner.rs treats ServerDefault the same as NeverExpire), not
+ // Kafka's own 7-day default - deliberate for now (imposing a retention policy is a product
+ // decision this bridge shouldn't make unasked), but a real surprise for anyone repointing
+ // a Kafka app that assumes bounded retention. Flagged in the README; revisit once there's
+ // a way to configure it (env var, topic-mapping field) rather than hardcoding a number.
+ let options = TopicCreateOptions {
+ partitions_count: Some(partition_count),
+ ..TopicCreateOptions::default()
+ };
+ match with_request_timeout(self.client.create_topic(stream_id, topic_name, &options)).await
+ {
+ Ok(created) => {
+ info!("created Iggy topic '{topic_name}' with {partition_count} partitions");
+ // Cheap: TopicDetails is already in hand, no extra round trip. `partitions_count`
+ // is a hard argument to create_topic (Some(partition_count), never None), so the
+ // server has no "resolve at admission" substitution to fall back on here - but
+ // checking anyway, the same way the other two branches check their own
+ // postcondition, means a future server-side clamp/cap fails loudly here instead
+ // of this method silently reporting success under a broken contract.
+ if created.partitions_count != partition_count {
+ return Err(BridgeError::PartitionCountMismatch {
+ topic: kafka_topic.to_string(),
+ existing: created.partitions_count,
+ requested: partition_count,
+ });
+ }
+ Ok(())
+ }
+ // Lost a create race - re-verify by name rather than trusting the race outcome alone.
+ // The winner may have created it with a different partition count than this call
+ // requested, so this needs the same mismatch check the existing-topic branch above
+ // makes - skipping it here would let two concurrent ensure_topic(N) / ensure_topic(M)
+ // calls for the same topic both return Ok(()).
+ //
+ // Untested: reaching this arm needs a real concurrent second caller mid-race, and
+ // `IggyBridge` holds a concrete `IggyClient` with no seam for a fake that returns
+ // `TopicNameAlreadyExists` on demand. A test that spins up two real concurrent callers
+ // would hit it only sometimes - flaky, and passing wouldn't prove this arm ran. Left
+ // as a known gap until something introduces a client seam.
+ Err(BridgeError::Iggy(IggyError::TopicNameAlreadyExists(_, _))) => {
+ let existing = with_request_timeout(self.client.get_topic(stream_id, &identifier))
+ .await?
+ .ok_or_else(|| {
+ BridgeError::Iggy(IggyError::TopicNameNotFound(
+ topic_name.to_string(),
+ stream_id.to_string(),
+ ))
+ })?;
+ if existing.partitions_count != partition_count {
+ return Err(BridgeError::PartitionCountMismatch {
+ topic: kafka_topic.to_string(),
+ existing: existing.partitions_count,
+ requested: partition_count,
+ });
+ }
+ Ok(())
+ }
+ Err(err) => Err(err),
+ }
+ }
+}
diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs
index 333e34968..6cd0dc64a 100644
--- a/gateways/kafka/src/main.rs
+++ b/gateways/kafka/src/main.rs
@@ -17,13 +17,14 @@
use std::fmt::Display;
use std::str::FromStr;
+use std::sync::Arc;
use std::time::Duration;
use tokio::signal;
use tokio::sync::{Semaphore, broadcast};
-use tracing::warn;
+use tracing::{info, warn};
-use iggy_gateway_kafka::bridge::IggyBridgeConfig;
+use iggy_gateway_kafka::bridge::{IggyBridge, IggyBridgeConfig};
use iggy_gateway_kafka::server::{bind_listener, init_tracing};
use iggy_gateway_kafka::{GatewayConfig, KafkaGateway};
@@ -35,10 +36,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let _tracing_guard = init_tracing();
let config = load_config()?;
+ let bridge = connect_bridge().await?;
let listener = bind_listener(&config.bind_addr)
.map_err(|e| format!("failed to bind {}: {e}", config.bind_addr))?;
- let server = KafkaGateway::new(config);
+ let server = KafkaGateway::new(config).with_bridge(bridge);
let (tx, rx) = broadcast::channel(1);
let mut server_task = tokio::spawn(async move { server.run(listener, rx).await });
@@ -75,16 +77,16 @@ const KNOWN_KAFKA_ENV_VARS: &[&str] = &[
"IGGY_KAFKA_READ_TIMEOUT_SECS",
"IGGY_KAFKA_WRITE_TIMEOUT_SECS",
"IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS",
+ "IGGY_KAFKA_BRIDGE_ENABLED",
];
/// Rejects any `IGGY_KAFKA_*` env var not in [`KNOWN_KAFKA_ENV_VARS`] or
/// [`IggyBridgeConfig::KNOWN_ENV_VARS`] - a typo (e.g. `IGGY_KAFKA_BIN_ADDR`) would otherwise be
/// silently ignored: `core/configs`' central provider skips the whole `IGGY_KAFKA_` prefix, and
/// this crate's own `env_var()` only ever looks up exact known names, so nothing reads the
-/// misspelled var and nothing warns either. Checking both lists (not `IggyBridgeConfig`'s
-/// wiring into `main`) is deliberate: a user who exports a bridge var ahead of
-/// `#3535`/`#3536` landing must not see a spurious "unknown env var" rejection for a name this
-/// crate already recognizes.
+/// misspelled var and nothing warns either. Checking both lists is deliberate: a user who
+/// exports a bridge var while `IGGY_KAFKA_BRIDGE_ENABLED` is off must not see a spurious
+/// "unknown env var" rejection for a name this crate already recognizes.
fn reject_unknown_kafka_env_vars() -> Result<(), String> {
for (key, _) in std::env::vars() {
if key.starts_with("IGGY_KAFKA_")
@@ -100,26 +102,46 @@ fn reject_unknown_kafka_env_vars() -> Result<(), String> {
Ok(())
}
-/// Warns if any bridge-specific env var is set, since `main` doesn't read `IggyBridgeConfig` or
-/// call `IggyBridge::connect` yet (`#3535`/`#3536`).
+/// Connects the Iggy bridge when `IGGY_KAFKA_BRIDGE_ENABLED` is true, otherwise returns `None`
+/// and every handler keeps answering with its stub.
///
-/// `reject_unknown_kafka_env_vars`'s own accepted-but-unread carve-out for these vars exists so a
-/// user exporting them ahead of that wiring landing isn't told the name is unrecognized - but that
-/// silence is exactly what the guard's own doc comment warns an unread var would otherwise cause.
-/// This closes that gap: still accepted, no longer silent.
-fn warn_on_unused_bridge_env_vars() {
- let set: Vec<&str> = IggyBridgeConfig::KNOWN_ENV_VARS
- .iter()
- .copied()
- .filter(|var| std::env::var(var).is_ok())
- .collect();
- if !set.is_empty() {
- warn!(
- "{} set but not yet used: the Kafka <-> Iggy bridge isn't wired into the running \
- gateway until #3535/#3536 land",
- set.join(", ")
- );
+/// Off by default so that landing one API at a time never changes what an operator who has not
+/// asked for the bridge sees. A failure to connect is fatal rather than a downgrade to stubs: an
+/// operator who turned the bridge on wants to know it is unreachable at startup, not to discover
+/// it from `NOT_LEADER_OR_FOLLOWER` on every produce.
+async fn connect_bridge() -> Result<Option<Arc<IggyBridge>>, String> {
+ if !bridge_enabled()? {
+ let set: Vec<&str> = IggyBridgeConfig::KNOWN_ENV_VARS
+ .iter()
+ .copied()
+ .filter(|var| std::env::var(var).is_ok())
+ .collect();
+ if !set.is_empty() {
+ warn!(
+ "{} set but IGGY_KAFKA_BRIDGE_ENABLED is not true: the gateway is answering with \
+ stub responses and reads none of them",
+ set.join(", ")
+ );
+ }
+ return Ok(None);
}
+ let config = IggyBridgeConfig::from_env().map_err(|e| format!("invalid bridge config: {e}"))?;
+ let bridge = IggyBridge::connect(config)
+ .await
+ .map_err(|e| format!("failed to connect the Iggy bridge: {e}"))?;
+ info!("Iggy bridge connected");
+ Ok(Some(Arc::new(bridge)))
+}
+
+/// `IGGY_KAFKA_BRIDGE_ENABLED`, rejecting anything that is neither `true` nor `false`.
+///
+/// A typo would otherwise read as "off" and leave an operator who asked for the bridge staring
+/// at stub responses with nothing in the log to explain them.
+fn bridge_enabled() -> Result<bool, String> {
+ env_var("IGGY_KAFKA_BRIDGE_ENABLED").map_or(Ok(false), |raw| {
+ raw.parse()
+ .map_err(|e| format!("invalid IGGY_KAFKA_BRIDGE_ENABLED `{raw}`: {e}"))
+ })
}
/// Build [`GatewayConfig`] from `IGGY_KAFKA_*` env vars, rejecting values that would silently
@@ -127,7 +149,6 @@ fn warn_on_unused_bridge_env_vars() {
/// connection, a connection cap above `Semaphore::MAX_PERMITS` panics at startup).
fn load_config() -> Result<GatewayConfig, String> {
reject_unknown_kafka_env_vars()?;
- warn_on_unused_bridge_env_vars();
let mut config = GatewayConfig::default();
if let Some(bind_addr) = env_var("IGGY_KAFKA_BIND_ADDR") {
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index b246b60ca..4923933e9 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -15,25 +15,13 @@
// 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 std::sync::Arc;
-use crate::error::{KafkaProtocolError, Result};
-use crate::protocol::bounds_guard::{
- validate_api_versions_shape, validate_create_topics_shape, validate_fetch_shape,
- validate_list_offsets_shape, validate_metadata_shape, validate_produce_shape,
-};
-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,
+use bytes::Bytes;
+
+use crate::bridge::IggyBridge;
+use crate::protocol::handlers::{
+ api_versions, create_topics, dispatch, fetch, list_offsets, metadata, produce,
};
pub const API_KEY_PRODUCE: i16 = 0;
@@ -148,36 +136,12 @@ pub struct ApiVersionRange {
}
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,
- },
+ produce::RANGE,
+ fetch::RANGE,
+ list_offsets::RANGE,
+ metadata::RANGE,
+ api_versions::RANGE,
+ create_topics::RANGE,
];
#[must_use]
@@ -185,20 +149,59 @@ pub fn supported_api_ranges() -> &'static [ApiVersionRange] {
SUPPORTED_RANGES
}
-/// Default `max_frame_size` used by [`handle_request`] - the ~150 direct call sites across this
+/// Everything a handler needs that outlives one request.
+///
+/// `bridge` is `None` until `IGGY_KAFKA_BRIDGE_ENABLED` turns it on, and a handler that finds
+/// `None` answers with the stub it answers with today. That is what keeps foundation behavior
+/// the default while [`#3535`](https://github.com/apache/iggy/issues/3535) and
+/// [`#3536`](https://github.com/apache/iggy/issues/3536) wire one API at a time.
+///
+/// One `IggyBridge` is one `IggyClient`, whose TCP transport is lockstep, so every Kafka
+/// connection sharing this serializes behind whichever Iggy request is in flight. Sharing it
+/// through an `Arc` neither causes that nor worsens it (see the README's "Concurrency ceiling"),
+/// but a connection pool is owed before Produce and Fetch put it on a hot path.
+pub struct GatewayState {
+ pub broker: BrokerAdvertise,
+ pub bridge: Option<Arc<IggyBridge>>,
+ pub max_frame_size: usize,
+}
+
+impl GatewayState {
+ #[must_use]
+ pub const fn new(
+ broker: BrokerAdvertise,
+ bridge: Option<Arc<IggyBridge>>,
+ max_frame_size: usize,
+ ) -> Self {
+ Self {
+ broker,
+ bridge,
+ max_frame_size,
+ }
+ }
+
+ /// State with no bridge, so every handler takes its stub path.
+ #[must_use]
+ pub const fn stub(broker: BrokerAdvertise, max_frame_size: usize) -> Self {
+ Self::new(broker, None, max_frame_size)
+ }
+}
+
+/// Default `max_frame_size` used by [`handle_request`] - the direct call sites across this
/// crate's test suite that don't care about the response-size guard specifically. Production
/// traffic goes through [`handle_request_bounded`] instead (see `server.rs`'s call site), with
/// the connection's actual configured `max_frame_size`.
const DEFAULT_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
/// Handles one decoded request frame and returns how the connection should proceed.
-pub fn handle_request(
+pub async fn handle_request(
api_key: i16,
api_version: i16,
body: Bytes,
broker: &BrokerAdvertise,
) -> HandleOutcome {
- handle_request_bounded(api_key, api_version, body, broker, DEFAULT_MAX_FRAME_SIZE)
+ let state = GatewayState::stub(broker.clone(), DEFAULT_MAX_FRAME_SIZE);
+ handle_request_bounded(&state, api_key, api_version, body).await
}
/// Same as [`handle_request`], but rejects a request whose declared array/string lengths project
@@ -208,300 +211,13 @@ pub fn handle_request(
/// docs for the CPU/memory amplification this closes (a request within the old element budget
/// alone could still produce a multi-megabyte response from a single synchronous, non-yielding
/// call).
-pub fn handle_request_bounded(
- api_key: i16,
- api_version: i16,
- body: Bytes,
- broker: &BrokerAdvertise,
- max_frame_size: usize,
-) -> HandleOutcome {
- if api_key == API_KEY_PRODUCE {
- return handle_produce_request(api_version, body, max_frame_size);
- }
- handle_other_request(api_key, api_version, body, broker, max_frame_size)
-}
-
-/// 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)
-}
-
-/// Runs `validate` (see [`crate::protocol::bounds_guard`]) before [`decode_exhaustive`].
-///
-/// `kafka_protocol` validates wire-declared array/string lengths as non-negative but never
-/// against the bytes actually remaining in the frame, so a tiny frame declaring a huge count
-/// drives an allocation attempt that aborts the whole process (`handle_alloc_error`, not a
-/// panic - uncatchable, and it kills every connection, not just the offending one). `validate`
-/// rejects that class of frame first, on a walk that never allocates a collection.
-fn decode_guarded<T: Decodable>(
- version: i16,
- body: Bytes,
- validate: impl FnOnce(i16, &Bytes) -> Result<()>,
-) -> Result<T> {
- validate(version, &body)?;
- decode_exhaustive(version, body)
-}
-
-/// 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, max_frame_size: usize) -> 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). Fail-closed on a missing row
- // (i16::MIN, not i16::MAX): `handle_request` dispatches Produce on a hard-coded `api_key ==`
- // check, not from this table, so if a future edit ever drops the Produce row to disable the
- // API, a fail-open default here would leave Produce v0-2 acks=0 silently accepted on an API
- // the operator believes is off - unwrap_or(i16::MAX) is not the sound default the sibling
- // lookup at `unsupported_version_response` uses (`map_or(0, ...)`, i.e. fail-closed).
- if api_version > supported_max_version(API_KEY_PRODUCE).unwrap_or(i16::MIN) {
- 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_guarded::<ProduceRequest>(api_version, body, |v, b| {
- validate_produce_shape(v, b, max_frame_size)
- }) {
- // 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,
- // `api_version` is always in `[3, supported_max]` here: the `< 3` case returned above,
- // and the `> max` case returned at the top of this function - so it is always within
- // `SUPPORTED_RANGES`' Produce row and an `is_supported_version` re-check can never fail.
- Ok(req) => 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.
- // debug!, not warn!: the body is attacker-controlled, not operator-actionable, and
- // a client looping malformed bodies on one connection (never disconnected - this
- // arm returns NoResponse) has no rate limit here.
- tracing::debug!(%error, "failed to decode Produce request (no response)");
- HandleOutcome::NoResponse
- }
- }
-}
-
-fn handle_other_request(
- api_key: i16,
- api_version: i16,
- body: Bytes,
- broker: &BrokerAdvertise,
- max_frame_size: usize,
-) -> HandleOutcome {
- match api_key {
- API_KEY_API_VERSIONS => handle_api_versions(api_version, body),
- API_KEY_METADATA => handle_metadata(api_version, body, broker, max_frame_size),
- API_KEY_FETCH => handle_versioned_request(
- API_KEY_FETCH,
- api_version,
- body,
- |v, b| {
- decode_guarded::<FetchRequest>(v, b, |v, b| {
- validate_fetch_shape(v, b, max_frame_size)
- })
- },
- encode_fetch_response,
- encode_fetch_error_response,
- "Fetch",
- ),
- API_KEY_LIST_OFFSETS => handle_versioned_request(
- API_KEY_LIST_OFFSETS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<ListOffsetsRequest>(v, b, |v, b| {
- validate_list_offsets_shape(v, b, max_frame_size)
- })
- },
- encode_list_offsets_response,
- encode_list_offsets_error_response,
- "ListOffsets",
- ),
- API_KEY_CREATE_TOPICS => handle_versioned_request(
- API_KEY_CREATE_TOPICS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
- validate_create_topics_shape(v, b, max_frame_size)
- })
- },
- encode_create_topics_response,
- encode_create_topics_error_response,
- "CreateTopics",
- ),
- // Unknown API key: no api-specific response schema exists, so any body we send is
- // misparsed by the client against the schema it expected. Close is unambiguous.
- _ => HandleOutcome::Close,
- }
-}
-
-fn handle_api_versions(api_version: i16, body: Bytes) -> HandleOutcome {
- if !is_supported_version(API_KEY_API_VERSIONS, api_version) {
- // KIP-511: reply with v0 when the requested version is not understood.
- return respond_or_close(
- encode_api_versions_response(0, ERROR_UNSUPPORTED_VERSION),
- "ApiVersions",
- );
- }
- match decode_guarded::<ApiVersionsRequest>(api_version, body, validate_api_versions_shape) {
- Ok(_) => respond_or_close(
- encode_api_versions_response(api_version, ERROR_NONE),
- "ApiVersions",
- ),
- Err(error) => {
- // debug!, not warn!: attacker-controlled, not operator-actionable (see the same
- // note on the Produce decode-failure arm above).
- tracing::debug!(%error, "failed to decode ApiVersions request");
- respond_or_close(
- encode_api_versions_response(api_version, ERROR_INVALID_REQUEST),
- "ApiVersions",
- )
- }
- }
-}
-
-fn handle_metadata(
- api_version: i16,
- body: Bytes,
- broker: &BrokerAdvertise,
- max_frame_size: usize,
-) -> HandleOutcome {
- if !is_supported_version(API_KEY_METADATA, api_version) {
- // Clamping the response to the supported max 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 = supported_max_version(API_KEY_METADATA),
- "Metadata version unsupported; closing connection"
- );
- return HandleOutcome::Close;
- }
- match decode_metadata_topics(api_version, body, max_frame_size) {
- Ok(topics) => respond_or_close(
- encode_metadata_response(api_version, &topics, broker, ERROR_NONE),
- "Metadata",
- ),
- Err(error) => {
- // Metadata has no top-level error field; a malformed body cannot carry
- // INVALID_REQUEST in a version-correct way for every client. Close.
- // debug!, not warn!: attacker-controlled, not operator-actionable.
- tracing::debug!(
- %error,
- api_version,
- "Failed to decode Metadata request; closing connection"
- );
- HandleOutcome::Close
- }
- }
-}
-
-fn handle_versioned_request<T>(
+pub async fn handle_request_bounded(
+ state: &GatewayState,
api_key: i16,
api_version: i16,
body: Bytes,
- decode: impl FnOnce(i16, Bytes) -> Result<T>,
- encode_ok: impl FnOnce(i16, &T) -> Result<Bytes>,
- encode_err: impl Fn(i16, i16) -> Result<Bytes>,
- api_name: &str,
-) -> HandleOutcome {
- if is_supported_version(api_key, api_version) {
- match decode(api_version, body) {
- Ok(req) => respond_or_close(encode_ok(api_version, &req), api_name),
- Err(error) => {
- // debug!, not warn!: attacker-controlled, not operator-actionable.
- tracing::debug!(%error, "Failed to decode {api_name} request");
- respond_or_close(encode_err(api_version, ERROR_INVALID_REQUEST), api_name)
- }
- }
- } else {
- unsupported_version_response(api_key, api_version, |version| {
- encode_err(version, ERROR_UNSUPPORTED_VERSION)
- })
- }
-}
-
-/// Unsupported-version policy for APIs whose encoders only implement up to
-/// [`ApiVersionRange::max_version`].
-///
-/// - `api_version > max`: Close. `SUPPORTED_RANGES` is the governance boundary, not just an
-/// encoding-capability limit - `kafka_protocol` can often encode versions above our firewall
-/// max just fine, but responding there would silently widen what this gateway accepts.
-/// - `api_version < min`: Respond with an error shaped for that version when `kafka_protocol`
-/// can encode it, otherwise `encode` fails and [`respond_or_close`] closes instead. In
-/// practice every `SUPPORTED_RANGES` min was chosen at or above the oldest version
-/// `kafka_protocol` implements for that message, so this always closes today (e.g.
-/// `ListOffsets` v0's legacy `old_style_offsets` shape predates the crate's schema) - kept
-/// generic rather than hard-coded so a future `kafka_protocol` upgrade that widens a schema's
-/// floor is picked up automatically instead of silently staying on `Close`.
-fn unsupported_version_response(
- api_key: i16,
- api_version: i16,
- encode: impl FnOnce(i16) -> Result<Bytes>,
) -> HandleOutcome {
- let max_version = SUPPORTED_RANGES
- .iter()
- .find(|r| r.api_key == api_key)
- .map_or(0, |r| r.max_version);
- if api_version > max_version {
- tracing::warn!(
- api_key,
- api_version,
- max_version,
- "request version above encoder max; closing connection"
- );
- return HandleOutcome::Close;
- }
- respond_or_close(encode(api_version), "unsupported-version")
+ dispatch(state, api_key, api_version, body).await
}
#[must_use]
@@ -533,137 +249,3 @@ pub const fn advertised_min_version(api_key: i16, firewall_min: i16) -> i16 {
firewall_min
}
}
-
-fn encode_api_versions_response(api_version: i16, error_code: i16) -> Result<Bytes> {
- let api_keys = SUPPORTED_RANGES
- .iter()
- .map(|r| {
- ApiVersion::default()
- .with_api_key(r.api_key)
- .with_min_version(advertised_min_version(r.api_key, r.min_version))
- .with_max_version(r.max_version)
- })
- .collect();
- let resp = ApiVersionsResponse::default()
- .with_error_code(error_code)
- .with_api_keys(api_keys);
- encode_message(&resp, api_version, 128)
-}
-
-fn encode_metadata_response(
- response_version: i16,
- topics: &[StrBytes],
- broker: &BrokerAdvertise,
- topic_error_override: i16,
-) -> Result<Bytes> {
- // Stub has no topic catalog: echo requested names with UNKNOWN_TOPIC_OR_PARTITION,
- // or a forced override (unused today; kept for symmetry with other encoders).
- let topic_error = if topic_error_override == ERROR_NONE {
- ERROR_UNKNOWN_TOPIC_OR_PARTITION
- } else {
- topic_error_override
- };
-
- let response_topics = topics
- .iter()
- .map(|name| {
- MetadataResponseTopic::default()
- .with_error_code(topic_error)
- .with_name(Some(TopicName(name.clone())))
- })
- .collect();
-
- let broker_entry = MetadataResponseBroker::default()
- .with_node_id(BrokerId(1))
- .with_host(StrBytes::from_string(broker.host.clone()))
- .with_port(broker.port);
-
- let resp = MetadataResponse::default()
- .with_brokers(vec![broker_entry])
- .with_controller_id(BrokerId(1))
- .with_topics(response_topics);
-
- encode_message(&resp, response_version, 256)
-}
-
-/// Decodes a Metadata request body so the response can echo topic names.
-///
-/// A null topics array (`-1` legacy / `varint=0` compact) means "all topics" and decodes to an
-/// empty list for this stub. A null per-topic `name` (v10+ allows topic-id-only lookups) has no
-/// name to echo, so it errors rather than silently dropping the topic from the response.
-fn decode_metadata_topics(
- api_version: i16,
- body: Bytes,
- max_frame_size: usize,
-) -> Result<Vec<StrBytes>> {
- let req = decode_guarded::<MetadataRequest>(api_version, body, |v, b| {
- validate_metadata_shape(v, b, max_frame_size)
- })?;
- req.topics
- .unwrap_or_default()
- .into_iter()
- .map(|topic| {
- topic
- .name
- .map(|name| name.0)
- .ok_or(KafkaProtocolError::NullTopicName)
- })
- .collect()
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
-
- #[test]
- fn decode_metadata_topics_legacy_null_topic_name_fails() {
- let body = Bytes::from_static(&[
- 0x00, 0x00, 0x00, 0x01, // one topic
- 0xff, 0xff, // null topic name
- ]);
- let err = decode_metadata_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap_err();
- assert!(matches!(err, KafkaProtocolError::NullTopicName));
- }
-
- #[test]
- fn decode_metadata_topics_legacy_null_array_means_all_topics() {
- // -1 is the spec-defined "all topics" sentinel for the legacy i32 array count, not a
- // malformed request - must decode to an empty list.
- let body = Bytes::from_static(&[0xff, 0xff, 0xff, 0xff]); // -1
- let topics = decode_metadata_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap();
- assert!(topics.is_empty());
- }
-
- #[test]
- fn decode_metadata_topics_empty_body_is_malformed() {
- assert!(decode_metadata_topics(0, Bytes::new(), TEST_MAX_FRAME_SIZE).is_err());
- }
-
- #[test]
- fn decode_metadata_topics_flexible_truncated_after_topics_fails() {
- // topics = null (all topics) but missing allow_auto / auth flags / tagged fields.
- let body = Bytes::from_static(&[0x00]);
- assert!(decode_metadata_topics(9, body, TEST_MAX_FRAME_SIZE).is_err());
- }
-
- #[test]
- fn decode_api_versions_v3_requires_software_fields() {
- assert!(decode_exhaustive::<ApiVersionsRequest>(3, Bytes::new()).is_err());
- }
-
- #[test]
- fn decode_api_versions_v3_accepts_valid_body() {
- // Hand-encoded rather than round-tripped through `ApiVersionsRequest::encode`: encoding
- // is gated behind the crate's "client" feature, which this broker-only binary doesn't
- // enable.
- let body = Bytes::from_static(&[
- 0x0a, b'i', b'g', b'g', b'y', b'-', b't', b'e', b's',
- b't', // compact string (len 9)
- 0x06, b'0', b'.', b'1', b'.', b'0', // compact string (len 5)
- 0x00, // empty tagged fields
- ]);
- decode_exhaustive::<ApiVersionsRequest>(3, body).unwrap();
- }
-}
diff --git a/gateways/kafka/src/protocol/handlers/api_versions.rs b/gateways/kafka/src/protocol/handlers/api_versions.rs
new file mode 100644
index 000000000..f05ae4ce2
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/api_versions.rs
@@ -0,0 +1,108 @@
+// 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.
+
+//! `ApiVersions` (API key 18).
+
+use bytes::Bytes;
+use kafka_protocol::messages::api_versions_response::ApiVersion;
+use kafka_protocol::messages::{ApiVersionsRequest, ApiVersionsResponse};
+
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_API_VERSIONS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_NONE,
+ ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, advertised_min_version,
+ is_supported_version, supported_api_ranges,
+};
+use crate::protocol::bounds_guard::validate_api_versions_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_API_VERSIONS,
+ min_version: 0,
+ max_version: 3,
+};
+
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(_state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ if !is_supported_version(API_KEY_API_VERSIONS, api_version) {
+ // KIP-511: reply with v0 when the requested version is not understood.
+ return respond_or_close(encode_response(0, ERROR_UNSUPPORTED_VERSION), "ApiVersions");
+ }
+ match decode_guarded::<ApiVersionsRequest>(api_version, body, validate_api_versions_shape) {
+ Ok(_) => respond_or_close(encode_response(api_version, ERROR_NONE), "ApiVersions"),
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable (see the same
+ // note on the Produce decode-failure arm).
+ tracing::debug!(%error, "failed to decode ApiVersions request");
+ respond_or_close(
+ encode_response(api_version, ERROR_INVALID_REQUEST),
+ "ApiVersions",
+ )
+ }
+ }
+}
+
+/// Advertise exactly what the firewall allows, so a client never negotiates a version this
+/// gateway then refuses.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `api_version`.
+pub fn encode_response(api_version: i16, error_code: i16) -> Result<Bytes> {
+ let api_keys = supported_api_ranges()
+ .iter()
+ .map(|r| {
+ ApiVersion::default()
+ .with_api_key(r.api_key)
+ .with_min_version(advertised_min_version(r.api_key, r.min_version))
+ .with_max_version(r.max_version)
+ })
+ .collect();
+ let resp = ApiVersionsResponse::default()
+ .with_error_code(error_code)
+ .with_api_keys(api_keys);
+ encode_message(&resp, api_version, 128)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::protocol::handlers::decode_exhaustive;
+
+ #[test]
+ fn decode_v3_requires_software_fields() {
+ assert!(decode_exhaustive::<ApiVersionsRequest>(3, Bytes::new()).is_err());
+ }
+
+ #[test]
+ fn decode_v3_accepts_valid_body() {
+ // Hand-encoded rather than round-tripped through `ApiVersionsRequest::encode`: encoding
+ // is gated behind the crate's "client" feature, which this broker-only binary doesn't
+ // enable.
+ let body = Bytes::from_static(&[
+ 0x0a, b'i', b'g', b'g', b'y', b'-', b't', b'e', b's',
+ b't', // compact string (len 9)
+ 0x06, b'0', b'.', b'1', b'.', b'0', // compact string (len 5)
+ 0x00, // empty tagged fields
+ ]);
+ decode_exhaustive::<ApiVersionsRequest>(3, body).unwrap();
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/create_topics.rs b/gateways/kafka/src/protocol/handlers/create_topics.rs
new file mode 100644
index 000000000..b2532eec4
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/create_topics.rs
@@ -0,0 +1,131 @@
+// 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.
+
+//! `CreateTopics` (API key 19).
+
+use bytes::Bytes;
+use kafka_protocol::messages::create_topics_request::CreatableTopic;
+use kafka_protocol::messages::create_topics_response::CreatableTopicResult;
+use kafka_protocol::messages::{CreateTopicsRequest, CreateTopicsResponse};
+
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_PARTITIONS,
+ ERROR_INVALID_REPLICATION_FACTOR, ERROR_NONE, ERROR_NOT_CONTROLLER, GatewayState,
+ HandleOutcome,
+};
+use crate::protocol::bounds_guard::validate_create_topics_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, handle_versioned_request};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_CREATE_TOPICS,
+ min_version: 2,
+ max_version: 5,
+};
+
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ handle_versioned_request(
+ API_KEY_CREATE_TOPICS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
+ validate_create_topics_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "CreateTopics",
+ )
+}
+
+/// Well-formed `CreateTopics` response with a single placeholder topic.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ let topics = vec![
+ CreatableTopic::default()
+ .with_num_partitions(1)
+ .with_replication_factor(1),
+ ];
+ encode_inner(version, &topics, error_code)
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, req: &CreateTopicsRequest) -> Result<Bytes> {
+ encode_inner(version, &req.topics, ERROR_NONE)
+}
+
+/// Resolve per-topic `CreateTopics` error.
+///
+/// KIP-464: `num_partitions = -1` / `replication_factor = -1` mean broker default when either
+/// (a) the version is v4+, or (b) the topic carries a manual partition assignment (valid on
+/// v2/v3 as well). Otherwise non-positive values are [`ERROR_INVALID_PARTITIONS`] /
+/// [`ERROR_INVALID_REPLICATION_FACTOR`]. When validation passes, the stub returns
+/// [`ERROR_NOT_CONTROLLER`] so clients do not believe the topic was created.
+const fn topic_error(version: i16, topic: &CreatableTopic, forced_error: i16) -> i16 {
+ if forced_error != ERROR_NONE {
+ return forced_error;
+ }
+
+ let broker_default_ok = version >= 4 || !topic.assignments.is_empty();
+
+ let partitions_ok = if broker_default_ok {
+ topic.num_partitions == -1 || topic.num_partitions > 0
+ } else {
+ topic.num_partitions > 0
+ };
+ if !partitions_ok {
+ return ERROR_INVALID_PARTITIONS;
+ }
+
+ let replication_ok = if broker_default_ok {
+ topic.replication_factor == -1 || topic.replication_factor > 0
+ } else {
+ topic.replication_factor > 0
+ };
+ if !replication_ok {
+ return ERROR_INVALID_REPLICATION_FACTOR;
+ }
+
+ ERROR_NOT_CONTROLLER
+}
+
+fn encode_inner(version: i16, topics: &[CreatableTopic], forced_error: i16) -> Result<Bytes> {
+ let results = topics
+ .iter()
+ .map(|topic| {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(topic_error(version, topic, forced_error))
+ .with_error_message(None)
+ .with_num_partitions(topic.num_partitions)
+ .with_replication_factor(topic.replication_factor)
+ })
+ .collect();
+ let resp = CreateTopicsResponse::default().with_topics(results);
+ encode_message(&resp, version, 256)
+}
diff --git a/gateways/kafka/src/protocol/handlers/fetch.rs b/gateways/kafka/src/protocol/handlers/fetch.rs
new file mode 100644
index 000000000..6d86603d8
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/fetch.rs
@@ -0,0 +1,119 @@
+// 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.
+
+//! Fetch (API key 1).
+
+use bytes::Bytes;
+use kafka_protocol::messages::fetch_response::{FetchableTopicResponse, PartitionData};
+use kafka_protocol::messages::{FetchRequest, FetchResponse};
+
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_FETCH, ApiVersionRange, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, GatewayState,
+ HandleOutcome,
+};
+use crate::protocol::bounds_guard::validate_fetch_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, handle_versioned_request};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_FETCH,
+ min_version: 4,
+ max_version: 12,
+};
+
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ handle_versioned_request(
+ API_KEY_FETCH,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<FetchRequest>(v, b, |v, b| {
+ validate_fetch_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "Fetch",
+ )
+}
+
+/// Well-formed Fetch response. Uses top-level `error_code` at v7+, or a single
+/// placeholder topic/partition with per-partition `error_code` below v7.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ if version >= 7 {
+ return encode_inner(version, Vec::new(), error_code);
+ }
+ // No top-level error field below v7; the error surfaces on the placeholder partition instead.
+ let topics = vec![
+ FetchableTopicResponse::default().with_partitions(vec![partition_response(0, error_code)]),
+ ];
+ encode_inner(version, topics, ERROR_NONE)
+}
+
+/// Stub: discard the payload and return a retriable error so clients don't mistake "no real
+/// data yet" for a genuinely empty partition (same philosophy as Produce).
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, req: &FetchRequest) -> Result<Bytes> {
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| {
+ FetchableTopicResponse::default()
+ .with_topic(topic.topic.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|p| partition_response(p.partition, ERROR_NOT_LEADER_OR_FOLLOWER))
+ .collect(),
+ )
+ })
+ .collect();
+ encode_inner(version, topics, ERROR_NONE)
+}
+
+fn encode_inner(
+ version: i16,
+ topics: Vec<FetchableTopicResponse>,
+ top_level_error: i16,
+) -> Result<Bytes> {
+ let resp = FetchResponse::default()
+ .with_error_code(top_level_error)
+ .with_responses(topics);
+ encode_message(&resp, version, 512)
+}
+
+fn partition_response(partition: i32, error_code: i16) -> PartitionData {
+ PartitionData::default()
+ .with_partition_index(partition)
+ .with_error_code(error_code)
+ .with_last_stable_offset(0)
+ .with_log_start_offset(0)
+ .with_records(None)
+}
diff --git a/gateways/kafka/src/protocol/handlers/list_offsets.rs b/gateways/kafka/src/protocol/handlers/list_offsets.rs
new file mode 100644
index 000000000..f31397347
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/list_offsets.rs
@@ -0,0 +1,116 @@
+// 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.
+
+//! `ListOffsets` (API key 2).
+
+use bytes::Bytes;
+use kafka_protocol::messages::list_offsets_response::{
+ ListOffsetsPartitionResponse, ListOffsetsTopicResponse,
+};
+use kafka_protocol::messages::{ListOffsetsRequest, ListOffsetsResponse};
+
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, GatewayState,
+ HandleOutcome,
+};
+use crate::protocol::bounds_guard::validate_list_offsets_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, handle_versioned_request};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_LIST_OFFSETS,
+ min_version: 1,
+ max_version: 6,
+};
+
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ handle_versioned_request(
+ API_KEY_LIST_OFFSETS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<ListOffsetsRequest>(v, b, |v, b| {
+ validate_list_offsets_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "ListOffsets",
+ )
+}
+
+/// Well-formed `ListOffsets` response with a single placeholder topic/partition.
+///
+/// `kafka_protocol` has no encodable representation for `ListOffsets` v0 (the legacy
+/// `old_style_offsets` shape predates the schema this crate generates from); a v0 request now
+/// falls through `super::unsupported_version_response`'s encode-failure path to `Close`
+/// instead of the pre-migration downgraded response.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version` (always the
+/// case for `version == 0`).
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ let topics = vec![
+ ListOffsetsTopicResponse::default()
+ .with_partitions(vec![partition_response(0, error_code)]),
+ ];
+ encode_inner(version, topics)
+}
+
+/// Stub: discard the payload and return a retriable error, matching Produce/Fetch - a genuine
+/// offset lookup requires the same partition-leadership the stub doesn't have yet.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, req: &ListOffsetsRequest) -> Result<Bytes> {
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| {
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|p| {
+ partition_response(p.partition_index, ERROR_NOT_LEADER_OR_FOLLOWER)
+ })
+ .collect(),
+ )
+ })
+ .collect();
+ encode_inner(version, topics)
+}
+
+fn encode_inner(version: i16, topics: Vec<ListOffsetsTopicResponse>) -> Result<Bytes> {
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ encode_message(&resp, version, 256)
+}
+
+fn partition_response(partition: i32, error_code: i16) -> ListOffsetsPartitionResponse {
+ ListOffsetsPartitionResponse::default()
+ .with_partition_index(partition)
+ .with_error_code(error_code)
+}
diff --git a/gateways/kafka/src/protocol/handlers/metadata.rs b/gateways/kafka/src/protocol/handlers/metadata.rs
new file mode 100644
index 000000000..b92c35e30
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/metadata.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.
+
+//! Metadata (API key 3).
+
+use bytes::Bytes;
+use kafka_protocol::messages::metadata_response::{MetadataResponseBroker, MetadataResponseTopic};
+use kafka_protocol::messages::{BrokerId, MetadataRequest, MetadataResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::error::{KafkaProtocolError, Result};
+use crate::protocol::api::{
+ API_KEY_METADATA, ApiVersionRange, BrokerAdvertise, ERROR_NONE,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, is_supported_version,
+ supported_max_version,
+};
+use crate::protocol::bounds_guard::validate_metadata_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_METADATA,
+ min_version: 0,
+ max_version: 9,
+};
+
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ if !is_supported_version(API_KEY_METADATA, api_version) {
+ // Clamping the response to the supported max 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 = supported_max_version(API_KEY_METADATA),
+ "Metadata version unsupported; closing connection"
+ );
+ return HandleOutcome::Close;
+ }
+ match decode_topics(api_version, body, state.max_frame_size) {
+ Ok(topics) => respond_or_close(
+ encode_response(api_version, &topics, &state.broker, ERROR_NONE),
+ "Metadata",
+ ),
+ Err(error) => {
+ // Metadata has no top-level error field; a malformed body cannot carry
+ // INVALID_REQUEST in a version-correct way for every client. Close.
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(
+ %error,
+ api_version,
+ "Failed to decode Metadata request; closing connection"
+ );
+ HandleOutcome::Close
+ }
+ }
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `response_version`.
+pub fn encode_response(
+ response_version: i16,
+ topics: &[StrBytes],
+ broker: &BrokerAdvertise,
+ topic_error_override: i16,
+) -> Result<Bytes> {
+ // Stub has no topic catalog: echo requested names with UNKNOWN_TOPIC_OR_PARTITION,
+ // or a forced override (unused today; kept for symmetry with other encoders).
+ let topic_error = if topic_error_override == ERROR_NONE {
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ } else {
+ topic_error_override
+ };
+
+ let response_topics = topics
+ .iter()
+ .map(|name| {
+ MetadataResponseTopic::default()
+ .with_error_code(topic_error)
+ .with_name(Some(TopicName(name.clone())))
+ })
+ .collect();
+
+ let broker_entry = MetadataResponseBroker::default()
+ .with_node_id(BrokerId(1))
+ .with_host(StrBytes::from_string(broker.host.clone()))
+ .with_port(broker.port);
+
+ let resp = MetadataResponse::default()
+ .with_brokers(vec![broker_entry])
+ .with_controller_id(BrokerId(1))
+ .with_topics(response_topics);
+
+ encode_message(&resp, response_version, 256)
+}
+
+/// Decodes a Metadata request body so the response can echo topic names.
+///
+/// A null topics array (`-1` legacy / `varint=0` compact) means "all topics" and decodes to an
+/// empty list for this stub. A null per-topic `name` (v10+ allows topic-id-only lookups) has no
+/// name to echo, so it errors rather than silently dropping the topic from the response.
+fn decode_topics(api_version: i16, body: Bytes, max_frame_size: usize) -> Result<Vec<StrBytes>> {
+ let req = decode_guarded::<MetadataRequest>(api_version, body, |v, b| {
+ validate_metadata_shape(v, b, max_frame_size)
+ })?;
+ req.topics
+ .unwrap_or_default()
+ .into_iter()
+ .map(|topic| {
+ topic
+ .name
+ .map(|name| name.0)
+ .ok_or(KafkaProtocolError::NullTopicName)
+ })
+ .collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+
+ #[test]
+ fn decode_topics_legacy_null_topic_name_fails() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // one topic
+ 0xff, 0xff, // null topic name
+ ]);
+ let err = decode_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap_err();
+ assert!(matches!(err, KafkaProtocolError::NullTopicName));
+ }
+
+ #[test]
+ fn decode_topics_legacy_null_array_means_all_topics() {
+ // -1 is the spec-defined "all topics" sentinel for the legacy i32 array count, not a
+ // malformed request - must decode to an empty list.
+ let body = Bytes::from_static(&[0xff, 0xff, 0xff, 0xff]); // -1
+ let topics = decode_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap();
+ assert!(topics.is_empty());
+ }
+
+ #[test]
+ fn decode_topics_empty_body_is_malformed() {
+ assert!(decode_topics(0, Bytes::new(), TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn decode_topics_flexible_truncated_after_topics_fails() {
+ // topics = null (all topics) but missing allow_auto / auth flags / tagged fields.
+ let body = Bytes::from_static(&[0x00]);
+ assert!(decode_topics(9, body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
new file mode 100644
index 000000000..01ee09d83
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -0,0 +1,181 @@
+// 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.
+
+//! One module per Kafka API key.
+//!
+//! Each module owns its whole vertical: the firewall range it occupies, the response encoders
+//! for it, and the handler [`dispatch`] routes to. Adding an API key means adding a module, a
+//! line in [`dispatch`], and a line in `api::SUPPORTED_RANGES`, instead of editing a match, an
+//! encoder file and a bounds table that every other API also edits.
+//!
+//! Wire encoding (field order, version gating, compact versus legacy shapes) stays
+//! `kafka_protocol`'s responsibility. What lives here is policy: which placeholder values and
+//! error codes a request gets back, and when the connection closes instead.
+
+pub mod api_versions;
+pub mod create_topics;
+pub mod fetch;
+pub mod list_offsets;
+pub mod metadata;
+pub mod produce;
+
+use bytes::{Buf, Bytes, BytesMut};
+use kafka_protocol::protocol::{Decodable, Encodable};
+
+use crate::error::{KafkaProtocolError, Result};
+use crate::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_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version, supported_max_version,
+};
+
+/// Routes one decoded request body to the module that owns its API key.
+///
+/// An unknown key closes the connection: no api-specific response schema exists for it, so any
+/// body sent back is misparsed by the client against the schema it expected.
+pub async fn dispatch(
+ state: &GatewayState,
+ api_key: i16,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ match api_key {
+ API_KEY_PRODUCE => produce::handle(state, api_version, body).await,
+ API_KEY_FETCH => fetch::handle(state, api_version, body).await,
+ API_KEY_LIST_OFFSETS => list_offsets::handle(state, api_version, body).await,
+ API_KEY_METADATA => metadata::handle(state, api_version, body).await,
+ API_KEY_API_VERSIONS => api_versions::handle(state, api_version, body).await,
+ API_KEY_CREATE_TOPICS => create_topics::handle(state, api_version, body).await,
+ _ => HandleOutcome::Close,
+ }
+}
+
+/// Encode a `kafka_protocol` message, mapping its `anyhow::Error` (the crate has no stable
+/// decode/encode error taxonomy) to a variant callers can log or fold into
+/// [`HandleOutcome::Close`].
+pub(crate) fn encode_message<T: Encodable>(
+ msg: &T,
+ version: i16,
+ capacity: usize,
+) -> Result<Bytes> {
+ let mut buf = BytesMut::with_capacity(capacity);
+ msg.encode(&mut buf, version)
+ .map_err(|e| KafkaProtocolError::Malformed(e.to_string()))?;
+ Ok(buf.freeze())
+}
+
+/// 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.
+pub(crate) 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)
+}
+
+/// Runs `validate` (see [`crate::protocol::bounds_guard`]) before [`decode_exhaustive`].
+///
+/// `kafka_protocol` validates wire-declared array/string lengths as non-negative but never
+/// against the bytes actually remaining in the frame, so a tiny frame declaring a huge count
+/// drives an allocation attempt that aborts the whole process (`handle_alloc_error`, not a
+/// panic - uncatchable, and it kills every connection, not just the offending one). `validate`
+/// rejects that class of frame first, on a walk that never allocates a collection.
+pub(crate) fn decode_guarded<T: Decodable>(
+ version: i16,
+ body: Bytes,
+ validate: impl FnOnce(i16, &Bytes) -> Result<()>,
+) -> Result<T> {
+ validate(version, &body)?;
+ decode_exhaustive(version, body)
+}
+
+/// 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.
+pub(crate) 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
+ }
+ }
+}
+
+/// The shared shape of every API whose unsupported-version answer is an error response rather
+/// than a close, and whose decode failure is reportable.
+pub(crate) fn handle_versioned_request<T>(
+ api_key: i16,
+ api_version: i16,
+ body: Bytes,
+ decode: impl FnOnce(i16, Bytes) -> Result<T>,
+ encode_ok: impl FnOnce(i16, &T) -> Result<Bytes>,
+ encode_err: impl Fn(i16, i16) -> Result<Bytes>,
+ api_name: &str,
+) -> HandleOutcome {
+ if is_supported_version(api_key, api_version) {
+ match decode(api_version, body) {
+ Ok(req) => respond_or_close(encode_ok(api_version, &req), api_name),
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode {api_name} request");
+ respond_or_close(encode_err(api_version, ERROR_INVALID_REQUEST), api_name)
+ }
+ }
+ } else {
+ unsupported_version_response(api_key, api_version, |version| {
+ encode_err(version, ERROR_UNSUPPORTED_VERSION)
+ })
+ }
+}
+
+/// Unsupported-version policy for APIs whose encoders only implement up to
+/// `ApiVersionRange::max_version`.
+///
+/// - `api_version > max`: Close. `SUPPORTED_RANGES` is the governance boundary, not just an
+/// encoding-capability limit - `kafka_protocol` can often encode versions above our firewall
+/// max just fine, but responding there would silently widen what this gateway accepts.
+/// - `api_version < min`: Respond with an error shaped for that version when `kafka_protocol`
+/// can encode it, otherwise `encode` fails and [`respond_or_close`] closes instead. In
+/// practice every `SUPPORTED_RANGES` min was chosen at or above the oldest version
+/// `kafka_protocol` implements for that message, so this always closes today (e.g.
+/// `ListOffsets` v0's legacy `old_style_offsets` shape predates the crate's schema) - kept
+/// generic rather than hard-coded so a future `kafka_protocol` upgrade that widens a schema's
+/// floor is picked up automatically instead of silently staying on `Close`.
+pub(crate) fn unsupported_version_response(
+ api_key: i16,
+ api_version: i16,
+ encode: impl FnOnce(i16) -> Result<Bytes>,
+) -> HandleOutcome {
+ let max_version = supported_max_version(api_key).unwrap_or(0);
+ if api_version > max_version {
+ tracing::warn!(
+ api_key,
+ api_version,
+ max_version,
+ "request version above encoder max; closing connection"
+ );
+ return HandleOutcome::Close;
+ }
+ respond_or_close(encode(api_version), "unsupported-version")
+}
diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs
new file mode 100644
index 000000000..432345053
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/produce.rs
@@ -0,0 +1,151 @@
+// 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.
+
+//! Produce (API key 0).
+
+use bytes::Bytes;
+use kafka_protocol::messages::produce_response::{PartitionProduceResponse, TopicProduceResponse};
+use kafka_protocol::messages::{ProduceRequest, ProduceResponse};
+
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_PRODUCE, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, supported_max_version,
+};
+use crate::protocol::bounds_guard::validate_produce_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_PRODUCE,
+ min_version: 3,
+ max_version: 9,
+};
+
+/// 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 `api::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.
+#[expect(
+ clippy::unused_async,
+ reason = "the handler signature every API key shares; the first one to await the bridge \
+ retires its own attribute"
+)]
+pub async fn handle(state: &GatewayState, 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). Fail-closed on a missing row
+ // (i16::MIN, not i16::MAX): dispatch routes Produce on its api key, not from this table, so
+ // if a future edit ever drops the Produce row to disable the API, a fail-open default here
+ // would leave Produce v0-2 acks=0 silently accepted on an API the operator believes is off.
+ if api_version > supported_max_version(API_KEY_PRODUCE).unwrap_or(i16::MIN) {
+ 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_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ }),
+ };
+ }
+ match decode_guarded::<ProduceRequest>(api_version, body, |v, b| {
+ validate_produce_shape(v, b, state.max_frame_size)
+ }) {
+ // 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,
+ // `api_version` is always in `[3, supported_max]` here: the `< 3` case returned above,
+ // and the `> max` case returned at the top of this function - so it is always within
+ // `SUPPORTED_RANGES`' Produce row and an `is_supported_version` re-check can never fail.
+ Ok(req) => respond_or_close(encode_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.
+ // debug!, not warn!: the body is attacker-controlled, not operator-actionable, and
+ // a client looping malformed bodies on one connection (never disconnected - this
+ // arm returns NoResponse) has no rate limit here.
+ tracing::debug!(%error, "failed to decode Produce request (no response)");
+ HandleOutcome::NoResponse
+ }
+ }
+}
+
+/// Well-formed Produce response with a single placeholder topic/partition.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ let resp = ProduceResponse::default().with_responses(vec![
+ TopicProduceResponse::default()
+ .with_partition_responses(vec![partition_response(0, error_code)]),
+ ]);
+ encode_message(&resp, version, 512)
+}
+
+/// Stub: discard the payload and return a retriable error so clients keep data locally until
+/// the Iggy bridge lands (do not advertise silent success).
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, req: &ProduceRequest) -> Result<Bytes> {
+ let responses = req
+ .topic_data
+ .iter()
+ .map(|topic| {
+ TopicProduceResponse::default()
+ .with_name(topic.name.clone())
+ .with_partition_responses(
+ topic
+ .partition_data
+ .iter()
+ .map(|p| partition_response(p.index, ERROR_NOT_LEADER_OR_FOLLOWER))
+ .collect(),
+ )
+ })
+ .collect();
+ let resp = ProduceResponse::default().with_responses(responses);
+ encode_message(&resp, version, 512)
+}
+
+fn partition_response(index: i32, error_code: i16) -> PartitionProduceResponse {
+ PartitionProduceResponse::default()
+ .with_index(index)
+ .with_error_code(error_code)
+ .with_log_start_offset(0)
+}
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/protocol/mod.rs
index d7fd8a214..f07afdb1f 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/protocol/mod.rs
@@ -17,5 +17,5 @@
pub mod api;
pub mod bounds_guard;
+pub mod handlers;
pub mod header;
-pub mod responses;
diff --git a/gateways/kafka/src/protocol/responses.rs b/gateways/kafka/src/protocol/responses.rs
deleted file mode 100644
index b929b5696..000000000
--- a/gateways/kafka/src/protocol/responses.rs
+++ /dev/null
@@ -1,318 +0,0 @@
-// 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 response encoders (stub implementations).
-//!
-//! Wire encoding (field order, version gating, compact vs. legacy shapes) is
-//! `kafka_protocol`'s responsibility; everything here is stub *policy* - which placeholder
-//! values and error codes a request gets back before the Iggy bridge lands.
-
-use bytes::{Bytes, BytesMut};
-use kafka_protocol::messages::create_topics_request::CreatableTopic;
-use kafka_protocol::messages::create_topics_response::CreatableTopicResult;
-use kafka_protocol::messages::fetch_response::{FetchableTopicResponse, PartitionData};
-use kafka_protocol::messages::list_offsets_response::{
- ListOffsetsPartitionResponse, ListOffsetsTopicResponse,
-};
-use kafka_protocol::messages::produce_response::{PartitionProduceResponse, TopicProduceResponse};
-use kafka_protocol::messages::{
- CreateTopicsRequest, CreateTopicsResponse, FetchRequest, FetchResponse, ListOffsetsRequest,
- ListOffsetsResponse, ProduceRequest, ProduceResponse,
-};
-use kafka_protocol::protocol::Encodable;
-
-use crate::error::{KafkaProtocolError, Result};
-use crate::protocol::api::{
- ERROR_INVALID_PARTITIONS, ERROR_INVALID_REPLICATION_FACTOR, ERROR_NONE, ERROR_NOT_CONTROLLER,
- ERROR_NOT_LEADER_OR_FOLLOWER,
-};
-
-/// Encode a `kafka_protocol` message, mapping its `anyhow::Error` (the crate has no stable
-/// decode/encode error taxonomy) to a variant callers can log or fold into [`HandleOutcome::Close`].
-///
-/// [`HandleOutcome::Close`]: crate::protocol::api::HandleOutcome::Close
-pub(crate) fn encode_message<T: Encodable>(
- msg: &T,
- version: i16,
- capacity: usize,
-) -> Result<Bytes> {
- let mut buf = BytesMut::with_capacity(capacity);
- msg.encode(&mut buf, version)
- .map_err(|e| KafkaProtocolError::Malformed(e.to_string()))?;
- Ok(buf.freeze())
-}
-
-// ── Produce ──────────────────────────────────────────────────────────────────
-
-/// Well-formed Produce response with a single placeholder topic/partition.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_produce_error_response(version: i16, error_code: i16) -> Result<Bytes> {
- let resp = ProduceResponse::default().with_responses(vec![
- TopicProduceResponse::default()
- .with_partition_responses(vec![produce_partition_response(0, error_code)]),
- ]);
- encode_message(&resp, version, 512)
-}
-
-/// Stub: discard the payload and return a retriable error so clients keep data locally until
-/// the Iggy bridge lands (do not advertise silent success).
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_produce_response(version: i16, req: &ProduceRequest) -> Result<Bytes> {
- let responses = req
- .topic_data
- .iter()
- .map(|topic| {
- TopicProduceResponse::default()
- .with_name(topic.name.clone())
- .with_partition_responses(
- topic
- .partition_data
- .iter()
- .map(|p| produce_partition_response(p.index, ERROR_NOT_LEADER_OR_FOLLOWER))
- .collect(),
- )
- })
- .collect();
- let resp = ProduceResponse::default().with_responses(responses);
- encode_message(&resp, version, 512)
-}
-
-fn produce_partition_response(index: i32, error_code: i16) -> PartitionProduceResponse {
- PartitionProduceResponse::default()
- .with_index(index)
- .with_error_code(error_code)
- .with_log_start_offset(0)
-}
-
-// ── Fetch ────────────────────────────────────────────────────────────────────
-
-/// Well-formed Fetch response. Uses top-level `error_code` at v7+, or a single
-/// placeholder topic/partition with per-partition `error_code` below v7.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_fetch_error_response(version: i16, error_code: i16) -> Result<Bytes> {
- if version >= 7 {
- return encode_fetch_response_inner(version, Vec::new(), error_code);
- }
- // No top-level error field below v7; the error surfaces on the placeholder partition instead.
- let topics = vec![
- FetchableTopicResponse::default()
- .with_partitions(vec![fetch_partition_response(0, error_code)]),
- ];
- encode_fetch_response_inner(version, topics, ERROR_NONE)
-}
-
-/// Stub: discard the payload and return a retriable error so clients don't mistake "no real
-/// data yet" for a genuinely empty partition (same philosophy as Produce).
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_fetch_response(version: i16, req: &FetchRequest) -> Result<Bytes> {
- let topics = req
- .topics
- .iter()
- .map(|topic| {
- FetchableTopicResponse::default()
- .with_topic(topic.topic.clone())
- .with_partitions(
- topic
- .partitions
- .iter()
- .map(|p| {
- fetch_partition_response(p.partition, ERROR_NOT_LEADER_OR_FOLLOWER)
- })
- .collect(),
- )
- })
- .collect();
- encode_fetch_response_inner(version, topics, ERROR_NONE)
-}
-
-fn encode_fetch_response_inner(
- version: i16,
- topics: Vec<FetchableTopicResponse>,
- top_level_error: i16,
-) -> Result<Bytes> {
- let resp = FetchResponse::default()
- .with_error_code(top_level_error)
- .with_responses(topics);
- encode_message(&resp, version, 512)
-}
-
-fn fetch_partition_response(partition: i32, error_code: i16) -> PartitionData {
- PartitionData::default()
- .with_partition_index(partition)
- .with_error_code(error_code)
- .with_last_stable_offset(0)
- .with_log_start_offset(0)
- .with_records(None)
-}
-
-// ── ListOffsets ──────────────────────────────────────────────────────────────
-
-/// Well-formed `ListOffsets` response with a single placeholder topic/partition.
-///
-/// `kafka_protocol` has no encodable representation for `ListOffsets` v0 (the legacy
-/// `old_style_offsets` shape predates the schema this crate generates from); a v0 request now
-/// falls through `super::api::unsupported_version_response`'s encode-failure path to `Close`
-/// instead of the pre-migration downgraded response.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version` (always the
-/// case for `version == 0`).
-pub fn encode_list_offsets_error_response(version: i16, error_code: i16) -> Result<Bytes> {
- let topics = vec![
- ListOffsetsTopicResponse::default()
- .with_partitions(vec![list_offsets_partition_response(0, error_code)]),
- ];
- encode_list_offsets_response_inner(version, topics)
-}
-
-/// Stub: discard the payload and return a retriable error, matching Produce/Fetch - a genuine
-/// offset lookup requires the same partition-leadership the stub doesn't have yet.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_list_offsets_response(version: i16, req: &ListOffsetsRequest) -> Result<Bytes> {
- let topics = req
- .topics
- .iter()
- .map(|topic| {
- ListOffsetsTopicResponse::default()
- .with_name(topic.name.clone())
- .with_partitions(
- topic
- .partitions
- .iter()
- .map(|p| {
- list_offsets_partition_response(
- p.partition_index,
- ERROR_NOT_LEADER_OR_FOLLOWER,
- )
- })
- .collect(),
- )
- })
- .collect();
- encode_list_offsets_response_inner(version, topics)
-}
-
-fn encode_list_offsets_response_inner(
- version: i16,
- topics: Vec<ListOffsetsTopicResponse>,
-) -> Result<Bytes> {
- let resp = ListOffsetsResponse::default().with_topics(topics);
- encode_message(&resp, version, 256)
-}
-
-fn list_offsets_partition_response(
- partition: i32,
- error_code: i16,
-) -> ListOffsetsPartitionResponse {
- ListOffsetsPartitionResponse::default()
- .with_partition_index(partition)
- .with_error_code(error_code)
-}
-
-// ── CreateTopics ─────────────────────────────────────────────────────────────
-
-/// Well-formed `CreateTopics` response with a single placeholder topic.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_create_topics_error_response(version: i16, error_code: i16) -> Result<Bytes> {
- let topics = vec![
- CreatableTopic::default()
- .with_num_partitions(1)
- .with_replication_factor(1),
- ];
- encode_create_topics_response_inner(version, &topics, error_code)
-}
-
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_create_topics_response(version: i16, req: &CreateTopicsRequest) -> Result<Bytes> {
- encode_create_topics_response_inner(version, &req.topics, ERROR_NONE)
-}
-
-/// Resolve per-topic `CreateTopics` error.
-///
-/// KIP-464: `num_partitions = -1` / `replication_factor = -1` mean broker default when either
-/// (a) the version is v4+, or (b) the topic carries a manual partition assignment (valid on
-/// v2/v3 as well). Otherwise non-positive values are [`ERROR_INVALID_PARTITIONS`] /
-/// [`ERROR_INVALID_REPLICATION_FACTOR`]. When validation passes, the stub returns
-/// [`ERROR_NOT_CONTROLLER`] so clients do not believe the topic was created.
-const fn create_topics_topic_error(version: i16, topic: &CreatableTopic, forced_error: i16) -> i16 {
- if forced_error != ERROR_NONE {
- return forced_error;
- }
-
- let broker_default_ok = version >= 4 || !topic.assignments.is_empty();
-
- let partitions_ok = if broker_default_ok {
- topic.num_partitions == -1 || topic.num_partitions > 0
- } else {
- topic.num_partitions > 0
- };
- if !partitions_ok {
- return ERROR_INVALID_PARTITIONS;
- }
-
- let replication_ok = if broker_default_ok {
- topic.replication_factor == -1 || topic.replication_factor > 0
- } else {
- topic.replication_factor > 0
- };
- if !replication_ok {
- return ERROR_INVALID_REPLICATION_FACTOR;
- }
-
- ERROR_NOT_CONTROLLER
-}
-
-fn encode_create_topics_response_inner(
- version: i16,
- topics: &[CreatableTopic],
- topic_error: i16,
-) -> Result<Bytes> {
- let results = topics
- .iter()
- .map(|topic| {
- CreatableTopicResult::default()
- .with_name(topic.name.clone())
- .with_error_code(create_topics_topic_error(version, topic, topic_error))
- .with_error_message(None)
- .with_num_partitions(topic.num_partitions)
- .with_replication_factor(topic.replication_factor)
- })
- .collect();
- let resp = CreateTopicsResponse::default().with_topics(results);
- encode_message(&resp, version, 256)
-}
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index b21be3ae1..58b9eaed7 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -31,9 +31,10 @@ use tokio_util::task::TaskTracker;
use tracing::{debug, error, info, warn};
use tracing_appender::non_blocking::WorkerGuard;
+use crate::bridge::IggyBridge;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- BrokerAdvertise, DEFAULT_KAFKA_PORT, HandleOutcome, handle_request_bounded,
+ BrokerAdvertise, DEFAULT_KAFKA_PORT, GatewayState, HandleOutcome, handle_request_bounded,
};
use crate::protocol::header::{request_header_version, response_header_version};
use std::io;
@@ -162,6 +163,7 @@ pub fn bind_listener(addr: &str) -> Result<TcpListener> {
pub struct KafkaGateway {
config: Arc<GatewayConfig>,
+ bridge: Option<Arc<IggyBridge>>,
}
impl KafkaGateway {
@@ -169,9 +171,20 @@ impl KafkaGateway {
pub fn new(config: GatewayConfig) -> Self {
Self {
config: Arc::new(config),
+ bridge: None,
}
}
+ /// Serve requests against `bridge` instead of the stub answers.
+ ///
+ /// `None` is the default and keeps every handler on its stub, which is what makes the
+ /// bridge safe to land one API at a time.
+ #[must_use]
+ pub fn with_bridge(mut self, bridge: Option<Arc<IggyBridge>>) -> Self {
+ self.bridge = bridge;
+ self
+ }
+
/// 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
@@ -186,14 +199,16 @@ impl KafkaGateway {
mut shutdown: broadcast::Receiver<()>,
) -> Result<()> {
let local_addr = listener.local_addr()?;
- let broker = Arc::new(BrokerAdvertise::from_server_config(
- &self.config,
- local_addr,
- )?);
+ let broker = BrokerAdvertise::from_server_config(&self.config, local_addr)?;
info!(
"kafka listener bound on {} (advertised as {}:{})",
local_addr, broker.host, broker.port
);
+ let state = Arc::new(GatewayState::new(
+ broker,
+ self.bridge.clone(),
+ self.config.max_frame_size,
+ ));
let tracker = TaskTracker::new();
let conn_limiter = Arc::new(Semaphore::new(self.config.max_connections));
@@ -238,12 +253,12 @@ impl KafkaGateway {
warn!(%peer, "TCP_KEEPALIVE failed: {e}");
}
let cfg = Arc::clone(&self.config);
- let broker = Arc::clone(&broker);
+ let state = Arc::clone(&state);
let conn_cancel = cancel.child_token();
tracker.spawn(async move {
let _permit = permit;
if let Err(err) =
- handle_connection(stream, cfg, peer, broker, conn_cancel).await
+ handle_connection(stream, cfg, peer, state, conn_cancel).await
{
// debug!, not warn!: every `KafkaProtocolError` that can
// reach here is either a malformed/oversized frame from the
@@ -311,7 +326,7 @@ async fn handle_connection(
mut stream: TcpStream,
config: Arc<GatewayConfig>,
peer: SocketAddr,
- broker: Arc<BrokerAdvertise>,
+ state: Arc<GatewayState>,
cancel: CancellationToken,
) -> Result<()> {
debug!(%peer, "connection accepted");
@@ -350,13 +365,9 @@ async fn handle_connection(
// `RequestHeader::decode` advances `body` past the header fields it consumed via
// `Buf::advance`, so `body` is already exactly the request payload.
- let outcome = handle_request_bounded(
- req.request_api_key,
- req.request_api_version,
- body,
- &broker,
- config.max_frame_size,
- );
+ let outcome =
+ handle_request_bounded(&state, req.request_api_key, req.request_api_version, body)
+ .await;
if dispatch_outcome(&mut stream, &peer, &config, &req, resp_hdr_ver, outcome).await? {
return Ok(());
}
diff --git a/gateways/kafka/tests/api_handler_tests.rs b/gateways/kafka/tests/api_handler_tests.rs
index eeffc5c15..580cb857e 100644
--- a/gateways/kafka/tests/api_handler_tests.rs
+++ b/gateways/kafka/tests/api_handler_tests.rs
@@ -47,9 +47,10 @@ use wire::{
// ── ApiVersions ─────────────────────────────────────────────────────────────
-#[test]
-fn api_versions_v1_response_non_flexible_format() {
+#[tokio::test]
+async fn api_versions_v1_response_non_flexible_format() {
let body = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
@@ -72,14 +73,15 @@ fn api_versions_v1_response_non_flexible_format() {
}
}
-#[test]
-fn api_versions_v3_response_flexible_format() {
+#[tokio::test]
+async fn api_versions_v3_response_flexible_format() {
let body = handle_request(
API_KEY_API_VERSIONS,
3,
build_api_versions_flexible_request("apache-iggy", "0.1.0"),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
@@ -108,14 +110,15 @@ fn api_versions_v3_response_flexible_format() {
// ── Metadata ─────────────────────────────────────────────────────────────────
-#[test]
-fn metadata_response_has_broker_array_and_topic_array() {
+#[tokio::test]
+async fn metadata_response_has_broker_array_and_topic_array() {
let body = handle_request(
API_KEY_METADATA,
0,
build_metadata_all_topics_legacy(0),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
@@ -132,24 +135,27 @@ fn metadata_response_has_broker_array_and_topic_array() {
assert_eq!(topic_count, 0);
}
-#[test]
-fn metadata_empty_body_closes_connection() {
+#[tokio::test]
+async fn metadata_empty_body_closes_connection() {
assert!(
- handle_request(API_KEY_METADATA, 0, Bytes::new(), &default_broker()).is_close(),
+ handle_request(API_KEY_METADATA, 0, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"empty Metadata body is malformed, not all-topics"
);
}
-#[test]
-fn api_versions_v3_empty_body_returns_invalid_request() {
+#[tokio::test]
+async fn api_versions_v3_empty_body_returns_invalid_request() {
let body = handle_request(API_KEY_API_VERSIONS, 3, Bytes::new(), &default_broker())
+ .await
.expect_response("ApiVersions has a top-level error_code");
let mut d = Decoder::new(body);
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
}
-#[test]
-fn unsupported_metadata_version_closes_connection() {
+#[tokio::test]
+async fn unsupported_metadata_version_closes_connection() {
// Above max: a clamped v9 body is unparsable to a v99 client, so Close is the honest contract.
assert!(
handle_request(
@@ -158,6 +164,7 @@ fn unsupported_metadata_version_closes_connection() {
build_metadata_flexible_request_v10(&["orders"]),
&default_broker(),
)
+ .await
.is_close(),
"Metadata above supported max must close rather than return a clamped body"
);
@@ -165,9 +172,9 @@ fn unsupported_metadata_version_closes_connection() {
// ── Misc ────────────────────────────────────────────────────────────────────
-#[test]
-fn unknown_api_key_closes_connection() {
- let outcome = handle_request(999, 0, Bytes::new(), &default_broker());
+#[tokio::test]
+async fn unknown_api_key_closes_connection() {
+ let outcome = handle_request(999, 0, Bytes::new(), &default_broker()).await;
assert!(
outcome.is_close(),
"unknown api_key must close (no parseable response schema)"
@@ -182,9 +189,10 @@ fn version_support_table_is_applied() {
assert!(!is_supported_version(API_KEY_METADATA, -1));
}
-#[test]
-fn apiversions_unsupported_version_uses_v0_encoding_without_throttle() {
+#[tokio::test]
+async fn apiversions_unsupported_version_uses_v0_encoding_without_throttle() {
let body = handle_request(API_KEY_API_VERSIONS, 99, Bytes::new(), &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
// v0: error_code(2) + api_keys i32 count(4) + 6 entries × 6 bytes = 42 - no throttle_time_ms.
assert_eq!(body.len(), 42);
@@ -194,8 +202,8 @@ fn apiversions_unsupported_version_uses_v0_encoding_without_throttle() {
assert_eq!(d.remaining(), 36);
}
-#[test]
-fn produce_malformed_body_with_acks_one_stays_silent() {
+#[tokio::test]
+async fn produce_malformed_body_with_acks_one_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
@@ -207,13 +215,15 @@ fn produce_malformed_body_with_acks_one_stays_silent() {
0x00, 0x00, 0x00, 0x01, // one topic
]);
assert!(
- handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).is_no_response(),
+ handle_request(API_KEY_PRODUCE, 3, body, &default_broker())
+ .await
+ .is_no_response(),
"malformed Produce body must stay silent regardless of acks"
);
}
-#[test]
-fn fetch_malformed_body_returns_invalid_request() {
+#[tokio::test]
+async fn fetch_malformed_body_returns_invalid_request() {
let response = handle_request(
API_KEY_FETCH,
12,
@@ -224,14 +234,15 @@ fn fetch_malformed_body_returns_invalid_request() {
]),
&default_broker(),
)
+ .await
.expect_response("fetch must return error response");
let mut d = Decoder::new(response);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
}
-#[test]
-fn list_offsets_malformed_body_returns_invalid_request() {
+#[tokio::test]
+async fn list_offsets_malformed_body_returns_invalid_request() {
let response = handle_request(
API_KEY_LIST_OFFSETS,
6,
@@ -243,6 +254,7 @@ fn list_offsets_malformed_body_returns_invalid_request() {
]),
&default_broker(),
)
+ .await
.expect_response("list offsets must return error response");
let mut d = Decoder::new(response);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
@@ -256,8 +268,8 @@ fn list_offsets_malformed_body_returns_invalid_request() {
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
}
-#[test]
-fn create_topics_malformed_body_returns_invalid_request() {
+#[tokio::test]
+async fn create_topics_malformed_body_returns_invalid_request() {
let response = handle_request(
API_KEY_CREATE_TOPICS,
5,
@@ -267,6 +279,7 @@ fn create_topics_malformed_body_returns_invalid_request() {
]),
&default_broker(),
)
+ .await
.expect_response("create topics must return error response");
let mut d = Decoder::new(response);
assert_eq!(d.read_i32().unwrap(), 0);
@@ -278,8 +291,8 @@ fn create_topics_malformed_body_returns_invalid_request() {
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
}
-#[test]
-fn metadata_null_topic_name_closes_connection() {
+#[tokio::test]
+async fn metadata_null_topic_name_closes_connection() {
assert!(
handle_request(
API_KEY_METADATA,
@@ -290,6 +303,7 @@ fn metadata_null_topic_name_closes_connection() {
]),
&default_broker(),
)
+ .await
.is_close(),
"null topic name cannot be echoed in a Metadata response"
);
@@ -297,8 +311,8 @@ fn metadata_null_topic_name_closes_connection() {
// ── Full handler regression - every scoped API key x version through `handle_request` ──
-#[test]
-fn handle_request_succeeds_for_every_supported_version_with_fixture() {
+#[tokio::test]
+async fn handle_request_succeeds_for_every_supported_version_with_fixture() {
for &(api_key, name, min_ver, max_ver) in scope::SCOPED_API_KEYS {
if api_key == API_KEY_METADATA {
for version in min_ver..=max_ver {
@@ -308,6 +322,7 @@ fn handle_request_succeeds_for_every_supported_version_with_fixture() {
build_metadata_all_topics_legacy(version)
};
let resp = handle_request(api_key, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
assert!(
!resp.is_empty(),
@@ -324,6 +339,7 @@ fn handle_request_succeeds_for_every_supported_version_with_fixture() {
Bytes::new()
};
let resp = handle_request(api_key, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
assert!(
!resp.is_empty(),
@@ -338,6 +354,7 @@ fn handle_request_succeeds_for_every_supported_version_with_fixture() {
continue;
};
let resp = handle_request(api_key, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
assert!(
!resp.is_empty(),
@@ -347,13 +364,14 @@ fn handle_request_succeeds_for_every_supported_version_with_fixture() {
}
}
-#[test]
-fn produce_stub_response_returns_retriable_not_leader() {
+#[tokio::test]
+async fn produce_stub_response_returns_retriable_not_leader() {
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())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let flexible = version >= 9;
let mut d = Decoder::new(resp);
@@ -375,13 +393,14 @@ fn produce_stub_response_returns_retriable_not_leader() {
}
}
-#[test]
-fn fetch_stub_response_returns_retriable_not_leader() {
+#[tokio::test]
+async fn fetch_stub_response_returns_retriable_not_leader() {
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())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let flexible = version >= 12;
let mut d = Decoder::new(resp);
@@ -415,12 +434,13 @@ fn fetch_stub_response_returns_retriable_not_leader() {
/// v9 legacy encoding with a populated forgotten-topics list, and v12 flexible encoding adding a
/// rack id (introduced at v11) on top of that. Both must decode and reach the same stub outcome
/// as an empty-topics request.
-#[test]
-fn fetch_decodes_request_with_forgotten_topics_and_rack_id() {
+#[tokio::test]
+async fn fetch_decodes_request_with_forgotten_topics_and_rack_id() {
for (version, rack_id) in [(9i16, None), (12i16, Some("rack-a"))] {
let body =
wire::build_fetch_request_with_sections(version, "orders", 2, Some("stale"), rack_id);
let resp = handle_request(API_KEY_FETCH, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let flexible = version >= 12;
let mut d = Decoder::new(resp);
@@ -445,13 +465,14 @@ fn fetch_decodes_request_with_forgotten_topics_and_rack_id() {
}
}
-#[test]
-fn list_offsets_stub_response_returns_retriable_not_leader() {
+#[tokio::test]
+async fn list_offsets_stub_response_returns_retriable_not_leader() {
for version in 1i16..=6 {
let Some(body) = load_fixture_body_or_skip(2, "ListOffsets", version) else {
continue;
};
let resp = handle_request(API_KEY_LIST_OFFSETS, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let flexible = version >= 6;
let mut d = Decoder::new(resp);
@@ -479,10 +500,11 @@ fn list_offsets_stub_response_returns_retriable_not_leader() {
/// `list_offsets_stub_response_returns_retriable_not_leader` above is fixture-driven; this
/// exercises `encode_list_offsets_response`'s per-partition echo loop with a hand-built request
/// carrying a real (non-empty) topic/partition, independent of any `.bin` fixture.
-#[test]
-fn list_offsets_decodes_request_with_real_topic_and_partition() {
+#[tokio::test]
+async fn list_offsets_decodes_request_with_real_topic_and_partition() {
let body = wire::build_list_offsets_branch_request(6, "orders", 2);
let resp = handle_request(API_KEY_LIST_OFFSETS, 6, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(resp);
let _throttle = d.read_i32().unwrap();
@@ -544,8 +566,8 @@ fn read_broker_flexible(d: &mut Decoder) -> (String, i32) {
(host, port)
}
-#[test]
-fn metadata_corrupt_partial_body_closes_connection() {
+#[tokio::test]
+async fn metadata_corrupt_partial_body_closes_connection() {
assert!(
handle_request(
API_KEY_METADATA,
@@ -553,19 +575,21 @@ fn metadata_corrupt_partial_body_closes_connection() {
Bytes::from_static(&[0x00, 0x00]),
&default_broker(),
)
+ .await
.is_close(),
"truncated Metadata body must close; there is no top-level error field"
);
}
-#[test]
-fn metadata_v0_empty_topics_stub_broker() {
+#[tokio::test]
+async fn metadata_v0_empty_topics_stub_broker() {
let body = handle_request(
API_KEY_METADATA,
0,
metadata_request_legacy(0, 0),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let (host, port) = read_broker_legacy(&mut d);
@@ -579,14 +603,15 @@ fn metadata_v0_empty_topics_stub_broker() {
);
}
-#[test]
-fn metadata_v0_three_topics_each_unknown() {
+#[tokio::test]
+async fn metadata_v0_three_topics_each_unknown() {
let body = handle_request(
API_KEY_METADATA,
0,
metadata_request_legacy(0, 3),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let _ = read_broker_legacy(&mut d);
@@ -601,14 +626,15 @@ fn metadata_v0_three_topics_each_unknown() {
}
}
-#[test]
-fn metadata_v1_includes_controller_id() {
+#[tokio::test]
+async fn metadata_v1_includes_controller_id() {
let body = handle_request(
API_KEY_METADATA,
1,
metadata_request_legacy(1, 0),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
// Metadata v1 has no throttle_time_ms (added in v3).
@@ -618,14 +644,15 @@ fn metadata_v1_includes_controller_id() {
assert_eq!(controller, 1);
}
-#[test]
-fn metadata_v2_includes_cluster_id_field() {
+#[tokio::test]
+async fn metadata_v2_includes_cluster_id_field() {
let body = handle_request(
API_KEY_METADATA,
2,
metadata_request_legacy(2, 0),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let _ = read_broker_legacy(&mut d);
@@ -635,8 +662,8 @@ fn metadata_v2_includes_cluster_id_field() {
assert_eq!(d.read_i32().unwrap(), 0);
}
-#[test]
-fn metadata_all_legacy_versions_produce_valid_response() {
+#[tokio::test]
+async fn metadata_all_legacy_versions_produce_valid_response() {
for version in 0i16..=8 {
let body = handle_request(
API_KEY_METADATA,
@@ -644,6 +671,7 @@ fn metadata_all_legacy_versions_produce_valid_response() {
metadata_request_legacy(version, 1),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
if version >= 3 {
@@ -664,14 +692,15 @@ fn metadata_all_legacy_versions_produce_valid_response() {
}
}
-#[test]
-fn metadata_v9_flexible_encoding() {
+#[tokio::test]
+async fn metadata_v9_flexible_encoding() {
let body = handle_request(
API_KEY_METADATA,
9,
metadata_request_flexible(2),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let _throttle = d.read_i32().unwrap();
@@ -701,14 +730,15 @@ fn metadata_v9_flexible_encoding() {
assert_eq!(d.remaining(), 0);
}
-#[test]
-fn metadata_v8_includes_authorized_operations_legacy() {
+#[tokio::test]
+async fn metadata_v8_includes_authorized_operations_legacy() {
let body = handle_request(
API_KEY_METADATA,
8,
metadata_request_legacy(8, 1),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let _throttle = d.read_i32().unwrap();
@@ -726,13 +756,14 @@ fn metadata_v8_includes_authorized_operations_legacy() {
assert_eq!(d.remaining(), 0);
}
-#[test]
-fn create_topics_stub_response_returns_not_controller() {
+#[tokio::test]
+async fn create_topics_stub_response_returns_not_controller() {
for version in 2i16..=5 {
let Some(body) = load_fixture_body_or_skip(19, "CreateTopics", version) else {
continue;
};
let resp = handle_request(API_KEY_CREATE_TOPICS, version, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let flexible = version >= 5;
let mut d = Decoder::new(resp);
@@ -757,10 +788,11 @@ fn create_topics_stub_response_returns_not_controller() {
/// `create_topics_stub_response_returns_not_controller` above is fixture-driven; this exercises
/// `encode_create_topics_response`'s per-topic echo with a hand-built request carrying a real
/// topic, replica assignment, and config entry, independent of any `.bin` fixture.
-#[test]
-fn create_topics_decodes_request_with_real_topic_and_assignment() {
+#[tokio::test]
+async fn create_topics_decodes_request_with_real_topic_and_assignment() {
let body = wire::build_create_topics_request_with_sections(5, "orders");
let resp = handle_request(API_KEY_CREATE_TOPICS, 5, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(resp);
let _throttle = d.read_i32().unwrap();
@@ -776,20 +808,22 @@ fn create_topics_decodes_request_with_real_topic_and_assignment() {
// ── Produce acks=0 (broker must stay silent even on a malformed body) ──────
-#[test]
-fn produce_acks_zero_malformed_body_stays_silent() {
+#[tokio::test]
+async fn produce_acks_zero_malformed_body_stays_silent() {
// topics array declares 1 element but no topic data follows - decode fails after acks=0
// was on the wire, though `kafka_protocol`'s one-shot decode no longer exposes that acks
// was read. The handler must stay silent regardless.
let body = build_produce_v3_body(0, 1);
assert!(
- handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).is_no_response(),
+ handle_request(API_KEY_PRODUCE, 3, body, &default_broker())
+ .await
+ .is_no_response(),
"handler must not respond when acks=0 even if decode fails after acks"
);
}
-#[test]
-fn produce_acks_zero_stays_silent_on_advertised_but_unsupported_version() {
+#[tokio::test]
+async fn produce_acks_zero_stays_silent_on_advertised_but_unsupported_version() {
// ApiVersions advertises Produce min=0 (KAFKA-18659) while the firewall's real floor is 3
// (SUPPORTED_RANGES). A spec-compliant client can legitimately send v0-2 with acks=0; the
// firewall must not run before acks is decoded, or this silence contract breaks for exactly
@@ -797,7 +831,9 @@ fn produce_acks_zero_stays_silent_on_advertised_but_unsupported_version() {
for version in 0i16..3 {
let body = build_produce_legacy_request(version, 0, None, None);
assert!(
- handle_request(API_KEY_PRODUCE, version, body, &default_broker()).is_no_response(),
+ handle_request(API_KEY_PRODUCE, version, body, &default_broker())
+ .await
+ .is_no_response(),
"Produce v{version} acks=0 must stay silent even though the firewall doesn't accept v{version}"
);
}
@@ -805,11 +841,12 @@ fn produce_acks_zero_stays_silent_on_advertised_but_unsupported_version() {
// ── Metadata topic name echo (must not hardcode a placeholder topic name) ──
-#[test]
-fn metadata_v1_echoes_requested_topic_name_in_response() {
+#[tokio::test]
+async fn metadata_v1_echoes_requested_topic_name_in_response() {
let topic = "orders";
let request = build_metadata_legacy_request(&[topic]);
let body = handle_request(API_KEY_METADATA, 1, request, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
@@ -826,14 +863,15 @@ fn metadata_v1_echoes_requested_topic_name_in_response() {
// ── Metadata (spec + SCOPE.md coverage) ─────────────────────────────────────
-#[test]
-fn metadata_v0_empty_topics_returns_zero_length_topic_array() {
+#[tokio::test]
+async fn metadata_v0_empty_topics_returns_zero_length_topic_array() {
let body = handle_request(
API_KEY_METADATA,
0,
build_metadata_legacy_request(&[]),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
let _brokers = d.read_i32().unwrap();
@@ -844,27 +882,29 @@ fn metadata_v0_empty_topics_returns_zero_length_topic_array() {
assert_eq!(d.remaining(), 0);
}
-#[test]
-fn metadata_v3_includes_throttle_time_ms_before_brokers() {
+#[tokio::test]
+async fn metadata_v3_includes_throttle_time_ms_before_brokers() {
let body = handle_request(
API_KEY_METADATA,
3,
build_metadata_legacy_request_for_version(3, &[]),
&default_broker(),
)
+ .await
.expect_response("metadata v3 empty topics must succeed");
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0, "throttle_time_ms");
}
-#[test]
-fn metadata_v9_flexible_empty_topics_returns_zero_topics() {
+#[tokio::test]
+async fn metadata_v9_flexible_empty_topics_returns_zero_topics() {
let body = handle_request(
API_KEY_METADATA,
9,
build_metadata_flexible_request(&[]),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
d.read_i32().unwrap(); // throttle
@@ -886,8 +926,8 @@ fn metadata_v9_flexible_empty_topics_returns_zero_topics() {
assert_eq!(topic_count, 0);
}
-#[test]
-fn metadata_v9_flexible_echoes_each_requested_topic_name() {
+#[tokio::test]
+async fn metadata_v9_flexible_echoes_each_requested_topic_name() {
let topics = ["orders", "payments", "inventory"];
let body = handle_request(
API_KEY_METADATA,
@@ -895,6 +935,7 @@ fn metadata_v9_flexible_echoes_each_requested_topic_name() {
build_metadata_flexible_request(&topics),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
@@ -951,8 +992,8 @@ fn metadata_v9_flexible_echoes_each_requested_topic_name() {
);
}
-#[test]
-fn metadata_v1_legacy_multiple_topics_echo_names() {
+#[tokio::test]
+async fn metadata_v1_legacy_multiple_topics_echo_names() {
let topics = ["alpha", "beta"];
let body = handle_request(
API_KEY_METADATA,
@@ -960,6 +1001,7 @@ fn metadata_v1_legacy_multiple_topics_echo_names() {
build_metadata_legacy_request(&topics),
&default_broker(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
diff --git a/gateways/kafka/tests/bridge_iggy_integration_tests.rs b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
index 649b666c1..286f58ad7 100644
--- a/gateways/kafka/tests/bridge_iggy_integration_tests.rs
+++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
@@ -21,15 +21,10 @@
//! module invoked from a real (non-unit) test rather than only compiled.
use std::collections::HashMap;
-use std::fs::{File, OpenOptions};
-use std::path::PathBuf;
-use std::process::{Child, Command};
-use std::sync::OnceLock;
use std::time::Duration;
use iggy::prelude::{
- AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder, IggyMessage,
- MessageClient, Partitioning, StreamClient, TopicClient,
+ Identifier, IggyMessage, MessageClient, Partitioning, StreamClient, TopicClient,
};
use secrecy::SecretString;
use serial_test::serial;
@@ -38,395 +33,10 @@ use iggy_gateway_kafka::bridge::{
BridgeError, IggyBridge, IggyBridgeConfig, TopicMapping, TopicOverride,
};
-/// Desired slot count - the actual count `port_band()` computes may be smaller if the machine's
-/// real ephemeral range leaves little room, but this crate never needs more than a handful live
-/// at once (this binary's own tests already run one at a time - see `.config/nextest.toml`'s
-/// `kafka_bridge` test-group).
-const DESIRED_SLOTS: u16 = 200;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
-/// Lowest port `port_band()` will ever place its band at - clear of the well-known/privileged
-/// range (0-1023).
-const MIN_CANDIDATE_PORT: u16 = 10000;
-
-/// Ephemeral range assumed when the kernel's own can't be read (no `/proc` - e.g. macOS),
-/// matching the Linux default. Mirrors `core/integration`'s own `port_reserver.rs`.
-const DEFAULT_EPHEMERAL_RANGE: (u16, u16) = (32768, 60999);
-
-const IP_LOCAL_PORT_RANGE: &str = "/proc/sys/net/ipv4/ip_local_port_range";
-
-fn parse_ephemeral_range(range: &str) -> Option<(u16, u16)> {
- let mut bounds = range.split_whitespace();
- let floor: u16 = bounds.next()?.parse().ok()?;
- let ceiling: u16 = bounds.next()?.parse().ok()?;
- Some((floor, ceiling))
-}
-
-/// The range the kernel picks `bind(0)` ports from, read once.
-fn ephemeral_range() -> (u16, u16) {
- static RANGE: OnceLock<(u16, u16)> = OnceLock::new();
- *RANGE.get_or_init(|| {
- std::fs::read_to_string(IP_LOCAL_PORT_RANGE)
- .ok()
- .as_deref()
- .and_then(parse_ephemeral_range)
- .unwrap_or(DEFAULT_EPHEMERAL_RANGE)
- })
-}
-
-/// First port and slot count of a band clear of the kernel's ephemeral range: below the floor by
-/// preference (matches `core/integration`'s own `port_reserver.rs`), above the ceiling when the
-/// floor leaves no room there.
-///
-/// A *hardcoded* band, chosen once without reading the real range (this file's own prior version:
-/// `15000..15199`, on the unverified assumption that every real deployment's floor sits above
-/// it), is wrong on any box tuned wider - `net.ipv4.ip_local_port_range = "1024 65535"` swallows
-/// that whole band. `flock` alone doesn't save it either: it is advisory, so it only excludes
-/// another `PortGuard`-based process, never an unrelated `bind(0)` elsewhere on the box landing on
-/// the same number from the kernel's own ephemeral pool.
-///
-/// Known residual gap, not fixed by the "matches `port_reserver.rs`" preference above: this
-/// `PortGuard` locks its own directory (`iggy-kafka-gateway-test-port-locks`), disjoint from
-/// `port_reserver.rs`'s (`iggy-test-port-locks`), so the two share no lock state even when their
-/// *bands* land on the same numbers. That only happens when the ephemeral floor leaves neither
-/// allocator room below it - `port_reserver.rs` needs room below its own, higher `BAND_START`
-/// (20000) to use the below-floor branch at all, while this function only needs room below
-/// `MIN_CANDIDATE_PORT` (10000) - so a floor at or under 10000 pushes *both* to their
-/// above-the-ceiling fallback, anchored at the identical `ceiling + 1`: this band takes the first
-/// [`DESIRED_SLOTS`] of it, `port_reserver.rs` takes the entire rest up to `u16::MAX`. A port
-/// this crate's tests reserve there is invisible to `port_reserver.rs`'s own lock, and vice versa,
-/// so two suites in the same workspace could hand out the same port under two different locks.
-/// `core/integration/src/harness/mod.rs`'s `mod port_reserver;` is private, so sharing its lock
-/// directory (the complete fix) needs a visibility change there first; narrow enough (an
-/// ephemeral floor this low is uncommon - the Linux default is 32768) that this is disclosed
-/// rather than worked around here.
-fn port_band() -> (u16, u16) {
- static BAND: OnceLock<(u16, u16)> = OnceLock::new();
- *BAND.get_or_init(|| {
- let (floor, ceiling) = ephemeral_range();
- band_for(floor, ceiling)
- })
-}
-
-/// Pure band-selection logic, split out from [`port_band`] so it's testable against synthetic
-/// ranges without needing to fake `/proc` contents.
-fn band_for(floor: u16, ceiling: u16) -> (u16, u16) {
- if floor > MIN_CANDIDATE_PORT {
- let room = floor - MIN_CANDIDATE_PORT;
- return (MIN_CANDIDATE_PORT, room.min(DESIRED_SLOTS));
- }
- let Some(start) = ceiling.checked_add(1) else {
- panic!(
- "kernel ephemeral range [{floor}, {ceiling}] leaves no room for a test port band \
- below {MIN_CANDIDATE_PORT} nor above {ceiling} - narrow the range, e.g. \
- sysctl -w net.ipv4.ip_local_port_range='32768 60999'"
- );
- };
- // start >= 1 (ceiling + 1), so this never overflows u16.
- let room = u16::MAX - start + 1;
- (start, room.min(DESIRED_SLOTS))
-}
-
-#[test]
-fn given_a_readable_range_when_parsed_should_take_both_bounds() {
- assert_eq!(
- parse_ephemeral_range("32768\t60999\n"),
- Some((32768, 60999))
- );
- assert_eq!(parse_ephemeral_range("1024 65535"), Some((1024, 65535)));
- assert_eq!(parse_ephemeral_range(""), None);
- assert_eq!(parse_ephemeral_range("32768"), None);
- assert_eq!(parse_ephemeral_range("garbage 60999"), None);
-}
-
-#[test]
-fn given_room_below_the_floor_when_choosing_a_band_should_take_it() {
- let (start, slots) = band_for(32768, 60999);
- assert_eq!(start, MIN_CANDIDATE_PORT);
- assert!(
- start + slots <= 32768,
- "band [{start}, {}] runs into the ephemeral floor",
- start + slots - 1
- );
-}
-
-/// A box tuned `net.ipv4.ip_local_port_range = "1024 60999"` is exactly the case the prior
-/// hardcoded `15000..15199` band silently broke under: room below the floor down to
-/// `MIN_CANDIDATE_PORT` is gone, but the ceiling still leaves room above it.
-#[test]
-fn given_no_room_below_the_floor_when_choosing_a_band_should_fall_back_above_the_ceiling() {
- let (start, slots) = band_for(1024, 60999);
- assert!(slots > 0, "must find room above the ceiling, not just fail");
- assert!(
- start > 60999,
- "band must start above the ephemeral ceiling, got {start}"
- );
-}
-
-#[test]
-#[should_panic(expected = "leaves no room")]
-fn given_no_room_on_either_side_when_choosing_a_band_should_fail_loudly() {
- band_for(1024, u16::MAX);
-}
-
-/// Exclusive claim on one port, released (and the port freed for reuse) when dropped - including
-/// on an unclean process exit, since the OS drops the `flock` with the file descriptor. Unlike
-/// bind-then-drop, this process itself never binds the port before `iggy-server` does; see
-/// `port_band`'s doc comment for the limits of that guarantee against *other* processes.
-struct PortGuard {
- port: u16,
- _lock: File,
-}
-
-impl PortGuard {
- fn acquire() -> Self {
- let (band_start, band_slots) = port_band();
- let lock_dir = std::env::temp_dir().join("iggy-kafka-gateway-test-port-locks");
- std::fs::create_dir_all(&lock_dir).expect("create port lock dir");
- for offset in 0..band_slots {
- let port = band_start + offset;
- let path = lock_dir.join(format!("{port}.lock"));
- let Ok(file) = OpenOptions::new()
- .read(true)
- .write(true)
- .create(true)
- .truncate(false)
- .open(&path)
- else {
- continue;
- };
- if file.try_lock().is_ok() {
- return Self { port, _lock: file };
- }
- }
- panic!(
- "no free port slot in [{band_start}, {}]",
- band_start + band_slots - 1
- );
- }
-}
-
-/// Locates the already-built `iggy-server` binary. Does not build it - see this crate's
-/// `docs/TEST_SUITE.md` for the prerequisite, the same one `core/integration`'s own
-/// server-spawning tests already carry.
-///
-/// Walks up from this test binary's own path (`env::current_exe()`) rather than calling
-/// `assert_cmd::cargo::cargo_bin("iggy-server")` directly: `cargo_bin`'s `CARGO_BIN_EXE_*`
-/// env-var lookup only resolves for binaries owned by *this* package, so for `iggy-server`
-/// (a different package) it always falls through to `cargo_bin`'s own `legacy_cargo_bin`, which
-/// panics with its own internal message - not the "binary not found" text `TEST_SUITE.md`
-/// promises - when the binary is not yet built (`CARGO_BIN_EXE_iggy-server` is unset, so its
-/// panic path prints that plus every binary name it *did* find, not a helpful next step). Ancestor
-/// directories of `current_exe()` reach the same `target/<profile>/` directory
-/// `legacy_cargo_bin` derives - correct under `CARGO_TARGET_DIR`, `--release`, or a `--target`
-/// triple subdirectory for the identical reason: it reads back from where cargo actually placed
-/// *this* binary, not a guessed path - so checking existence there first, before ever calling
-/// into `assert_cmd`, gives this crate's own message instead of a Cargo internal one. An earlier
-/// version of this function drove `cargo build --package server --bin iggy-server` directly from
-/// every server-spawning test (nextest runs each as its own process, so the `OnceLock` memoized
-/// nothing across them), serialized by `.config/nextest.toml`'s `kafka_bridge` test-group but
-/// still real, avoidable per-test overhead this version has none of.
-///
-/// # Panics
-///
-/// Panics if `iggy-server` cannot be found alongside this test binary's own target directory.
-fn iggy_server_binary() -> PathBuf {
- let current_exe = std::env::current_exe().expect("resolve this test binary's own path");
- let binary_name = format!("iggy-server{}", std::env::consts::EXE_SUFFIX);
- current_exe
- .ancestors()
- .map(|dir| dir.join(&binary_name))
- .find(|candidate| candidate.is_file())
- .unwrap_or_else(|| {
- panic!(
- "iggy-server binary not found near {} - build it first with \
- `cargo build --package server --bin iggy-server`",
- current_exe.display()
- )
- })
-}
-
-struct TestServer {
- child: Child,
- address: String,
- password: String,
- _port_guard: PortGuard,
-}
-
-impl TestServer {
- /// Spawns `iggy-server` with the default `iggy`/`iggy` root credentials. See
- /// [`Self::spawn_with_password`] for the general form.
- async fn spawn(data_dir: &std::path::Path) -> Self {
- Self::spawn_with_password(data_dir, "iggy").await
- }
-
- /// Spawns `iggy-server` with an isolated temp data dir, a locked TCP port, and the given root
- /// password, then blocks until its listener is ready or the startup budget is exhausted.
- ///
- /// A dedicated password parameter (not just the `spawn()` default everywhere) lets a
- /// password-shaped regression test (special characters, say) reuse this harness's
- /// `PortGuard`/graceful-`Drop`/`wait_ready` machinery instead of hand-rolling a second,
- /// `Drop`-less spawn: a `Drop`-less copy has no guard to run `graceful_kill` on a panic before
- /// its assertions, orphaning a process that still holds its `PortGuard` slot, which then
- /// fails the next test that draws that slot to bind.
- async fn spawn_with_password(data_dir: &std::path::Path, password: &str) -> Self {
- let port_guard = PortGuard::acquire();
- let address = format!("127.0.0.1:{}", port_guard.port);
-
- let mut command = Command::new(iggy_server_binary());
- command
- .env("IGGY_PATH", data_dir.display().to_string())
- .env("IGGY_TCP_ADDRESS", &address)
- .env("IGGY_HTTP_ENABLED", "false")
- .env("IGGY_QUIC_ENABLED", "false")
- // WebSocket defaults to enabled on a fixed 127.0.0.1:8092 (config.toml), unlike TCP
- // which reads a per-test port from PortGuard - every spawned server here would fight
- // over that one port otherwise, and a bind failure aborts boot.
- .env("IGGY_WEBSOCKET_ENABLED", "false")
- // Matches core/integration's own spawned-server harness (harness/handle/server.rs):
- // pinned shards (config.toml default: pin_cores = true) of concurrently running
- // servers pile onto the same cores and starve each other. This crate's own tests are
- // serialized against each other (.config/nextest.toml's kafka_bridge test-group -
- // #[serial] alone does not survive nextest, which runs each test as its own process),
- // but a spawned server here can still land alongside a pinned server from a different
- // package's test in the same nextest run.
- .env("IGGY_SHARDING_PIN_CORES", "false")
- // Only half of core/integration's own pairing: unpinning cores stops a server from
- // being pinned to a specific range, but config.toml's own default
- // (cpu_allocation = "numa:auto") still sizes the shard pool to the whole machine with
- // nothing set here - every server this crate's tests spawn would still compete for
- // the entire box against unrelated packages' tests in the same nextest run. A small
- // fixed range (matching the same harness's own fallback for an unreadable core count)
- // caps that instead.
- .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
- // `--with-default-root-credentials` is off by default (args.rs) - without these,
- // a fresh server provisions no loginable root user at all, and every bridge connect
- // attempt fails with "invalid credentials" no matter what this test passes.
- .env("IGGY_ROOT_USERNAME", "iggy")
- .env("IGGY_ROOT_PASSWORD", password);
- let child = command.spawn().expect("spawn iggy-server");
-
- let mut server = Self {
- child,
- address,
- password: password.to_string(),
- _port_guard: port_guard,
- };
- server.wait_ready().await;
- server
- }
-
- /// Polls with a bare TCP connect, not a full `IggyBridge::connect`: the latter carries
- /// `RECONNECTION_RETRIES` (3 dials at ~1s apart) on every failed attempt, so a poll loop built
- /// on it pays several real seconds per iteration instead of running at its own 100ms cadence,
- /// and a *successful* poll iteration would authenticate a client and then drop it without
- /// `close()`, leaking a session server-side. A TCP accept slightly ahead of the app being
- /// ready to authenticate is fine - every caller's own subsequent `IggyBridge::connect` already
- /// retries a few times, which covers the last few hundred ms of that gap.
- ///
- /// Checks `try_wait()` every iteration: a server that fails at boot (bad config, a port
- /// stolen between `PortGuard::acquire` and its own bind) exits almost immediately, and without
- /// this check that reads as "still starting" for the full 30s budget - the eventual failure
- /// names the wrong cause ("did not become ready") instead of the real one (exited early, with
- /// whatever it printed before dying).
- async fn wait_ready(&mut self) {
- let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
- loop {
- if let Some(status) = self.child.try_wait().expect("poll child status") {
- panic!(
- "iggy-server at {} exited during startup with {status}",
- self.address
- );
- }
- if tokio::net::TcpStream::connect(self.address.as_str())
- .await
- .is_ok()
- {
- return;
- }
- assert!(
- tokio::time::Instant::now() < deadline,
- "iggy-server at {} did not become ready within the startup budget",
- self.address
- );
- tokio::time::sleep(Duration::from_millis(100)).await;
- }
- }
-
- fn test_config(&self) -> IggyBridgeConfig {
- IggyBridgeConfig {
- address: self.address.clone(),
- username: "iggy".to_string(),
- password: SecretString::from(self.password.clone()),
- topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new())
- .expect("valid mapping for this test's fixture data"),
- }
- }
-}
-
-/// SIGTERM, wait up to `SIGTERM_TIMEOUT`, then SIGKILL if it hasn't exited. Mirrors
-/// `core/integration`'s own `harness::handle::common::graceful_kill` (not reused directly - that
-/// crate is heavyweight, and pulling it in for one function isn't worth it; this crate already
-/// depends directly on `iggy`/`core/sdk`, so `core/integration` wouldn't add a *new*
-/// core/sdk-change-reruns-these-tests edge, just an unrelated dependency). A bare SIGKILL skips
-/// `iggy-server`'s shutdown path entirely, which is a materially different exit than what the
-/// binary is actually built to do on `SIGTERM` - `.kill()` alone bypassed that in every test run
-/// before this fix.
-const SIGTERM_TIMEOUT: Duration = Duration::from_secs(5);
-const SIGKILL_POLL_INTERVAL: Duration = Duration::from_millis(50);
-
-fn graceful_kill(child: &mut Child) {
- // `wait_ready` calls `try_wait()` too (and panics on an early exit, unwinding into this via
- // `Drop`) - if it already reaped the child, `child.id()` is a PID the OS is free to hand to an
- // unrelated process by the time we get here, and a raw `libc::kill` (unlike `std::Child::kill`,
- // which checks its own cached exit status first and no-ops instead) has no such guard against
- // signaling that PID anyway. Checking here first closes the same gap `std` already closes for
- // its own `kill`.
- if matches!(child.try_wait(), Ok(Some(_))) {
- return;
- }
-
- let pid = child.id() as libc::pid_t;
- // Safety: `pid` is this process's own live child, confirmed by the `try_wait` check above;
- // sending it a signal is exactly what `Child::kill` itself does internally.
- unsafe {
- libc::kill(pid, libc::SIGTERM);
- }
-
- let deadline = std::time::Instant::now() + SIGTERM_TIMEOUT;
- while std::time::Instant::now() < deadline {
- match child.try_wait() {
- Ok(None) => std::thread::sleep(SIGKILL_POLL_INTERVAL),
- Ok(Some(_)) | Err(_) => return,
- }
- }
-
- let _ = child.kill();
-}
-
-impl Drop for TestServer {
- fn drop(&mut self) {
- graceful_kill(&mut self.child);
- let _ = self.child.wait();
- }
-}
-
-/// Builds and connects a raw `IggyClient` against `server` - for producing test data directly,
-/// independent of the `IggyBridge` under test. Uses the fluent builder, not a hand-built
-/// `iggy://user:pass@host` string: `ConnectionString` splits on `@` then `:`, which breaks for
-/// any password containing either character.
-async fn raw_client(server: &TestServer) -> IggyClient {
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(server.address.clone())
- .with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword(
- "iggy".to_string(),
- SecretString::from("iggy"),
- )))
- .build()
- .expect("build raw test client");
- client.connect().await.expect("connect raw test client");
- client
-}
+use iggy_server::{PortGuard, TestServer, raw_client};
#[tokio::test]
#[serial]
diff --git a/gateways/kafka/tests/broker_advertise_tests.rs b/gateways/kafka/tests/broker_advertise_tests.rs
index a4be84574..fc8b6c6e4 100644
--- a/gateways/kafka/tests/broker_advertise_tests.rs
+++ b/gateways/kafka/tests/broker_advertise_tests.rs
@@ -34,8 +34,8 @@ fn default_matches_standard_gateway_port() {
assert_eq!(b.port, 9093);
}
-#[test]
-fn metadata_reflects_broker_addr() {
+#[tokio::test]
+async fn metadata_reflects_broker_addr() {
let broker = BrokerAdvertise {
host: "203.0.113.7".to_string(),
port: 9093,
@@ -43,6 +43,7 @@ fn metadata_reflects_broker_addr() {
let mut req = Encoder::with_capacity(4);
req.write_i32(0);
let body = handle_request(API_KEY_METADATA, 0, req.freeze(), &broker)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
diff --git a/gateways/kafka/tests/common/iggy_server.rs b/gateways/kafka/tests/common/iggy_server.rs
new file mode 100644
index 000000000..b2dbd3a71
--- /dev/null
+++ b/gateways/kafka/tests/common/iggy_server.rs
@@ -0,0 +1,411 @@
+// 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.
+
+//! Real `iggy-server` process harness, shared by every suite in this crate that needs one.
+//!
+//! Included with `#[path = "common/iggy_server.rs"]`, so its own unit tests run once per
+//! including binary. They are pure functions over a synthetic port range, so the repeat costs
+//! nothing.
+
+#![allow(dead_code)]
+
+use std::collections::HashMap;
+use std::fs::{File, OpenOptions};
+use std::path::PathBuf;
+use std::process::{Child, Command};
+use std::sync::OnceLock;
+use std::time::Duration;
+
+use iggy::prelude::{AutoLogin, Client, Credentials, IggyClient, IggyClientBuilder};
+use secrecy::SecretString;
+
+use iggy_gateway_kafka::bridge::{IggyBridgeConfig, TopicMapping};
+
+/// Desired slot count - the actual count `port_band()` computes may be smaller if the machine's
+/// real ephemeral range leaves little room, but this crate never needs more than a handful live
+/// at once (this binary's own tests already run one at a time - see `.config/nextest.toml`'s
+/// `kafka_bridge` test-group).
+const DESIRED_SLOTS: u16 = 200;
+
+/// Lowest port `port_band()` will ever place its band at - clear of the well-known/privileged
+/// range (0-1023).
+const MIN_CANDIDATE_PORT: u16 = 10000;
+
+/// Ephemeral range assumed when the kernel's own can't be read (no `/proc` - e.g. macOS),
+/// matching the Linux default. Mirrors `core/integration`'s own `port_reserver.rs`.
+const DEFAULT_EPHEMERAL_RANGE: (u16, u16) = (32768, 60999);
+
+const IP_LOCAL_PORT_RANGE: &str = "/proc/sys/net/ipv4/ip_local_port_range";
+
+fn parse_ephemeral_range(range: &str) -> Option<(u16, u16)> {
+ let mut bounds = range.split_whitespace();
+ let floor: u16 = bounds.next()?.parse().ok()?;
+ let ceiling: u16 = bounds.next()?.parse().ok()?;
+ Some((floor, ceiling))
+}
+
+/// The range the kernel picks `bind(0)` ports from, read once.
+fn ephemeral_range() -> (u16, u16) {
+ static RANGE: OnceLock<(u16, u16)> = OnceLock::new();
+ *RANGE.get_or_init(|| {
+ std::fs::read_to_string(IP_LOCAL_PORT_RANGE)
+ .ok()
+ .as_deref()
+ .and_then(parse_ephemeral_range)
+ .unwrap_or(DEFAULT_EPHEMERAL_RANGE)
+ })
+}
+
+/// First port and slot count of a band clear of the kernel's ephemeral range: below the floor by
+/// preference (matches `core/integration`'s own `port_reserver.rs`), above the ceiling when the
+/// floor leaves no room there.
+///
+/// A *hardcoded* band, chosen once without reading the real range (this file's own prior version:
+/// `15000..15199`, on the unverified assumption that every real deployment's floor sits above
+/// it), is wrong on any box tuned wider - `net.ipv4.ip_local_port_range = "1024 65535"` swallows
+/// that whole band. `flock` alone doesn't save it either: it is advisory, so it only excludes
+/// another `PortGuard`-based process, never an unrelated `bind(0)` elsewhere on the box landing on
+/// the same number from the kernel's own ephemeral pool.
+///
+fn port_band() -> (u16, u16) {
+ static BAND: OnceLock<(u16, u16)> = OnceLock::new();
+ *BAND.get_or_init(|| {
+ let (floor, ceiling) = ephemeral_range();
+ band_for(floor, ceiling)
+ })
+}
+
+/// Pure band-selection logic, split out from [`port_band`] so it's testable against synthetic
+/// ranges without needing to fake `/proc` contents.
+fn band_for(floor: u16, ceiling: u16) -> (u16, u16) {
+ if floor > MIN_CANDIDATE_PORT {
+ let room = floor - MIN_CANDIDATE_PORT;
+ return (MIN_CANDIDATE_PORT, room.min(DESIRED_SLOTS));
+ }
+ // Below this floor both allocators fall back to `ceiling + 1`: this one takes the first
+ // `DESIRED_SLOTS` of it, `core/integration`'s `port_reserver.rs` takes the rest up to
+ // `u16::MAX`, and the two lock different files (`{port}.lock` against `slot-{slot}.lock`,
+ // in different directories), so neither sees the other's reservation. Refusing is the
+ // honest answer: a port handed out here would collide silently, and the run would fail as
+ // if the code under test were broken.
+ panic!(
+ "kernel ephemeral range [{floor}, {ceiling}] leaves no room for a test port band below \
+ {MIN_CANDIDATE_PORT}, and the band above {ceiling} is shared with core/integration's \
+ own allocator under a different lock - narrow the range, e.g. \
+ sysctl -w net.ipv4.ip_local_port_range='32768 60999'"
+ );
+}
+
+#[test]
+fn given_a_readable_range_when_parsed_should_take_both_bounds() {
+ assert_eq!(
+ parse_ephemeral_range("32768\t60999\n"),
+ Some((32768, 60999))
+ );
+ assert_eq!(parse_ephemeral_range("1024 65535"), Some((1024, 65535)));
+ assert_eq!(parse_ephemeral_range(""), None);
+ assert_eq!(parse_ephemeral_range("32768"), None);
+ assert_eq!(parse_ephemeral_range("garbage 60999"), None);
+}
+
+#[test]
+fn given_room_below_the_floor_when_choosing_a_band_should_take_it() {
+ let (start, slots) = band_for(32768, 60999);
+ assert_eq!(start, MIN_CANDIDATE_PORT);
+ assert!(
+ start + slots <= 32768,
+ "band [{start}, {}] runs into the ephemeral floor",
+ start + slots - 1
+ );
+}
+
+/// A box tuned `net.ipv4.ip_local_port_range = "1024 60999"` leaves room above the ceiling, but
+/// `core/integration`'s allocator falls back to the same anchor under a lock this one cannot
+/// see, so the only sound answer is to refuse.
+#[test]
+#[should_panic(expected = "shared with core/integration")]
+fn given_no_room_below_the_floor_when_choosing_a_band_should_refuse() {
+ band_for(1024, 60999);
+}
+
+/// Boundary: a ceiling at the `u16` maximum leaves nothing above it either, and must refuse
+/// rather than overflow on the way to saying so.
+#[test]
+#[should_panic(expected = "leaves no room")]
+fn given_a_ceiling_at_the_u16_maximum_when_choosing_a_band_should_refuse() {
+ band_for(1024, u16::MAX);
+}
+
+/// Exclusive claim on one port, released (and the port freed for reuse) when dropped - including
+/// on an unclean process exit, since the OS drops the `flock` with the file descriptor. Unlike
+/// bind-then-drop, this process itself never binds the port before `iggy-server` does; see
+/// `port_band`'s doc comment for the limits of that guarantee against *other* processes.
+pub struct PortGuard {
+ pub port: u16,
+ _lock: File,
+}
+
+impl PortGuard {
+ pub fn acquire() -> Self {
+ let (band_start, band_slots) = port_band();
+ let lock_dir = std::env::temp_dir().join("iggy-kafka-gateway-test-port-locks");
+ std::fs::create_dir_all(&lock_dir).expect("create port lock dir");
+ for offset in 0..band_slots {
+ let port = band_start + offset;
+ let path = lock_dir.join(format!("{port}.lock"));
+ let Ok(file) = OpenOptions::new()
+ .read(true)
+ .write(true)
+ .create(true)
+ .truncate(false)
+ .open(&path)
+ else {
+ continue;
+ };
+ if file.try_lock().is_ok() {
+ return Self { port, _lock: file };
+ }
+ }
+ panic!(
+ "no free port slot in [{band_start}, {}]",
+ band_start + band_slots - 1
+ );
+ }
+}
+
+/// Locates the already-built `iggy-server` binary. Does not build it - see this crate's
+/// `docs/TEST_SUITE.md` for the prerequisite, the same one `core/integration`'s own
+/// server-spawning tests already carry.
+///
+/// Walks up from this test binary's own path (`env::current_exe()`) rather than calling
+/// `assert_cmd::cargo::cargo_bin("iggy-server")` directly: `cargo_bin`'s `CARGO_BIN_EXE_*`
+/// env-var lookup only resolves for binaries owned by *this* package, so for `iggy-server`
+/// (a different package) it always falls through to `cargo_bin`'s own `legacy_cargo_bin`, which
+/// panics with its own internal message - not the "binary not found" text `TEST_SUITE.md`
+/// promises - when the binary is not yet built (`CARGO_BIN_EXE_iggy-server` is unset, so its
+/// panic path prints that plus every binary name it *did* find, not a helpful next step). Ancestor
+/// directories of `current_exe()` reach the same `target/<profile>/` directory
+/// `legacy_cargo_bin` derives - correct under `CARGO_TARGET_DIR`, `--release`, or a `--target`
+/// triple subdirectory for the identical reason: it reads back from where cargo actually placed
+/// *this* binary, not a guessed path - so checking existence there first, before ever calling
+/// into `assert_cmd`, gives this crate's own message instead of a Cargo internal one. An earlier
+/// version of this function drove `cargo build --package server --bin iggy-server` directly from
+/// every server-spawning test (nextest runs each as its own process, so the `OnceLock` memoized
+/// nothing across them), serialized by `.config/nextest.toml`'s `kafka_bridge` test-group but
+/// still real, avoidable per-test overhead this version has none of.
+///
+/// # Panics
+///
+/// Panics if `iggy-server` cannot be found alongside this test binary's own target directory.
+fn iggy_server_binary() -> PathBuf {
+ let current_exe = std::env::current_exe().expect("resolve this test binary's own path");
+ let binary_name = format!("iggy-server{}", std::env::consts::EXE_SUFFIX);
+ current_exe
+ .ancestors()
+ .map(|dir| dir.join(&binary_name))
+ .find(|candidate| candidate.is_file())
+ .unwrap_or_else(|| {
+ panic!(
+ "iggy-server binary not found near {} - build it first with \
+ `cargo build --package server --bin iggy-server`",
+ current_exe.display()
+ )
+ })
+}
+
+pub struct TestServer {
+ child: Child,
+ address: String,
+ password: String,
+ _port_guard: PortGuard,
+}
+
+impl TestServer {
+ /// Spawns `iggy-server` with the default `iggy`/`iggy` root credentials. See
+ /// [`Self::spawn_with_password`] for the general form.
+ pub async fn spawn(data_dir: &std::path::Path) -> Self {
+ Self::spawn_with_password(data_dir, "iggy").await
+ }
+
+ /// Spawns `iggy-server` with an isolated temp data dir, a locked TCP port, and the given root
+ /// password, then blocks until its listener is ready or the startup budget is exhausted.
+ ///
+ /// A dedicated password parameter (not just the `spawn()` default everywhere) lets a
+ /// password-shaped regression test (special characters, say) reuse this harness's
+ /// `PortGuard`/graceful-`Drop`/`wait_ready` machinery instead of hand-rolling a second,
+ /// `Drop`-less spawn: a `Drop`-less copy has no guard to run `graceful_kill` on a panic before
+ /// its assertions, orphaning a process that still holds its `PortGuard` slot, which then
+ /// fails the next test that draws that slot to bind.
+ pub async fn spawn_with_password(data_dir: &std::path::Path, password: &str) -> Self {
+ let port_guard = PortGuard::acquire();
+ let address = format!("127.0.0.1:{}", port_guard.port);
+
+ let mut command = Command::new(iggy_server_binary());
+ command
+ .env("IGGY_PATH", data_dir.display().to_string())
+ .env("IGGY_TCP_ADDRESS", &address)
+ .env("IGGY_HTTP_ENABLED", "false")
+ .env("IGGY_QUIC_ENABLED", "false")
+ // WebSocket defaults to enabled on a fixed 127.0.0.1:8092 (config.toml), unlike TCP
+ // which reads a per-test port from PortGuard - every spawned server here would fight
+ // over that one port otherwise, and a bind failure aborts boot.
+ .env("IGGY_WEBSOCKET_ENABLED", "false")
+ // Matches core/integration's own spawned-server harness (harness/handle/server.rs):
+ // pinned shards (config.toml default: pin_cores = true) of concurrently running
+ // servers pile onto the same cores and starve each other. This crate's own tests are
+ // serialized against each other (.config/nextest.toml's kafka_bridge test-group -
+ // #[serial] alone does not survive nextest, which runs each test as its own process),
+ // but a spawned server here can still land alongside a pinned server from a different
+ // package's test in the same nextest run.
+ .env("IGGY_SHARDING_PIN_CORES", "false")
+ // Only half of core/integration's own pairing: unpinning cores stops a server from
+ // being pinned to a specific range, but config.toml's own default
+ // (cpu_allocation = "numa:auto") still sizes the shard pool to the whole machine with
+ // nothing set here - every server this crate's tests spawn would still compete for
+ // the entire box against unrelated packages' tests in the same nextest run. A small
+ // fixed range (matching the same harness's own fallback for an unreadable core count)
+ // caps that instead.
+ .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
+ // `--with-default-root-credentials` is off by default (args.rs) - without these,
+ // a fresh server provisions no loginable root user at all, and every bridge connect
+ // attempt fails with "invalid credentials" no matter what this test passes.
+ .env("IGGY_ROOT_USERNAME", "iggy")
+ .env("IGGY_ROOT_PASSWORD", password);
+ let child = command.spawn().expect("spawn iggy-server");
+
+ let mut server = Self {
+ child,
+ address,
+ password: password.to_string(),
+ _port_guard: port_guard,
+ };
+ server.wait_ready().await;
+ server
+ }
+
+ /// Polls with a bare TCP connect, not a full `IggyBridge::connect`: the latter carries
+ /// `RECONNECTION_RETRIES` (3 dials at ~1s apart) on every failed attempt, so a poll loop built
+ /// on it pays several real seconds per iteration instead of running at its own 100ms cadence,
+ /// and a *successful* poll iteration would authenticate a client and then drop it without
+ /// `close()`, leaking a session server-side. A TCP accept slightly ahead of the app being
+ /// ready to authenticate is fine - every caller's own subsequent `IggyBridge::connect` already
+ /// retries a few times, which covers the last few hundred ms of that gap.
+ ///
+ /// Checks `try_wait()` every iteration: a server that fails at boot (bad config, a port
+ /// stolen between `PortGuard::acquire` and its own bind) exits almost immediately, and without
+ /// this check that reads as "still starting" for the full 30s budget - the eventual failure
+ /// names the wrong cause ("did not become ready") instead of the real one (exited early, with
+ /// whatever it printed before dying).
+ async fn wait_ready(&mut self) {
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+ loop {
+ if let Some(status) = self.child.try_wait().expect("poll child status") {
+ panic!(
+ "iggy-server at {} exited during startup with {status}",
+ self.address
+ );
+ }
+ if tokio::net::TcpStream::connect(self.address.as_str())
+ .await
+ .is_ok()
+ {
+ return;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "iggy-server at {} did not become ready within the startup budget",
+ self.address
+ );
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+ }
+
+ pub fn test_config(&self) -> IggyBridgeConfig {
+ IggyBridgeConfig {
+ address: self.address.clone(),
+ username: "iggy".to_string(),
+ password: SecretString::from(self.password.clone()),
+ topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new())
+ .expect("valid mapping for this test's fixture data"),
+ }
+ }
+}
+
+/// SIGTERM, wait up to `SIGTERM_TIMEOUT`, then SIGKILL if it hasn't exited. Mirrors
+/// `core/integration`'s own `harness::handle::common::graceful_kill` (not reused directly - that
+/// crate is heavyweight, and pulling it in for one function isn't worth it; this crate already
+/// depends directly on `iggy`/`core/sdk`, so `core/integration` wouldn't add a *new*
+/// core/sdk-change-reruns-these-tests edge, just an unrelated dependency). A bare SIGKILL skips
+/// `iggy-server`'s shutdown path entirely, which is a materially different exit than what the
+/// binary is actually built to do on `SIGTERM` - `.kill()` alone bypassed that in every test run
+/// before this fix.
+const SIGTERM_TIMEOUT: Duration = Duration::from_secs(5);
+const SIGKILL_POLL_INTERVAL: Duration = Duration::from_millis(50);
+
+fn graceful_kill(child: &mut Child) {
+ // `wait_ready` calls `try_wait()` too (and panics on an early exit, unwinding into this via
+ // `Drop`) - if it already reaped the child, `child.id()` is a PID the OS is free to hand to an
+ // unrelated process by the time we get here, and a raw `libc::kill` (unlike `std::Child::kill`,
+ // which checks its own cached exit status first and no-ops instead) has no such guard against
+ // signaling that PID anyway. Checking here first closes the same gap `std` already closes for
+ // its own `kill`.
+ if matches!(child.try_wait(), Ok(Some(_))) {
+ return;
+ }
+
+ let pid = child.id() as libc::pid_t;
+ // Safety: `pid` is this process's own live child, confirmed by the `try_wait` check above;
+ // sending it a signal is exactly what `Child::kill` itself does internally.
+ unsafe {
+ libc::kill(pid, libc::SIGTERM);
+ }
+
+ let deadline = std::time::Instant::now() + SIGTERM_TIMEOUT;
+ while std::time::Instant::now() < deadline {
+ match child.try_wait() {
+ Ok(None) => std::thread::sleep(SIGKILL_POLL_INTERVAL),
+ Ok(Some(_)) | Err(_) => return,
+ }
+ }
+
+ let _ = child.kill();
+}
+
+impl Drop for TestServer {
+ fn drop(&mut self) {
+ graceful_kill(&mut self.child);
+ let _ = self.child.wait();
+ }
+}
+
+/// Builds and connects a raw `IggyClient` against `server` - for producing test data directly,
+/// independent of the `IggyBridge` under test. Uses the fluent builder, not a hand-built
+/// `iggy://user:pass@host` string: `ConnectionString` splits on `@` then `:`, which breaks for
+/// any password containing either character.
+pub async fn raw_client(server: &TestServer) -> IggyClient {
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address(server.address.clone())
+ .with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword(
+ "iggy".to_string(),
+ SecretString::from("iggy"),
+ )))
+ .build()
+ .expect("build raw test client");
+ client.connect().await.expect("connect raw test client");
+ client
+}
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index a992c14e9..ebbd9568a 100644
--- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs
+++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
@@ -29,8 +29,8 @@ use iggy_gateway_kafka::protocol::api::{
use codec::Encoder;
use wire::build_api_versions_flexible_request;
-#[test]
-fn golden_apiversions_v3_flexible_response_fixture() {
+#[tokio::test]
+async fn golden_apiversions_v3_flexible_response_fixture() {
// Field-by-field structural tests (`version_firewall_tests.rs`'s `apiversions_advertises_
// exact_supported_ranges_v3_flexible`) check values against `SCOPED_API_KEYS`, but only this
// exact-byte pin catches a wrong tagged-fields byte, wrong varint encoding, or misordered
@@ -38,6 +38,7 @@ fn golden_apiversions_v3_flexible_response_fixture() {
let broker = BrokerAdvertise::default();
let request = build_api_versions_flexible_request("iggy-test", "0.1.0");
let actual = handle_request(API_KEY_API_VERSIONS, 3, request, &broker)
+ .await
.expect_response("test request has acks != 0 and expects a response");
// error_code=0, api_count=6 (compact array: N+1=7)
@@ -63,10 +64,11 @@ fn golden_apiversions_v3_flexible_response_fixture() {
assert_eq!(actual.as_ref(), &expected);
}
-#[test]
-fn golden_apiversions_v1_response_fixture() {
+#[tokio::test]
+async fn golden_apiversions_v1_response_fixture() {
let broker = BrokerAdvertise::default();
let actual = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &broker)
+ .await
.expect_response("test request has acks != 0 and expects a response");
// error_code=0, api_count=6
@@ -91,8 +93,8 @@ fn golden_apiversions_v1_response_fixture() {
assert_eq!(actual.as_ref(), &expected);
}
-#[test]
-fn golden_metadata_v0_single_topic_response_fixture() {
+#[tokio::test]
+async fn golden_metadata_v0_single_topic_response_fixture() {
let mut request = Encoder::with_capacity(32);
request.write_i32(1); // one topic
request
@@ -101,6 +103,7 @@ fn golden_metadata_v0_single_topic_response_fixture() {
let req_bytes = request.freeze();
let actual = handle_request(API_KEY_METADATA, 0, req_bytes, &BrokerAdvertise::default())
+ .await
.expect_response("test request has acks != 0 and expects a response");
// Metadata v0 layout: brokers[], topics[] (no controller_id - added in v1)
diff --git a/gateways/kafka/tests/listener_robustness_tests.rs b/gateways/kafka/tests/listener_robustness_tests.rs
index 9c8995b1a..a9c67f599 100644
--- a/gateways/kafka/tests/listener_robustness_tests.rs
+++ b/gateways/kafka/tests/listener_robustness_tests.rs
@@ -231,6 +231,7 @@ async fn e2e_response_frames_have_positive_big_endian_length_prefix() {
request.clone(),
&BrokerAdvertise::default(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response")
.len();
// The wire frame also carries the response header `handle_request`'s return value doesn't
diff --git a/gateways/kafka/tests/response_negative_tests.rs b/gateways/kafka/tests/response_negative_tests.rs
index 58bc110ea..db05f69d4 100644
--- a/gateways/kafka/tests/response_negative_tests.rs
+++ b/gateways/kafka/tests/response_negative_tests.rs
@@ -31,11 +31,7 @@ use iggy_gateway_kafka::protocol::api::{
ERROR_INVALID_PARTITIONS, ERROR_INVALID_REPLICATION_FACTOR, ERROR_INVALID_REQUEST,
ERROR_NOT_CONTROLLER, ERROR_UNSUPPORTED_VERSION,
};
-use iggy_gateway_kafka::protocol::responses::{
- encode_create_topics_error_response, encode_create_topics_response,
- encode_fetch_error_response, encode_fetch_response, encode_list_offsets_error_response,
- encode_produce_error_response,
-};
+use iggy_gateway_kafka::protocol::handlers::{create_topics, fetch, list_offsets, produce};
use codec::Decoder;
@@ -53,7 +49,7 @@ fn creatable_topic(name: &str, num_partitions: i32, replication_factor: i16) ->
#[test]
fn create_topics_response_flags_non_positive_partition_count_v2() {
let req = CreateTopicsRequest::default().with_topics(vec![creatable_topic("bad-topic", 0, 1)]);
- let body = encode_create_topics_response(2, &req).unwrap();
+ let body = create_topics::encode_response(2, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_i32().unwrap(), 1); // topics len
@@ -70,7 +66,7 @@ fn create_topics_v5_broker_default_partitions_is_not_invalid_partitions() {
let req = CreateTopicsRequest::default()
.with_topics(vec![creatable_topic("default-parts", -1, 2)])
.with_validate_only(true);
- let body = encode_create_topics_response(5, &req).unwrap();
+ let body = create_topics::encode_response(5, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_varint().unwrap(), 2); // one topic
@@ -92,7 +88,7 @@ fn create_topics_v5_flags_zero_and_below_minus_one_partition_count() {
num_partitions,
1,
)]);
- let body = encode_create_topics_response(5, &req).unwrap();
+ let body = create_topics::encode_response(5, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0);
assert_eq!(d.read_varint().unwrap(), 2);
@@ -116,7 +112,7 @@ fn create_topics_v5_flags_invalid_replication_factor() {
1,
replication_factor,
)]);
- let body = encode_create_topics_response(5, &req).unwrap();
+ let body = create_topics::encode_response(5, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0);
assert_eq!(d.read_varint().unwrap(), 2);
@@ -136,7 +132,7 @@ fn create_topics_v5_flags_invalid_replication_factor() {
fn create_topics_v2_rejects_broker_default_sentinel() {
// KIP-464 defaults apply from v4; on v2 without assignments, -1 is INVALID_PARTITIONS.
let req = CreateTopicsRequest::default().with_topics(vec![creatable_topic("legacy", -1, 1)]);
- let body = encode_create_topics_response(2, &req).unwrap();
+ let body = create_topics::encode_response(2, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0);
assert_eq!(d.read_i32().unwrap(), 1);
@@ -158,7 +154,7 @@ fn create_topics_v2_with_assignments_allows_broker_default_sentinels() {
let req = CreateTopicsRequest::default().with_topics(vec![
creatable_topic("assigned", -1, -1).with_assignments(vec![assignment]),
]);
- let body = encode_create_topics_response(2, &req).unwrap();
+ let body = create_topics::encode_response(2, &req).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0);
assert_eq!(d.read_i32().unwrap(), 1);
@@ -171,7 +167,7 @@ fn create_topics_v2_with_assignments_allows_broker_default_sentinels() {
#[test]
fn create_topics_error_response_carries_explicit_error_code() {
- let body = encode_create_topics_error_response(5, ERROR_UNSUPPORTED_VERSION).unwrap();
+ let body = create_topics::encode_error_response(5, ERROR_UNSUPPORTED_VERSION).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0);
assert_eq!(d.read_varint().unwrap(), 2);
@@ -184,7 +180,7 @@ fn create_topics_error_response_carries_explicit_error_code() {
#[test]
fn fetch_error_response_v7_uses_top_level_error_and_no_topics() {
- let body = encode_fetch_error_response(7, ERROR_INVALID_REQUEST).unwrap();
+ let body = fetch::encode_error_response(7, ERROR_INVALID_REQUEST).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
@@ -195,7 +191,7 @@ fn fetch_error_response_v7_uses_top_level_error_and_no_topics() {
#[test]
fn fetch_error_response_v12_uses_flexible_empty_topics() {
- let body = encode_fetch_error_response(12, ERROR_UNSUPPORTED_VERSION).unwrap();
+ let body = fetch::encode_error_response(12, ERROR_UNSUPPORTED_VERSION).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION);
@@ -208,9 +204,9 @@ fn fetch_error_response_v12_uses_flexible_empty_topics() {
#[test]
fn list_offsets_error_response_v6_uses_flexible_shape() {
// v0's legacy `old_style_offsets` shape has no `kafka_protocol` encoder - see
- // `responses::encode_list_offsets_error_response` and the version-firewall e2e coverage for
+ // `list_offsets::encode_error_response` and the version-firewall e2e coverage for
// that behavior. This exercises the lowest version the crate can actually encode.
- let body = encode_list_offsets_error_response(6, ERROR_UNSUPPORTED_VERSION).unwrap();
+ let body = list_offsets::encode_error_response(6, ERROR_UNSUPPORTED_VERSION).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_i32().unwrap(), 0); // throttle
assert_eq!(d.read_varint().unwrap(), 2); // one topic
@@ -229,7 +225,7 @@ fn list_offsets_error_response_v6_uses_flexible_shape() {
#[test]
fn produce_error_response_v9_uses_flexible_record_errors_shape() {
- let body = encode_produce_error_response(9, ERROR_INVALID_REQUEST).unwrap();
+ let body = produce::encode_error_response(9, ERROR_INVALID_REQUEST).unwrap();
let mut d = Decoder::new(body);
assert_eq!(d.read_varint().unwrap(), 2); // one topic
assert_eq!(
@@ -252,25 +248,21 @@ fn success_responses_can_still_encode_empty_request_vectors() {
.with_acks(1)
.with_timeout_ms(1_000)
.with_topic_data(Vec::<TopicProduceData>::new());
- assert!(
- !iggy_gateway_kafka::protocol::responses::encode_produce_response(3, &produce)
- .unwrap()
- .is_empty()
- );
+ assert!(!produce::encode_response(3, &produce).unwrap().is_empty());
let fetch = FetchRequest::default().with_topics(Vec::<FetchTopic>::new());
- assert!(!encode_fetch_response(4, &fetch).unwrap().is_empty());
+ assert!(!fetch::encode_response(4, &fetch).unwrap().is_empty());
let list_offsets = ListOffsetsRequest::default().with_topics(Vec::<ListOffsetsTopic>::new());
assert!(
- !iggy_gateway_kafka::protocol::responses::encode_list_offsets_response(1, &list_offsets)
+ !list_offsets::encode_response(1, &list_offsets)
.unwrap()
.is_empty()
);
let create_topics = CreateTopicsRequest::default().with_topics(Vec::<CreatableTopic>::new());
assert!(
- !encode_create_topics_response(2, &create_topics)
+ !create_topics::encode_response(2, &create_topics)
.unwrap()
.is_empty()
);
diff --git a/gateways/kafka/tests/server_e2e_tests.rs b/gateways/kafka/tests/server_e2e_tests.rs
index 84ddde5d0..b09a04479 100644
--- a/gateways/kafka/tests/server_e2e_tests.rs
+++ b/gateways/kafka/tests/server_e2e_tests.rs
@@ -234,7 +234,7 @@ async fn e2e_produce_v3_acks_one_still_returns_response() {
#[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`),
+ // (it predates the schema the crate generates from - see `list_offsets::encode_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;
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index 4c8b01323..c0ea9f315 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -100,9 +100,10 @@ fn is_supported_version_matches_scope_table() {
/// (Produce, Fetch, `ListOffsets`, Metadata, `ApiVersions`, `CreateTopics`) -
/// `supported_ranges_table_has_six_entries` plus `is_supported_version_matches_scope_table`
/// already pin that both tables cover the same six keys.
-#[test]
-fn apiversions_advertises_exact_supported_ranges_v1() {
+#[tokio::test]
+async fn apiversions_advertises_exact_supported_ranges_v1() {
let body = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
assert_eq!(d.read_i16().unwrap(), 0);
@@ -128,14 +129,15 @@ fn apiversions_advertises_exact_supported_ranges_v1() {
/// See `apiversions_advertises_exact_supported_ranges_v1` for why this checks against
/// `SCOPED_API_KEYS`, not `supported_api_ranges()`.
-#[test]
-fn apiversions_advertises_exact_supported_ranges_v3_flexible() {
+#[tokio::test]
+async 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(),
)
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(body);
assert_eq!(d.read_i16().unwrap(), 0);
@@ -166,8 +168,8 @@ fn apiversions_advertises_exact_supported_ranges_v3_flexible() {
// `SCOPED_API_KEYS` (independent of `supported_api_ranges()`, the function under test) plus the
// acks=0/acks!=0 behavioral split this shorter version never covered.
-#[test]
-fn apiversions_all_versions_return_success() {
+#[tokio::test]
+async 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")
@@ -175,15 +177,17 @@ fn apiversions_all_versions_return_success() {
Bytes::new()
};
let body = handle_request(API_KEY_API_VERSIONS, version, request, &default_broker())
+ .await
.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() {
+#[tokio::test]
+async fn apiversions_out_of_range_returns_unsupported_in_body() {
let body = handle_request(API_KEY_API_VERSIONS, 99, Bytes::new(), &default_broker())
+ .await
.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);
@@ -193,8 +197,8 @@ fn metadata_request_one_topic() -> Bytes {
build_metadata_legacy_request(&["test-topic"])
}
-#[test]
-fn metadata_below_min_version_closes_connection() {
+#[tokio::test]
+async fn metadata_below_min_version_closes_connection() {
assert!(
handle_request(
API_KEY_METADATA,
@@ -202,13 +206,14 @@ fn metadata_below_min_version_closes_connection() {
metadata_request_one_topic(),
&default_broker(),
)
+ .await
.is_close(),
"Metadata below supported min must close rather than return a clamped body"
);
}
-#[test]
-fn metadata_above_max_version_closes_connection() {
+#[tokio::test]
+async fn metadata_above_max_version_closes_connection() {
// v10 request uses flexible encoding; a clamped v9 reply would not survive client parsing.
assert!(
handle_request(
@@ -217,6 +222,7 @@ fn metadata_above_max_version_closes_connection() {
build_metadata_flexible_request_v10(&["test-topic"]),
&default_broker(),
)
+ .await
.is_close(),
"Metadata above supported max must close rather than return a clamped body"
);
@@ -238,8 +244,8 @@ async fn e2e_metadata_above_max_version_closes_tcp_connection() {
);
}
-#[test]
-fn produce_below_min_version_with_nonzero_acks_closes_connection() {
+#[tokio::test]
+async 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
@@ -250,83 +256,98 @@ fn produce_below_min_version_with_nonzero_acks_closes_connection() {
2,
build_produce_v2_body(1, 0),
&default_broker(),
- );
+ )
+ .await;
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() {
+#[tokio::test]
+async 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(),
+ handle_request(API_KEY_FETCH, 3, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"Fetch v3 has no encodable response shape and must close"
);
}
-#[test]
-fn fetch_unsupported_version_above_max_closes_connection() {
+#[tokio::test]
+async 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(),
+ handle_request(API_KEY_FETCH, 13, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"Fetch above encoder max must close rather than return a clamped body"
);
}
-#[test]
-fn produce_unsupported_version_above_max_closes_connection() {
+#[tokio::test]
+async fn produce_unsupported_version_above_max_closes_connection() {
assert!(
- handle_request(API_KEY_PRODUCE, 13, Bytes::new(), &default_broker()).is_close(),
+ handle_request(API_KEY_PRODUCE, 13, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"Produce above encoder max must close rather than return a clamped body"
);
}
-#[test]
-fn create_topics_unsupported_version_above_max_closes_connection() {
+#[tokio::test]
+async fn create_topics_unsupported_version_above_max_closes_connection() {
assert!(
- handle_request(API_KEY_CREATE_TOPICS, 7, Bytes::new(), &default_broker()).is_close(),
+ handle_request(API_KEY_CREATE_TOPICS, 7, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"CreateTopics above encoder max must close rather than return a clamped body"
);
}
-#[test]
-fn list_offsets_unsupported_version_above_max_closes_connection() {
+#[tokio::test]
+async fn list_offsets_unsupported_version_above_max_closes_connection() {
assert!(
- handle_request(API_KEY_LIST_OFFSETS, 7, Bytes::new(), &default_broker()).is_close(),
+ handle_request(API_KEY_LIST_OFFSETS, 7, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"ListOffsets above encoder max must close rather than return a clamped body"
);
}
-#[test]
-fn list_offsets_v0_closes_connection() {
+#[tokio::test]
+async 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(),
+ handle_request(API_KEY_LIST_OFFSETS, 0, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"ListOffsets v0 has no encodable response shape and must close"
);
}
-#[test]
-fn create_topics_below_min_version_closes_connection() {
+#[tokio::test]
+async 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(),
+ handle_request(API_KEY_CREATE_TOPICS, 1, Bytes::new(), &default_broker())
+ .await
+ .is_close(),
"CreateTopics v1 has no encodable response shape and must close"
);
}
-#[test]
-fn unsupported_api_keys_close_connection() {
+#[tokio::test]
+async 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());
+ let outcome = handle_request(key, 0, Bytes::new(), &default_broker()).await;
assert!(
outcome.is_close(),
"unknown api_key {key} must close (no parseable response schema)"
@@ -338,8 +359,8 @@ fn unsupported_api_keys_close_connection() {
// `api_handler_tests.rs::handle_request_succeeds_for_every_supported_version_with_fixture` -
// that generic per-scoped-API loop already exercises the exact same fixtures and version ranges.
-#[test]
-fn corrupt_produce_body_with_acks_stays_silent() {
+#[tokio::test]
+async 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
@@ -351,27 +372,30 @@ fn corrupt_produce_body_with_acks_stays_silent() {
0xFF, 0xFF, 0xFF, // truncated topics count
]);
assert!(
- handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).is_no_response(),
+ handle_request(API_KEY_PRODUCE, 3, body, &default_broker())
+ .await
+ .is_no_response(),
"malformed Produce body must stay silent regardless of acks"
);
}
-#[test]
-fn corrupt_produce_body_before_acks_is_silent() {
+#[tokio::test]
+async 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());
+ let outcome = handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).await;
assert!(
outcome.is_no_response(),
"produce decode failure before acks must be silent"
);
}
-#[test]
-fn corrupt_fetch_body_returns_invalid_request_error() {
+#[tokio::test]
+async 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())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(resp);
assert_eq!(d.read_i32().unwrap(), 0);
@@ -384,11 +408,13 @@ fn corrupt_fetch_body_returns_invalid_request_error() {
// ── ListOffsets v0 (no encodable representation in kafka_protocol) ─────────
-#[test]
-fn list_offsets_v0_with_topic_closes_connection() {
+#[tokio::test]
+async 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(),
+ handle_request(API_KEY_LIST_OFFSETS, 0, request_body, &default_broker())
+ .await
+ .is_close(),
"ListOffsets v0 has no encodable response shape and must close, even with a well-formed body"
);
}
@@ -571,10 +597,10 @@ async fn apiversions_v4_out_of_range_e2e_returns_unsupported() {
// ── Out-of-scope API keys (SCOPE.md unsupported list) ───────────────────────
-#[test]
-fn out_of_scope_api_keys_close_without_panic() {
+#[tokio::test]
+async fn out_of_scope_api_keys_close_without_panic() {
for &(api_key, name) in OUT_OF_SCOPE_API_KEYS {
- let outcome = handle_request(api_key, 0, Bytes::new(), &default_broker());
+ let outcome = handle_request(api_key, 0, Bytes::new(), &default_broker()).await;
assert!(outcome.is_close(), "{name} (key {api_key}) must close");
}
}
@@ -685,8 +711,8 @@ async fn each_scoped_api_below_min_version_e2e_closes_except_api_versions() {
);
}
-#[test]
-fn produce_advertises_min_zero_but_firewall_rejects_below_v3() {
+#[tokio::test]
+async fn produce_advertises_min_zero_but_firewall_rejects_below_v3() {
let range = SCOPED_API_KEYS
.iter()
.find(|(k, _, _, _)| *k == API_KEY_PRODUCE)
@@ -706,6 +732,7 @@ fn produce_advertises_min_zero_but_firewall_rejects_below_v3() {
build_produce_v2_body(0, 0),
&default_broker(),
)
+ .await
.is_no_response(),
"Produce v2 acks=0 must stay silent, not close"
);
@@ -718,6 +745,7 @@ fn produce_advertises_min_zero_but_firewall_rejects_below_v3() {
build_produce_v2_body(1, 0),
&default_broker(),
)
+ .await
.is_close(),
"Produce v2 acks != 0 has no encodable response shape and must close"
);
@@ -770,10 +798,11 @@ async fn create_topics_v1_unsupported_e2e_closes_connection() {
/// `ERROR_INVALID_REQUEST` specifically - never `ERROR_UNSUPPORTED_VERSION`, which
/// `handle_versioned_request` only reaches on the *unsupported-version* branch, not a decode
/// failure - so this is a precise decode, not a scan for either.
-#[test]
-fn corrupt_list_offsets_body_returns_invalid_request_error() {
+#[tokio::test]
+async fn corrupt_list_offsets_body_returns_invalid_request_error() {
let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF]);
let resp = handle_request(API_KEY_LIST_OFFSETS, 1, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(resp);
// v1 has no throttle_time_ms (added v2+).
@@ -786,10 +815,11 @@ fn corrupt_list_offsets_body_returns_invalid_request_error() {
/// v2 (legacy, non-flexible) - see `corrupt_list_offsets_body_returns_invalid_request_error`
/// above for why this is a precise decode of `ERROR_INVALID_REQUEST`, not a scan for either code.
-#[test]
-fn corrupt_create_topics_body_returns_invalid_request_error() {
+#[tokio::test]
+async fn corrupt_create_topics_body_returns_invalid_request_error() {
let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF]);
let resp = handle_request(API_KEY_CREATE_TOPICS, 2, body, &default_broker())
+ .await
.expect_response("test request has acks != 0 and expects a response");
let mut d = Decoder::new(resp);
assert_eq!(d.read_i32().unwrap(), 0, "throttle");