krishvishal commented on code in PR #4043: URL: https://github.com/apache/iggy/pull/4043#discussion_r4014632095
########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,503 @@ +// 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 +/// [`CONNECT_TIMEOUT`] for the latter. +const RECONNECTION_RETRIES: u32 = 3; + +/// Wall-clock ceiling on the whole `client.connect()` call in [`IggyBridge::connect`], including +/// every attempt [`RECONNECTION_RETRIES`] makes internally. +/// +/// Without this, an unreachable-but-not-refusing address hangs far longer than "a few seconds": +/// `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, and the plain `TcpStream::connect` underneath has no +/// deadline of its own. Against a firewall that drops SYN packets instead of refusing them, each +/// of the up to `RECONNECTION_RETRIES + 1` dial attempts 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. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Ceiling on a single Iggy client call made *after* [`IggyBridge::connect`] already succeeded - +/// `get_stream`, `create_stream`, `get_topic`, `create_topic`, and `shutdown`. +/// +/// [`CONNECT_TIMEOUT`] alone only bounds the *initial* connect. `TcpClient::send_raw_with_response` +/// (`tcp_client.rs`) reconnects internally on any transport error, mid-call, through the exact +/// same undead-lined `TcpStream::connect` path `CONNECT_TIMEOUT` exists to bound - so a bridge +/// call made after a successful `connect()` (a later network partition, Iggy moved to a new +/// address) could still hang on the kernel's own SYN-retry window with nothing here to stop it. +/// Same value as `CONNECT_TIMEOUT`: both bound the same underlying dial, just entered from a +/// different call site. +const REQUEST_TIMEOUT: Duration = CONNECT_TIMEOUT; Review Comment: warning: `REQUEST_TIMEOUT` is 15s and the SDK's own reply deadline is 30s (`core/sdk/src/tcp/tcp_client.rs:65`, applied as the read deadline at `:1271`). So this fires first on every slow reply rather than backstopping the SDK. The part that bites is what it cancels. `send_raw_vsr_attempt` takes the connection's stream mutex at `tcp_client.rs:1420-1421` inside a `tokio::spawn`, and the handle is joined at `:1556`. When this timer fires the outer future drops, the `JoinHandle` drops with it, and the task detaches still holding that lock. With one `IggyClient` for the gateway that lock is the only path to Iggy, so the next bridge call queues behind an abandoned request until the SDK's own 30s runs out. Raising the constant past 30s stops the cancellation, but a reconnect replay (`tcp_client.rs:279,308`) starts a fresh 30s budget, so any fixed number can still land mid-attempt. The clean version is to not cancel a post-connect call at all and let the SDK's read deadline do the bounding. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,503 @@ +// 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 +/// [`CONNECT_TIMEOUT`] for the latter. +const RECONNECTION_RETRIES: u32 = 3; + +/// Wall-clock ceiling on the whole `client.connect()` call in [`IggyBridge::connect`], including +/// every attempt [`RECONNECTION_RETRIES`] makes internally. +/// +/// Without this, an unreachable-but-not-refusing address hangs far longer than "a few seconds": +/// `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, and the plain `TcpStream::connect` underneath has no +/// deadline of its own. Against a firewall that drops SYN packets instead of refusing them, each +/// of the up to `RECONNECTION_RETRIES + 1` dial attempts 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. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Ceiling on a single Iggy client call made *after* [`IggyBridge::connect`] already succeeded - +/// `get_stream`, `create_stream`, `get_topic`, `create_topic`, and `shutdown`. +/// +/// [`CONNECT_TIMEOUT`] alone only bounds the *initial* connect. `TcpClient::send_raw_with_response` +/// (`tcp_client.rs`) reconnects internally on any transport error, mid-call, through the exact +/// same undead-lined `TcpStream::connect` path `CONNECT_TIMEOUT` exists to bound - so a bridge +/// call made after a successful `connect()` (a later network partition, Iggy moved to a new +/// address) could still hang on the kernel's own SYN-retry window with nothing here to stop it. +/// Same value as `CONNECT_TIMEOUT`: both bound the same underlying dial, just entered from a +/// different call site. +const REQUEST_TIMEOUT: Duration = CONNECT_TIMEOUT; + +/// 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`]. +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::Iggy(IggyError::CannotEstablishConnection))? Review Comment: warning: an elapsed deadline becomes `CannotEstablishConnection`, which `:146,148` routes to `ERROR_NOT_LEADER_OR_FOLLOWER` (6). Two things wrong with that. The outcome is not a known non-delivery. The SDK writes, flushes and reads inside a detached `tokio::spawn` (`core/sdk/src/tcp/tcp_client.rs:1420`), so by the time this fires the bytes are most likely already on the socket. The doc at `:115-118` in this same file is the rule that an unknown-outcome failure must not take a code from the known-safe-to-retry set, which is why `TransientNotCommitted` goes to `ERROR_REQUEST_TIMED_OUT` at `:149`. The timeout path is the same shape and takes the code that rule excludes. Code 6 is also an `InvalidMetadataException`, so a Kafka client refreshes metadata against a single-broker gateway and learns nothing from the trip. A `BridgeError::Timeout` variant plus one arm in `to_kafka_error_code` covers it. No `IggyError` change and no wire numbering involved. Worth settling before #3535, since Produce will go through this same helper. ########## gateways/kafka/src/bridge/config.rs: ########## @@ -0,0 +1,345 @@ +// 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::path::Path; + +use secrecy::SecretString; +use tracing::warn; + +use crate::bridge::error::BridgeError; +use crate::bridge::topic_map::{TopicMapping, validate_identifier_name}; + +const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090"; +/// Matches `DEFAULT_ROOT_USERNAME` (`core/server/src/boot/credentials.rs`) - the root user's +/// username is always `iggy` regardless of how its password was provisioned, so defaulting this +/// one field is safe. The password is a different story - see why there is no +/// `DEFAULT_IGGY_PASSWORD` at [`IggyBridgeConfig::from_env`]. +const DEFAULT_IGGY_USERNAME: &str = "iggy"; + +/// Connection + topic-mapping config for [`IggyBridge`](crate::bridge::iggy_bridge::IggyBridge). +/// +/// `Debug` is safe to derive: `password` is `SecretString`, which redacts on `Debug` by design +/// (`secrecy` crate) - never add a plain `String` credential field here without the same +/// treatment (see `connector-pr-review` blocker B1 in the connectors subsystem for why). +#[derive(Debug, Clone)] +pub struct IggyBridgeConfig { + pub address: String, + pub username: String, + pub password: SecretString, + pub topic_mapping: TopicMapping, +} + +impl IggyBridgeConfig { + /// The complete set of `IGGY_KAFKA_*` vars this module reads. `main.rs`'s + /// `reject_unknown_kafka_env_vars` checks this list *and* its own separate + /// `KNOWN_KAFKA_ENV_VARS`, not one merged copy - a var added only here is already + /// recognized there with no corresponding edit needed, and vice versa. A var this module + /// reads still has to be listed *somewhere* the guard checks, or a typo silently no-ops + /// instead of surfacing (`IGGY_KAFKA_` is a `DELEGATED_ENV_VAR_PREFIXES` entry in + /// `core/configs`, so the central provider's own typo-detection doesn't cover this namespace + /// either). + pub const KNOWN_ENV_VARS: &'static [&'static str] = &[ + "IGGY_KAFKA_IGGY_ADDR", + "IGGY_KAFKA_IGGY_USERNAME", + "IGGY_KAFKA_IGGY_PASSWORD", + "IGGY_KAFKA_IGGY_STREAM", + "IGGY_KAFKA_TOPIC_MAP_PATH", + ]; + + /// Builds config from `IGGY_KAFKA_*` env vars, defaulting to the Iggy server's own + /// out-of-the-box address and root username. + /// + /// `IGGY_KAFKA_IGGY_PASSWORD` has no default and must be set explicitly. `iggy-server` only + /// provisions the well-known `iggy`/`iggy` root credentials when started with + /// `--with-default-root-credentials` (`args.rs`, itself documented "INSECURE - FOR DEVELOPMENT + /// ONLY!"); without that flag it generates a random password (`boot/credentials.rs`) that no + /// constant here could ever guess. Defaulting the password to `"iggy"` would make this bridge + /// silently authenticate as root, with the credential in no config file and no log, on any + /// server that does happen to run dev-flagged - the failure mode of getting it wrong is a + /// loud, immediate auth rejection instead. + /// + /// `IGGY_KAFKA_IGGY_STREAM` and `IGGY_KAFKA_TOPIC_MAP_PATH` both influence the mapping's + /// default stream; when both are set, the TOML file's own `default_stream` wins and + /// `IGGY_KAFKA_IGGY_STREAM` is ignored entirely for the topics it covers (a TOML file is a + /// complete mapping document, not an overlay) - a `warn!` fires so that isn't silently + /// mysterious to whoever set the env var expecting it to matter. + /// + /// # Errors + /// + /// Returns [`BridgeError::InvalidConfig`] if `IGGY_KAFKA_IGGY_PASSWORD` is unset or empty, if + /// `IGGY_KAFKA_IGGY_USERNAME` is set but empty, or if `IGGY_KAFKA_TOPIC_MAP_PATH` is set but + /// the file is missing or fails to parse. + pub fn from_env() -> Result<Self, BridgeError> { + let address = + std::env::var("IGGY_KAFKA_IGGY_ADDR").unwrap_or_else(|_| DEFAULT_IGGY_ADDR.to_string()); + let username = std::env::var("IGGY_KAFKA_IGGY_USERNAME") + .unwrap_or_else(|_| DEFAULT_IGGY_USERNAME.to_string()); + // set-but-empty (`VAR=""`) is `Ok("")` from `env::var`, not `Err` - distinct from unset, + // and unchecked here would otherwise pass both cleanly through to a login attempt neither + // could ever satisfy. + if username.is_empty() { + return Err(BridgeError::InvalidConfig( + "IGGY_KAFKA_IGGY_USERNAME must not be empty".to_string(), + )); + } + let password = std::env::var("IGGY_KAFKA_IGGY_PASSWORD").map_err(|_| { + BridgeError::InvalidConfig( + "IGGY_KAFKA_IGGY_PASSWORD must be set - iggy-server generates a random root \ + password unless started with --with-default-root-credentials, so there is no \ + safe default to fall back to" + .to_string(), + ) + })?; + if password.is_empty() { + return Err(BridgeError::InvalidConfig( + "IGGY_KAFKA_IGGY_PASSWORD must not be empty".to_string(), + )); + } + let stream_env = std::env::var("IGGY_KAFKA_IGGY_STREAM").ok(); + // TopicMapping::new (below, on the no-file branch) validates default_stream too now, so + // removing this check wouldn't let an invalid value through uncaught - but its own message + // would say "topic mapping's default_stream", not IGGY_KAFKA_IGGY_STREAM, leaving whoever + // reads the error to work out which env var that phrase actually refers to. Checked here + // first so the message names the var an operator can actually go fix. + if let Some(ref stream) = stream_env { Review Comment: warning: this runs before the map-file branch at `:123`, so an empty or whitespace-padded `IGGY_KAFKA_IGGY_STREAM` hard-fails startup even when `IGGY_KAFKA_TOPIC_MAP_PATH` is set. `README.md:118-120` and the `warn!` at `:126` both tell the operator the var is ignored entirely in that case, so the failure contradicts the message it ships with. Moving the check into the `None` arm fixes it and removes the duplicate, since `TopicMapping::new` at `topic_map.rs:178` already validates `default_stream` on that branch. That also retires the comment above, which exists only to explain the overlap. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,503 @@ +// 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 +/// [`CONNECT_TIMEOUT`] for the latter. +const RECONNECTION_RETRIES: u32 = 3; + +/// Wall-clock ceiling on the whole `client.connect()` call in [`IggyBridge::connect`], including +/// every attempt [`RECONNECTION_RETRIES`] makes internally. +/// +/// Without this, an unreachable-but-not-refusing address hangs far longer than "a few seconds": +/// `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, and the plain `TcpStream::connect` underneath has no +/// deadline of its own. Against a firewall that drops SYN packets instead of refusing them, each +/// of the up to `RECONNECTION_RETRIES + 1` dial attempts 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. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Ceiling on a single Iggy client call made *after* [`IggyBridge::connect`] already succeeded - +/// `get_stream`, `create_stream`, `get_topic`, `create_topic`, and `shutdown`. +/// +/// [`CONNECT_TIMEOUT`] alone only bounds the *initial* connect. `TcpClient::send_raw_with_response` +/// (`tcp_client.rs`) reconnects internally on any transport error, mid-call, through the exact +/// same undead-lined `TcpStream::connect` path `CONNECT_TIMEOUT` exists to bound - so a bridge +/// call made after a successful `connect()` (a later network partition, Iggy moved to a new +/// address) could still hang on the kernel's own SYN-retry window with nothing here to stop it. +/// Same value as `CONNECT_TIMEOUT`: both bound the same underlying dial, just entered from a +/// different call site. +const REQUEST_TIMEOUT: Duration = CONNECT_TIMEOUT; + +/// 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`]. +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::Iggy(IggyError::CannotEstablishConnection))? + .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::Iggy`] if the address is malformed, the TCP connection fails, connecting + /// takes longer than `CONNECT_TIMEOUT` (an unreachable-and-silently-dropping address, not + /// just a refused one, is covered - see that constant's doc), 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::Iggy`] if the underlying client reports a shutdown failure (e.g. the + /// socket was already in a state that rejects a clean shutdown), or if it takes longer than + /// `REQUEST_TIMEOUT`. + 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::Iggy`] for connectivity/auth failures, or if a + /// call takes longer than `REQUEST_TIMEOUT`. 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::Iggy`] if the mapped stream doesn't exist (see + /// the note above on which resource is actually missing), or if the call takes longer than + /// `REQUEST_TIMEOUT`. Returns [`BridgeError::PartitionOutOfRange`] if any requested + /// partition is beyond the topic's partition count. + pub async fn high_watermarks( + &self, + kafka_topic: &str, + partitions: &[u32], + ) -> Result<Vec<(u32, i64)>, BridgeError> { Review Comment: warning: one `Result` for the whole batch. `ListOffsetsPartitionResponse` carries an error code per partition (`gateways/kafka/src/protocol/responses.rs:233`), so the first out-of-range index raised at `:454` throws away every watermark the call already resolved. `Vec<(u32, Result<i64, BridgeError>)>` keeps the good ones and lets #3537 fill each partition's own `error_code` without falling back to one call per partition. This is the cheap moment for that change: after the handler lands it is a signature change plus a caller rewrite. ########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,918 @@ +// 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. + +//! Integration tests for `IggyBridge` against a real `iggy-server` process - not the +//! `KafkaGateway` under test elsewhere in this suite. `#3533` acceptance criteria this file +//! exercises directly: `ensure_stream_and_topic` idempotent on repeated calls, and the bridge +//! 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, +}; +use secrecy::SecretString; +use serial_test::serial; + +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; + +/// 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)); + } + 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. +/// +/// `assert_cmd::cargo::cargo_bin`, matching `core/integration`'s own +/// `harness::handle::server::start` - not a from-scratch `cargo build` invocation. An earlier +/// version of this function drove `cargo build --package server --bin iggy-server` directly, +/// reasoning that `Command::cargo_bin` "only resolves `CARGO_BIN_EXE_*` for binaries owned by +/// *this* package" - true of the `CARGO_BIN_EXE_*` env-var lookup alone, but `cargo_bin` falls +/// back to `legacy_cargo_bin` when that's unset, which derives the target directory from +/// `env::current_exe()` (this test binary's own path) and looks for a same-named file there - +/// correct under `CARGO_TARGET_DIR`, `--release`, or a `--target` triple subdirectory precisely +/// because it reads back from where cargo actually placed *this* binary, not a guessed path. That +/// version also re-ran a `cargo build` from every one of this file's server-spawning tests +/// (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. +fn iggy_server_binary() -> PathBuf { + assert_cmd::cargo::cargo_bin("iggy-server") +} + +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") Review Comment: warning: this takes the `pin_cores` half of `core/integration/src/harness/handle/server.rs:297-313` but not the `IGGY_SHARDING_CPU_ALLOCATION` half. `core/server/config.toml:712` defaults to `numa:auto`, so every server spawned here still sizes its shard pool to the whole machine, alongside unrelated packages' tests in the same nextest run. The integration harness caps each spawned server at a small core range, and that cap is what lets its own tests run in parallel. Setting it here is what would let the `kafka_bridge` group run wider than one thread. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,503 @@ +// 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 +/// [`CONNECT_TIMEOUT`] for the latter. +const RECONNECTION_RETRIES: u32 = 3; + +/// Wall-clock ceiling on the whole `client.connect()` call in [`IggyBridge::connect`], including +/// every attempt [`RECONNECTION_RETRIES`] makes internally. +/// +/// Without this, an unreachable-but-not-refusing address hangs far longer than "a few seconds": +/// `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, and the plain `TcpStream::connect` underneath has no +/// deadline of its own. Against a firewall that drops SYN packets instead of refusing them, each +/// of the up to `RECONNECTION_RETRIES + 1` dial attempts 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. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Ceiling on a single Iggy client call made *after* [`IggyBridge::connect`] already succeeded - +/// `get_stream`, `create_stream`, `get_topic`, `create_topic`, and `shutdown`. +/// +/// [`CONNECT_TIMEOUT`] alone only bounds the *initial* connect. `TcpClient::send_raw_with_response` +/// (`tcp_client.rs`) reconnects internally on any transport error, mid-call, through the exact +/// same undead-lined `TcpStream::connect` path `CONNECT_TIMEOUT` exists to bound - so a bridge +/// call made after a successful `connect()` (a later network partition, Iggy moved to a new +/// address) could still hang on the kernel's own SYN-retry window with nothing here to stop it. +/// Same value as `CONNECT_TIMEOUT`: both bound the same underlying dial, just entered from a +/// different call site. +const REQUEST_TIMEOUT: Duration = CONNECT_TIMEOUT; + +/// 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`]. +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::Iggy(IggyError::CannotEstablishConnection))? + .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::Iggy`] if the address is malformed, the TCP connection fails, connecting + /// takes longer than `CONNECT_TIMEOUT` (an unreachable-and-silently-dropping address, not + /// just a refused one, is covered - see that constant's doc), 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?; Review Comment: warning: now that every call goes through `with_request_timeout`, `CONNECT_TIMEOUT` bounds nothing on its own. Its only references are its own definition, `const REQUEST_TIMEOUT: Duration = CONNECT_TIMEOUT;` at `:69`, and doc comments. `README.md:84-88` still describes the two as separate bounds applied at separate places. Values are equal today so behaviour is unchanged, but the names now claim a distinction the code does not make. Either apply `CONNECT_TIMEOUT` here and `REQUEST_TIMEOUT` everywhere else, or keep one constant and say so in the README. ########## gateways/kafka/src/bridge/topic_map.rs: ########## @@ -0,0 +1,526 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use serde::Deserialize; + +use crate::bridge::error::BridgeError; + +/// Upper bound `Identifier::named` itself enforces (`identifier/mod.rs`). Checked here too so a +/// too-long name in the mapping file fails at config load with a message that names the file and +/// field at fault, instead of surfacing much later as an opaque `InvalidIdentifier` deep inside +/// `IggyBridge::ensure_stream`/`ensure_topic`. +const MAX_IDENTIFIER_LEN: usize = 255; + +/// Kafka's own topic-name cap (`Topic.MAX_NAME_LENGTH` in Kafka's own source) - smaller than +/// [`MAX_IDENTIFIER_LEN`], so a Kafka-side name needs its own limit, not Iggy's. +const MAX_KAFKA_TOPIC_NAME_LEN: usize = 249; + +/// Rejects an empty name, one with leading/trailing whitespace, or one over +/// [`MAX_IDENTIFIER_LEN`] bytes. +/// +/// Whitespace is rejected outright rather than trimmed and stored: silently trimming here (or +/// adding that later) would change the resolved Iggy stream/topic name under an already-running +/// deployment without the operator changing anything they can see in their own config file. +/// +/// `pub(crate)`: also used by `bridge::config` for `IGGY_KAFKA_IGGY_STREAM`, which feeds +/// `default_stream` through the same no-file path this module's own validation covers on the +/// TOML-file path. For a Kafka-side topic name (a TOML key, or a `kafka_topic` argument), use +/// [`validate_kafka_topic_name`] instead - Kafka's own limits and legal characters differ from +/// Iggy's. +pub(crate) fn validate_identifier_name(field: &str, value: &str) -> Result<(), BridgeError> { + if value.is_empty() { + return Err(BridgeError::InvalidConfig(format!( + "{field} must not be empty" + ))); + } + if value.trim() != value { + return Err(BridgeError::InvalidConfig(format!( + "{field} must not have leading or trailing whitespace: {value:?}" + ))); + } + if value.len() > MAX_IDENTIFIER_LEN { + return Err(BridgeError::InvalidConfig(format!( + "{field} is {} bytes, over the {MAX_IDENTIFIER_LEN}-byte limit Identifier::named \ + enforces: {value:?}", + value.len() + ))); + } + Ok(()) +} + +/// Rejects a Kafka topic name that fails Kafka's own naming rules: empty, leading/trailing +/// whitespace, over [`MAX_KAFKA_TOPIC_NAME_LEN`] bytes, or containing a byte outside +/// `[A-Za-z0-9._-]` (Kafka's own `Topic.legalChars`). +/// +/// Used both for TOML mapping-file keys (`from_toml_str`) and for the `kafka_topic` argument +/// `IggyBridge::ensure_stream_and_topic`/`high_watermark` take directly - the two paths must agree, +/// or a name the config file would reject can still reach Iggy unvalidated through the second +/// path. A whitespace-padded key is the sharpest failure mode this catches: `resolve`'s lookup is +/// an exact `HashMap::get`, so a padded key parses out of the TOML file cleanly and then matches +/// nothing at runtime, silently falling through to the default stream instead of erroring - the +/// one place in this module that would otherwise fail silently instead of loudly. +/// +/// `field` is `"topic mapping key"` or `"kafka_topic"` in current callers, not a generic template - +/// kept as a parameter (like [`validate_identifier_name`]) so the message names the actual site. +pub(crate) fn validate_kafka_topic_name(field: &str, value: &str) -> Result<(), BridgeError> { + if value.is_empty() { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!("{field} must not be empty"), + }); + } + if value.trim() != value { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!("{field} must not have leading or trailing whitespace"), + }); + } + if value.len() > MAX_KAFKA_TOPIC_NAME_LEN { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!( + "{field} is {} bytes, over Kafka's {MAX_KAFKA_TOPIC_NAME_LEN}-byte topic name limit", + value.len() + ), + }); + } + if let Some(illegal) = value Review Comment: warning: `.` is in the allowed byte set, so `.` and `..` clear every check in this function. Kafka's own `Topic.validate` rejects exactly those two names, and the doc at `:68` says this applies Kafka's rules. Not a traversal risk, since Iggy builds on-disk paths from numeric ids, but it is the one place this validator diverges from the rules it claims to implement. Two lines before the character scan. ########## gateways/kafka/src/bridge/topic_map.rs: ########## @@ -0,0 +1,526 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use serde::Deserialize; + +use crate::bridge::error::BridgeError; + +/// Upper bound `Identifier::named` itself enforces (`identifier/mod.rs`). Checked here too so a +/// too-long name in the mapping file fails at config load with a message that names the file and +/// field at fault, instead of surfacing much later as an opaque `InvalidIdentifier` deep inside +/// `IggyBridge::ensure_stream`/`ensure_topic`. +const MAX_IDENTIFIER_LEN: usize = 255; + +/// Kafka's own topic-name cap (`Topic.MAX_NAME_LENGTH` in Kafka's own source) - smaller than +/// [`MAX_IDENTIFIER_LEN`], so a Kafka-side name needs its own limit, not Iggy's. +const MAX_KAFKA_TOPIC_NAME_LEN: usize = 249; + +/// Rejects an empty name, one with leading/trailing whitespace, or one over +/// [`MAX_IDENTIFIER_LEN`] bytes. +/// +/// Whitespace is rejected outright rather than trimmed and stored: silently trimming here (or +/// adding that later) would change the resolved Iggy stream/topic name under an already-running +/// deployment without the operator changing anything they can see in their own config file. +/// +/// `pub(crate)`: also used by `bridge::config` for `IGGY_KAFKA_IGGY_STREAM`, which feeds +/// `default_stream` through the same no-file path this module's own validation covers on the +/// TOML-file path. For a Kafka-side topic name (a TOML key, or a `kafka_topic` argument), use +/// [`validate_kafka_topic_name`] instead - Kafka's own limits and legal characters differ from +/// Iggy's. +pub(crate) fn validate_identifier_name(field: &str, value: &str) -> Result<(), BridgeError> { + if value.is_empty() { + return Err(BridgeError::InvalidConfig(format!( + "{field} must not be empty" + ))); + } + if value.trim() != value { + return Err(BridgeError::InvalidConfig(format!( + "{field} must not have leading or trailing whitespace: {value:?}" + ))); + } + if value.len() > MAX_IDENTIFIER_LEN { + return Err(BridgeError::InvalidConfig(format!( + "{field} is {} bytes, over the {MAX_IDENTIFIER_LEN}-byte limit Identifier::named \ + enforces: {value:?}", + value.len() + ))); + } + Ok(()) +} + +/// Rejects a Kafka topic name that fails Kafka's own naming rules: empty, leading/trailing +/// whitespace, over [`MAX_KAFKA_TOPIC_NAME_LEN`] bytes, or containing a byte outside +/// `[A-Za-z0-9._-]` (Kafka's own `Topic.legalChars`). +/// +/// Used both for TOML mapping-file keys (`from_toml_str`) and for the `kafka_topic` argument +/// `IggyBridge::ensure_stream_and_topic`/`high_watermark` take directly - the two paths must agree, +/// or a name the config file would reject can still reach Iggy unvalidated through the second +/// path. A whitespace-padded key is the sharpest failure mode this catches: `resolve`'s lookup is +/// an exact `HashMap::get`, so a padded key parses out of the TOML file cleanly and then matches +/// nothing at runtime, silently falling through to the default stream instead of erroring - the +/// one place in this module that would otherwise fail silently instead of loudly. +/// +/// `field` is `"topic mapping key"` or `"kafka_topic"` in current callers, not a generic template - +/// kept as a parameter (like [`validate_identifier_name`]) so the message names the actual site. +pub(crate) fn validate_kafka_topic_name(field: &str, value: &str) -> Result<(), BridgeError> { + if value.is_empty() { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!("{field} must not be empty"), + }); + } + if value.trim() != value { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!("{field} must not have leading or trailing whitespace"), + }); + } + if value.len() > MAX_KAFKA_TOPIC_NAME_LEN { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!( + "{field} is {} bytes, over Kafka's {MAX_KAFKA_TOPIC_NAME_LEN}-byte topic name limit", + value.len() + ), + }); + } + if let Some(illegal) = value + .bytes() + .find(|b| !(b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'))) + { + return Err(BridgeError::InvalidKafkaTopicName { + kafka_topic: value.to_string(), + reason: format!( + "{field} contains {:?}, outside Kafka's legal topic-name characters \ + (letters, digits, '.', '_', '-')", + illegal as char + ), + }); + } + Ok(()) +} + +/// Explicit Kafka-topic → Iggy stream/topic override. Absent entries fall back to +/// [`TopicMapping::default_stream`] plus the Kafka topic name unchanged - see +/// [`TopicMapping::resolve`]. +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct TopicOverride { + pub stream: String, + pub topic: String, +} + +/// Kafka topic name → Iggy stream/topic mapping. +/// +/// Default rule (no override): the Iggy stream is [`default_stream`](Self::default_stream) and +/// the Iggy topic name is the Kafka topic name unchanged. A gateway that fronts a single Kafka +/// "cluster" for one Iggy stream never needs an override entry at all. +/// +/// Fields are private and every value is validated (empty/whitespace/length/injectivity - see +/// [`new`](Self::new)) on the only two ways to build one: [`new`](Self::new) and +/// [`from_toml_str`](Self::from_toml_str) (which deserializes into a private, unchecked +/// `RawTopicMapping` first, then calls `new`). Public fields plus `#[derive(Deserialize)]` +/// directly on this type would let any caller construct or mutate one straight from TOML or a +/// literal, skipping every check below - which is exactly what happened before this type had a +/// checked constructor: `bridge::config`'s no-file path built one by hand and needed its own +/// separate validation call to make up for it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TopicMapping { + default_stream: String, + topics: HashMap<String, TopicOverride>, +} + +/// Unchecked shadow of [`TopicMapping`], `Deserialize`'s only target - never constructed by hand, +/// never exposed. `TopicMapping::from_toml_str` is the sole path from TOML to a validated +/// `TopicMapping`, by deserializing into this first and then calling +/// [`TopicMapping::new`](TopicMapping::new). +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawTopicMapping { + default_stream: String, + #[serde(default)] + topics: HashMap<String, TopicOverride>, +} + +impl TopicMapping { + /// Builds a validated `TopicMapping` from already-parsed parts - the checked constructor the + /// no-file `IGGY_KAFKA_IGGY_STREAM` path in `bridge::config` uses, and the sole validation + /// gate `from_toml_str` also funnels through. + /// + /// # Errors + /// + /// Returns [`BridgeError::InvalidConfig`] if `default_stream` or any override's `stream`/ + /// `topic` is empty, has leading/trailing whitespace, or exceeds `MAX_IDENTIFIER_LEN`; if any + /// override key fails Kafka's own topic-naming rules (see + /// `validate_kafka_topic_name`); or if two override entries are not injective (see + /// [`resolve`](Self::resolve)). + pub fn new( + default_stream: String, + topics: HashMap<String, TopicOverride>, + ) -> Result<Self, BridgeError> { + validate_identifier_name("topic mapping's default_stream", &default_stream)?; + + let mut targets = HashSet::with_capacity(topics.len()); + for (kafka_topic, over) in &topics { + validate_kafka_topic_name("topic mapping key", kafka_topic)?; Review Comment: warning: a bad mapping-file key returns `InvalidKafkaTopicName`, which `error.rs:92` sends as `INVALID_TOPIC_EXCEPTION` (17). So an operator's TOML typo reaches a Kafka client as if the client had asked for a bad topic name. This function's own `# Errors` section at `:169-173` promises `InvalidConfig` for exactly this case, and `InvalidConfig` falls to -1, which is the honest answer for a fault the client cannot see. Same mismatch between the doc and the returned variant at `:248`, `:264` and `config.rs:83-85`. ########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,918 @@ +// 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. + +//! Integration tests for `IggyBridge` against a real `iggy-server` process - not the +//! `KafkaGateway` under test elsewhere in this suite. `#3533` acceptance criteria this file +//! exercises directly: `ensure_stream_and_topic` idempotent on repeated calls, and the bridge +//! 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, +}; +use secrecy::SecretString; +use serial_test::serial; + +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; + +/// 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)); + } + 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. +/// +/// `assert_cmd::cargo::cargo_bin`, matching `core/integration`'s own +/// `harness::handle::server::start` - not a from-scratch `cargo build` invocation. An earlier +/// version of this function drove `cargo build --package server --bin iggy-server` directly, +/// reasoning that `Command::cargo_bin` "only resolves `CARGO_BIN_EXE_*` for binaries owned by +/// *this* package" - true of the `CARGO_BIN_EXE_*` env-var lookup alone, but `cargo_bin` falls +/// back to `legacy_cargo_bin` when that's unset, which derives the target directory from +/// `env::current_exe()` (this test binary's own path) and looks for a same-named file there - +/// correct under `CARGO_TARGET_DIR`, `--release`, or a `--target` triple subdirectory precisely +/// because it reads back from where cargo actually placed *this* binary, not a guessed path. That +/// version also re-ran a `cargo build` from every one of this file's server-spawning tests +/// (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. +fn iggy_server_binary() -> PathBuf { + assert_cmd::cargo::cargo_bin("iggy-server") Review Comment: warning: when the binary is missing, this panics with the `assert_cmd` internal message, not the one `gateways/kafka/docs/TEST_SUITE.md:35` promises. `missing_cargo_bin` (`assert_cmd` 2.2.2, `src/cargo.rs:251-273`) prints `CARGO_BIN_EXE_iggy-server` is unset plus the available binary names, because `CARGO_BIN_EXE_iggy-gateway-kafka` is set in this test binary. The documented `binary not found` text never prints. A path existence check first, panicking with `cargo build --package server --bin iggy-server`, hands the reader the command instead of a Cargo internal. ########## gateways/kafka/README.md: ########## @@ -58,6 +58,120 @@ Before check-in, run the procedure in [docs/MANUAL_TESTING.md](docs/MANUAL_TESTI See [docs/SCOPE.md](docs/SCOPE.md) for [#3421](https://github.com/apache/iggy/issues/3421) deliverables, supported API key/version table, and post-foundation TODO backlog. +## Iggy bridge ([#3533](https://github.com/apache/iggy/issues/3533)) + +`src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy +streams/topics, provisions them on demand, and looks up high watermarks (one or many partitions of +a topic per call) for `ListOffsets`. +**Not wired into the live Produce/Fetch dispatch path yet** - that lands with +[#3535](https://github.com/apache/iggy/issues/3535)/[#3536](https://github.com/apache/iggy/issues/3536). +Exercised today by `bridge`'s own unit tests and `tests/bridge_iggy_integration_tests.rs` (spawns a +real `iggy-server`). + +### Connection config + +| Variable | Default | Description | +| --- | --- | --- | +| `IGGY_KAFKA_IGGY_ADDR` | `127.0.0.1:8090` | Address of the Iggy server to bridge to | +| `IGGY_KAFKA_IGGY_USERNAME` | `iggy` | Iggy username | +| `IGGY_KAFKA_IGGY_PASSWORD` | none - **required** | Iggy password. No default: `iggy-server` only uses the well-known `iggy`/`iggy` root credentials when started with `--with-default-root-credentials` (dev-only); otherwise it generates a random password, so a hardcoded default here could never be right and would invite running as root unnoticed | +| `IGGY_KAFKA_IGGY_STREAM` | `kafka` | Default Iggy stream for a Kafka topic with no explicit mapping override | +| `IGGY_KAFKA_TOPIC_MAP_PATH` | unset | Path to a topic-mapping TOML file (see below); omit to use only the default rule | + +The initial connect retries a fixed, bounded number of times (`RECONNECTION_RETRIES = 3`, not the +Iggy SDK client's own default of unlimited retries, one dial per second, forever), and the whole +attempt - retries included - is capped at `CONNECT_TIMEOUT` (15s) wall-clock, so `IggyBridge::connect` +fails in bounded time whether the address refuses the connection or silently drops it, instead of +blocking the calling task indefinitely. Every other bridge call (`ensure_stream_and_topic`, +`high_watermark(s)`, `close`) carries its own `REQUEST_TIMEOUT` (same 15s bound) for the same +reason: the SDK reconnects internally, mid-call, on a transport error, through the same +undead-lined dial path - a bridge call made well after the initial connect can still hit this if +Iggy becomes unreachable later. See `IggyBridge`'s own doc comment (its rustdoc is private, so +this isn't a followable link outside the crate - read the source at `src/bridge/iggy_bridge.rs`). + +### Topic mapping + +Default rule, no config file needed: a Kafka topic `orders` maps to Iggy stream +`IGGY_KAFKA_IGGY_STREAM` (default `kafka`), topic `orders` - the Kafka topic name carries over +unchanged. Override specific topics with a TOML file: + +```toml +default_stream = "kafka" + +[topics.orders] +stream = "billing" +topic = "orders_v2" + +# A Kafka topic name containing dots needs the key quoted, or TOML parses it as nested +# tables ([topics.org] containing [apache] containing [kafka]) instead of one topic named +# "org.apache.kafka.events". +[topics."org.apache.kafka.events"] +stream = "billing" +topic = "kafka_events" +``` + +Point `IGGY_KAFKA_TOPIC_MAP_PATH` at the file to load it; topics not listed under `[topics.*]` +still fall back to the default rule. + +`default_stream` is required in a map file - it has no `#[serde(default)]`, unlike `topics` - +so an override-only file with no `default_stream` key fails to load rather than falling back to +`kafka`. When both `IGGY_KAFKA_TOPIC_MAP_PATH` and `IGGY_KAFKA_IGGY_STREAM` are set, the file's own +`default_stream` always wins and the env var is ignored entirely: a TOML file is a complete mapping +document, not an overlay on top of the env var. + +### Provisioning and idempotency + +`ensure_stream_and_topic(kafka_topic, partition_count)` creates the mapped Iggy stream and topic +if either is missing. Idempotent when repeated with the *same* `partition_count`: a no-op if both +already exist with that count, and a `NameAlreadyExists` race against a concurrent caller creating +the same stream/topic is treated as success, not an error - the goal is "it exists," not "this +call created it." A *different* `partition_count` against an already-existing topic returns +`BridgeError::PartitionCountMismatch` rather than silently keeping the old count or growing it - +two concurrent callers requesting different counts for the same topic must not both see success. + +Topics created this way have **no message expiry** - Iggy's own server default, not Kafka's 7-day +default. Nothing is bounding retention until it's configured explicitly (Iggy's own topic options, +outside this bridge today); repointing a Kafka app that assumes bounded retention onto this bridge +will accumulate data indefinitely unless you set that up yourself. + +They also use Iggy's default **durability**, `Durability::Replicated` - quorum commit without an +additional stable-storage barrier, with the disk write itself threshold-gated (flushed at 1024 +messages or 1 MiB of unflushed data, whichever comes first). Kafka's own defaults take the same +posture, so this isn't a wrong choice, but on a single node both can lose an acked write to a power +cut before that threshold is reached - worth knowing rather than discovering later. + +### Concurrency ceiling + +One `IggyBridge` (one `IggyClient`) is meant to serve every Kafka connection this gateway handles, +and the Iggy SDK's TCP transport is lockstep - one request in flight per client, 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 +(`IGGY_KAFKA_MAX_CONNECTIONS`) does nothing to relieve this. No connection pooling exists yet - it +is a known gap to address before `#3535`/`#3536` put this on a hot path, not a design decision to +rely on. + +### Error mapping + +`BridgeError::to_kafka_error_code()` maps Iggy failures to Kafka wire error codes: + +- Stream/topic/partition not found → `UNKNOWN_TOPIC_OR_PARTITION` (3) +- A rejected *permission* (`Unauthorized`) → `TOPIC_AUTHORIZATION_FAILED` (29) - a real, + fixable-by-the-Kafka-operator ACL problem +- A rejected *login* (the bridge's own `IGGY_KAFKA_IGGY_USERNAME`/`_PASSWORD` are wrong) → + `UNKNOWN_SERVER_ERROR` (-1), deliberately **not** 29 - the Kafka client can't fix a bridge-side + credential misconfiguration, and blaming its own ACLs for one is worse than an unexplained + fatal error +- Connection-shaped failures → `NOT_LEADER_OR_FOLLOWER` (6, the same retriable code the + foundation's own stubs send, so a client backs off and retries) +- An Iggy commit whose outcome is genuinely unknown (`TransientNotCommitted`) → + `REQUEST_TIMED_OUT` (7) - retriable in real Kafka too, chosen because it's what a real broker + sends for the same shape of failure, not to make a client stop retrying +- An invalid Kafka-side topic name (empty, whitespace-padded, oversized, illegal characters) → + `INVALID_TOPIC_EXCEPTION` (17), checked before any Iggy call is made +- `PartitionCountMismatch` → `TOPIC_ALREADY_EXISTS` (36, not `INVALID_PARTITIONS` - that code's + own text is "below 1", a different condition) +- Anything else → `UNKNOWN_SERVER_ERROR` (-1) Review Comment: warning: the closing `Anything else` makes this list read as complete, and `TooManyPartitions` is missing from it. `error.rs:150` sends `INVALID_PARTITIONS` (37) for it and `error.rs:198-201` pins that with a test, so line 171 currently reads as if 37 is never sent at all. It is reachable through `ensure_topic`: `core/server/src/rewrite.rs:218-220` raises `TooManyPartitions` above the partition cap. ########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,918 @@ +// 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. + +//! Integration tests for `IggyBridge` against a real `iggy-server` process - not the +//! `KafkaGateway` under test elsewhere in this suite. `#3533` acceptance criteria this file +//! exercises directly: `ensure_stream_and_topic` idempotent on repeated calls, and the bridge +//! 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, +}; +use secrecy::SecretString; +use serial_test::serial; + +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; + +/// 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)); + } + 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"); Review Comment: warning: this locks its own directory, so it shares no state with `core/integration/src/harness/port_reserver.rs:158`. With an ephemeral floor of 10000 or below both reservers fall back above their ceilings, this band lands at 61000-61199 and the reserver's at 61000-65535, and two suites in the same workspace can hand out one port under two different locks. `mod port_reserver;` at `core/integration/src/harness/mod.rs:51` is private today, so sharing the slot files needs a visibility change first. A disjoint band is the cheaper fix if that is not worth it. ########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,918 @@ +// 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. + +//! Integration tests for `IggyBridge` against a real `iggy-server` process - not the +//! `KafkaGateway` under test elsewhere in this suite. `#3533` acceptance criteria this file +//! exercises directly: `ensure_stream_and_topic` idempotent on repeated calls, and the bridge +//! 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, +}; +use secrecy::SecretString; +use serial_test::serial; + +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; + +/// 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)); + } + 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. +/// +/// `assert_cmd::cargo::cargo_bin`, matching `core/integration`'s own +/// `harness::handle::server::start` - not a from-scratch `cargo build` invocation. An earlier +/// version of this function drove `cargo build --package server --bin iggy-server` directly, +/// reasoning that `Command::cargo_bin` "only resolves `CARGO_BIN_EXE_*` for binaries owned by +/// *this* package" - true of the `CARGO_BIN_EXE_*` env-var lookup alone, but `cargo_bin` falls +/// back to `legacy_cargo_bin` when that's unset, which derives the target directory from +/// `env::current_exe()` (this test binary's own path) and looks for a same-named file there - +/// correct under `CARGO_TARGET_DIR`, `--release`, or a `--target` triple subdirectory precisely +/// because it reads back from where cargo actually placed *this* binary, not a guessed path. That +/// version also re-ran a `cargo build` from every one of this file's server-spawning tests +/// (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. +fn iggy_server_binary() -> PathBuf { + assert_cmd::cargo::cargo_bin("iggy-server") +} + +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") + // `--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 +} + +#[tokio::test] +#[serial] +async fn ensure_stream_and_topic_is_idempotent_on_repeated_calls() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 3) + .await + .expect("first call creates the stream and topic"); + bridge + .ensure_stream_and_topic("orders", 3) + .await + .expect("second call is a no-op against the now-existing stream and topic"); + bridge + .ensure_stream_and_topic("orders", 3) + .await + .expect("third call is still a no-op"); + + // Read back real state, not just three Oks: proves the second and third calls were actually + // no-ops against the one topic the first call created, not e.g. three independent topics + // that all happen to satisfy Ok(()) individually. + let raw = raw_client(&server).await; + let topics = raw + .get_topics(&Identifier::named("kafka").expect("valid stream name")) + .await + .expect("get_topics call"); + assert_eq!(topics.len(), 1, "must be exactly one topic, not three"); + assert_eq!(topics[0].name, "orders"); + assert_eq!(topics[0].partitions_count, 3); +} + +/// Regression test for coverage: nothing previously exercised `PartitionCountMismatch` on a real +/// server - only unit tests constructed the variant directly. +#[tokio::test] +#[serial] +async fn ensure_stream_and_topic_rejects_a_second_call_with_a_different_partition_count() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 3) + .await + .expect("first call creates the topic with 3 partitions"); + + let err = bridge + .ensure_stream_and_topic("orders", 5) + .await + .expect_err( + "a different partition count against an existing topic must not silently succeed", + ); + match err { + BridgeError::PartitionCountMismatch { + topic, + existing, + requested, + } => { + assert_eq!(topic, "orders", "must report the Kafka-side name"); + assert_eq!(existing, 3); + assert_eq!(requested, 5); + } + other => panic!("expected PartitionCountMismatch, got {other:?}"), + } +} + +/// Regression test for coverage: every other test uses an empty `TopicMapping` (`HashMap::new()`), +/// so the Kafka-name-vs-Iggy-name distinction `ensure_topic`/`high_watermark` deliberately +/// maintain in their error paths (`kafka_topic`, not the resolved `topic_name`) can never actually +/// differ and so can never be caught wrong. This test maps a Kafka topic to a differently-named +/// Iggy stream/topic and checks both the happy path and the error paths report the Kafka-side +/// name a real caller (a future handler) would recognize. +#[tokio::test] +#[serial] +async fn bridge_operations_report_the_kafka_side_name_through_a_real_topic_mapping_override() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let mut config = server.test_config(); + let mut topics = HashMap::new(); + topics.insert( + "orders".to_string(), + TopicOverride { + stream: "billing".to_string(), + topic: "orders_v2".to_string(), + }, + ); + config.topic_mapping = TopicMapping::new("kafka".to_string(), topics) + .expect("valid mapping for this test's fixture data"); + let bridge = IggyBridge::connect(config) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 1) + .await + .expect("must create the mapped Iggy stream/topic, not one named after the Kafka topic"); + + // The Iggy-side resources are named differently from the Kafka topic - confirms the mapping + // actually took effect, not that resolve() was a no-op that happened to pass either way. + let raw = raw_client(&server).await; + let billing_topics = raw + .get_topics(&Identifier::named("billing").expect("valid stream name")) + .await + .expect("get_topics call"); + assert_eq!(billing_topics.len(), 1); + assert_eq!(billing_topics[0].name, "orders_v2"); + + let watermark = bridge + .high_watermark("orders", 0) + .await + .expect("must resolve through the mapping, not fail looking for a stream named 'orders'"); + assert_eq!(watermark, 0); + + let out_of_range = bridge + .high_watermark("orders", 5) + .await + .expect_err("partition 5 does not exist on a 1-partition topic"); + assert!( + matches!( + &out_of_range, + BridgeError::PartitionOutOfRange { topic, .. } if topic == "orders" + ), + "error must quote the Kafka topic name 'orders', not the Iggy name 'orders_v2': \ + {out_of_range:?}" + ); + + let mismatch = bridge + .ensure_stream_and_topic("orders", 2) + .await + .expect_err("different partition count against the mapped topic must not silently succeed"); + assert!( + matches!( + &mismatch, + BridgeError::PartitionCountMismatch { topic, .. } if topic == "orders" + ), + "error must quote the Kafka topic name 'orders', not the Iggy name 'orders_v2': \ + {mismatch:?}" + ); +} + +#[tokio::test] +#[serial] +async fn high_watermark_is_zero_for_a_fresh_empty_partition() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 1) + .await + .expect("stream and topic must exist before checking the watermark"); + + let watermark = bridge + .high_watermark("orders", 0) + .await + .expect("fresh topic must report a watermark, not an error"); + assert_eq!( + watermark, 0, + "a freshly created, empty partition's high watermark must be 0" + ); +} + +/// Pins the exact semantics of `Iggy::Partition::current_offset` (offset of the *last written* +/// message, not Kafka's "next offset to produce") against a real server - a test that only +/// checked the empty-topic case would pass under either interpretation and hide an off-by-one. +#[tokio::test] +#[serial] +async fn high_watermark_reflects_produced_messages() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 1) + .await + .expect("stream and topic must exist before producing"); + + let stream_id = Identifier::named("kafka").expect("valid stream name"); + let topic_id = Identifier::named("orders").expect("valid topic name"); + let mut messages: Vec<IggyMessage> = (0..3) + .map(|i| IggyMessage::from(format!("message-{i}"))) + .collect(); + let client = raw_client(&server).await; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("send 3 messages"); + + let watermark = bridge + .high_watermark("orders", 0) + .await + .expect("topic must report a watermark after producing"); + assert_eq!( + watermark, 3, + "high watermark after 3 messages (offsets 0, 1, 2) must be 3, not the last offset (2)" + ); +} + +#[tokio::test] +#[serial] +async fn high_watermark_rejects_out_of_range_partition() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("orders", 1) + .await + .expect("stream and topic must exist before checking the watermark"); + + let err = bridge + .high_watermark("orders", 5) + .await + .expect_err("partition 5 does not exist on a 1-partition topic"); + assert!(matches!(err, BridgeError::PartitionOutOfRange { .. })); +} + +#[tokio::test] +#[serial] +async fn ensure_stream_and_topic_is_idempotent_for_a_numeric_topic_name() { + // Regression test: Identifier::try_from/FromStr parses an all-digit string as a numeric ID, + // not a name - a second call for the same numeric-named topic would look it up by the wrong + // resource kind and fail with StreamIdNotFound/TopicIdNotFound despite the topic existing. + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + + bridge + .ensure_stream_and_topic("2024", 1) + .await + .expect("first call creates the numeric-named stream and topic"); + bridge + .ensure_stream_and_topic("2024", 1) + .await + .expect("second call must still find the numeric-named topic by name, not by ID"); +} + +/// Regression test: `ensure_stream` used to hand `ensure_topic` the stream's *numeric* id +/// (`Identifier::numeric`), and streams are backed by a recycled slab (freed keys are reused by +/// the next created stream) - a stream deleted and recreated between the two calls would leave +/// `ensure_topic` writing into whatever stream now holds that recycled numeric key, not the one +/// `ensure_stream_and_topic`'s caller actually resolved. The fix threads the *named* `Identifier` +/// through instead. This does not (cannot, from a black-box test) hit the exact `await` window the +/// bug lived in, but it does prove the name, not a captured numeric id, is what actually governs: +/// delete the stream, recreate it under the same name, and the very next call must still land the +/// topic on the live incarnation. +#[tokio::test] +#[serial] +async fn ensure_topic_targets_the_streams_live_incarnation_after_a_delete_and_recreate() { Review Comment: warning: this cannot fail on the regression the doc above it describes. The stream is deleted and recreated before the second `ensure_stream_and_topic`, so `ensure_stream` re-reads a live id on that call either way. An implementation that captured `details.id` and returned `Identifier::numeric(...)` passes every assertion at `:695-703`. To make it bite, capture the first incarnation's numeric id, recreate under the same name so the new id differs, then assert the second call still lands on the live incarnation. As written it guards the happy path, not the bug. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
