krishvishal commented on code in PR #4043: URL: https://github.com/apache/iggy/pull/4043#discussion_r3954378533
########## gateways/kafka/src/bridge/error.rs: ########## @@ -0,0 +1,185 @@ +// 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 iggy::prelude::IggyError; +use thiserror::Error; + +use crate::protocol::api::{ + ERROR_INVALID_PARTITIONS, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_TOPIC_AUTHORIZATION_FAILED, + ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, +}; + +/// Errors from the `IggyBridge`: connection lifecycle, config, and Iggy SDK calls. +#[derive(Debug, Error)] +pub enum BridgeError { + /// Bridge config is structurally invalid (empty address, missing credentials, malformed + /// topic-mapping TOML) - caught before any connection attempt. + #[error("invalid bridge configuration: {0}")] + InvalidConfig(String), + /// The Iggy client could not connect or authenticate, or a call failed after connecting. + /// Wraps the SDK's own error rather than re-deriving a parallel taxonomy. + #[error("Iggy client error: {0}")] + Iggy(#[from] IggyError), + /// `high_watermark` was asked about a partition index the topic doesn't have. + #[error( + "partition {partition} out of range for topic '{topic}' ({partitions_count} partitions)" + )] + PartitionOutOfRange { + topic: String, + partition: u32, + partitions_count: u32, + }, + /// `ensure_topic` was asked to ensure a topic that already exists with a different partition + /// count. `ensure_topic`'s whole contract is "the topic has `partition_count` partitions + /// afterward" - silently keeping the old count and returning `Ok(())` would let two + /// concurrent callers requesting different counts for the same topic both believe they + /// succeeded. + #[error( + "topic '{topic}' already exists with {existing} partitions, but {requested} were requested" + )] + PartitionCountMismatch { + topic: String, + existing: u32, + requested: u32, + }, +} + +impl BridgeError { + /// Maps this error to the Kafka protocol error code a handler should answer with. + /// + /// Connection-shaped failures reuse `NOT_LEADER_OR_FOLLOWER` (6) - the same retriable code + /// the foundation's Produce/Fetch stubs already send - so a client backs off and retries + /// rather than treating a transient Iggy outage as a permanent failure. Not-found maps to + /// `UNKNOWN_TOPIC_OR_PARTITION` (3). Anything without a closer analogue falls back to + /// `UNKNOWN_SERVER_ERROR` (-1). + #[must_use] + pub const fn to_kafka_error_code(&self) -> i16 { + match self { + Self::Iggy(err) => iggy_error_to_kafka_code(err), + Self::PartitionOutOfRange { .. } => ERROR_UNKNOWN_TOPIC_OR_PARTITION, + Self::PartitionCountMismatch { .. } => ERROR_INVALID_PARTITIONS, + // Not a wire-response case in practice: an invalid bridge config is caught at + // `IggyBridge::connect` before any handler exists to answer a Kafka request, so this + // is reachable only if a future caller starts constructing configs at request time. + // `UNKNOWN_SERVER_ERROR` at least doesn't claim a specific, wrong cause the way + // `UNSUPPORTED_VERSION` (misleadingly implies a Kafka API version mismatch) would. + Self::InvalidConfig(_) => ERROR_UNKNOWN_SERVER_ERROR, + } + } +} + +/// Kept private so this association can change without touching call sites. +const fn iggy_error_to_kafka_code(err: &IggyError) -> i16 { + match err { + IggyError::StreamIdNotFound(_) + | IggyError::StreamNameNotFound(_) + | IggyError::TopicIdNotFound(_, _) + | IggyError::TopicNameNotFound(_, _) => ERROR_UNKNOWN_TOPIC_OR_PARTITION, + IggyError::Unauthenticated + | IggyError::Unauthorized + | IggyError::InvalidCredentials + | IggyError::InvalidUsername + | IggyError::InvalidPassword => ERROR_TOPIC_AUTHORIZATION_FAILED, + IggyError::Disconnected | IggyError::CannotEstablishConnection => { + ERROR_NOT_LEADER_OR_FOLLOWER + } + _ => ERROR_UNKNOWN_SERVER_ERROR, Review Comment: **CRITICAL.** The Kafka error map doesn't match what the SDK returns. The SDK's retriable set is at `tcp_client.rs:215-224`: `Disconnected | EmptyResponse | Unauthenticated | StaleClient | NotConnected | CannotEstablishConnection | TcpError`. This map covers two of seven; the rest fall to `_ => ERROR_UNKNOWN_SERVER_ERROR`, which `kafka-protocol-0.18.0/src/error.rs:183` marks `retriable=false`. A producer drops the batch instead of retrying. Two of the missing five are common: `send_raw` returns `NotConnected` for any call during a reconnect (`tcp_client.rs:1234-1241`), and an Iggy view change raises `TransientNotAccepted` (58), doc string "retry, on any replica". Opposite problem in the same match: `Unauthenticated` maps to 29, also `retriable=false`, but `binary_auth.rs:27` shows it's the window before auto-sign-in completes. A reconnect kills a consumer with "Not authorized to access topics", pointing at ACLs this gateway doesn't have. Fix: use the SDK's list; `Unauthenticated` to 6; keep `ClientShutdown` on `-1`. `TransientNotCommitted` means unknown outcome, so retry risks a duplicate; use `REQUEST_TIMED_OUT` (7), not 6, to make that explicit. `unauthenticated_maps_to_topic_authorization_failed` asserts the old mapping and needs updating. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,310 @@ +// 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 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; + +/// 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. At the default `reconnection_interval` (1s), a fully unreachable address fails in a few +/// seconds rather than hanging. +const RECONNECTION_RETRIES: u32 = 3; + +/// 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`. +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, 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)?; + client.connect().await.map_err(BridgeError::Iggy)?; + info!("Iggy bridge connected to {}", config.address); + + Ok(Self { client, config }) + } + + /// Disconnects the underlying Iggy client. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the client reports a disconnect failure. + pub async fn close(&self) -> Result<(), BridgeError> { + self.client.disconnect().await.map_err(BridgeError::Iggy) Review Comment: **CRITICAL.** `close()` doesn't close. `disconnect()` (`client.rs:993-995`) never touches `heartbeat_handle`; only `Drop for IggyClient` (`:933-943`) aborts it. Five seconds later the heartbeat pings, gets `NotConnected` (in the retriable set), and `sign_in_credentials()` still returns the configured `AutoLogin`, so it redials and re-authenticates. Dropping the bridge is safer than closing it. Fix: `self.client.shutdown()`, and take `self` by value. Costs: `shutdown()` is terminal and can `Err` on a live socket; `close(self)` can't be called through an `Arc`. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,310 @@ +// 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 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; + +/// 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. At the default `reconnection_interval` (1s), a fully unreachable address fails in a few Review Comment: **CRITICAL.** `connect()` has no dial timeout, and `README.md:79-82` repeats this claim. `tcp_client.rs:1144`: `if candidates.len() < 2 { return self.establish(...) }` skips `FAILOVER_DIAL_TIMEOUT`, and `establish` (`:1034`) is a bare `TcpStream::connect`. A bridge config has one address. Against a firewall that DROPs, that's 4 passes at the kernel SYN timeout, roughly 8.5 min, not "a few seconds". The test dials an unbound loopback port, which RSTs instantly, so it only covers the easy case. Fix: `tokio::time::timeout`, and put the real bound in the doc. Set it above p99 handshake, not 2s, or slow-but-alive servers start failing. ########## gateways/kafka/src/bridge/config.rs: ########## @@ -0,0 +1,204 @@ +// 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; + +const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090"; +/// Matches the Iggy server's own default root user - not a made-up example, the same default +/// every fresh `iggy-server` and every CLI quick-start in this repo uses. +const DEFAULT_IGGY_USERNAME: &str = "iggy"; +const DEFAULT_IGGY_PASSWORD: &str = "iggy"; Review Comment: **CRITICAL.** This comment is wrong. `args.rs:109-120` shows `--with-default-root-credentials` defaults to `false` and is labelled "INSECURE - FOR DEVELOPMENT ONLY!"; otherwise the server generates a random password (`credentials.rs:112-118`). The gateway quietly runs as root on any dev-flagged Iggy, with the credential in no config file and no log. Fix: drop `DEFAULT_IGGY_PASSWORD`, require `IGGY_KAFKA_IGGY_PASSWORD`, fix the comment. `helm/.../deployment.yaml:120-143` already has the `secretKeyRef` pattern. Quick-starts then need two env vars, which is cheap now and not later. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,310 @@ +// 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 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; + +/// 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. At the default `reconnection_interval` (1s), a fully unreachable address fails in a few +/// seconds rather than hanging. +const RECONNECTION_RETRIES: u32 = 3; + +/// 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`. +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, 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)?; + client.connect().await.map_err(BridgeError::Iggy)?; + info!("Iggy bridge connected to {}", config.address); + + Ok(Self { client, config }) + } + + /// Disconnects the underlying Iggy client. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the client reports a disconnect failure. + pub async fn close(&self) -> Result<(), BridgeError> { + self.client.disconnect().await.map_err(BridgeError::Iggy) + } + + /// 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`]. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] for connectivity/auth/invalid-name failures. Returns + /// [`BridgeError::PartitionCountMismatch`] if the topic already exists with a different + /// partition count than `partition_count`. + pub async fn ensure_stream_and_topic( + &self, + kafka_topic: &str, + partition_count: u32, + ) -> Result<(), BridgeError> { + 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(()) + } + + /// Looks up (or creates) the stream named `stream_name`. + /// + /// `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. + async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> { + let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?; + if let Some(existing) = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + { + debug!("Iggy stream '{stream_name}' already exists"); + return Identifier::numeric(existing.id).map_err(BridgeError::Iggy); Review Comment: **CRITICAL.** This passes `ensure_topic` a reusable slab key. The doc at `:125-130` argues for `Identifier::named` over numeric, then converts to numeric anyway, while `high_watermark:275` resolves by name. `stream.rs:1595-1599` says keys are recycled on delete. Delete and recreate a stream between the two round trips and `create_topic` writes into the wrong stream, returning `Ok(())`. Fix: pass the `Identifier::named(stream_name)` already built at `:132`. Nothing breaks, and it removes three dead `.map_err` (`Identifier::numeric` is infallible) plus the `:226` error that prints a slab key where a stream name belongs. ########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,435 @@ +// 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, +}; +use secrecy::SecretString; +use serial_test::serial; + +use iggy_gateway_kafka::bridge::{BridgeError, IggyBridge, IggyBridgeConfig, TopicMapping}; + +/// First port of the local reservation band. `u16` has no room above the highest ephemeral +/// ceiling in use (macOS: 49152-65535), so "clear of ephemeral" only leaves room below the +/// lowest floor (Linux default: 32768) - picked here below `core/integration`'s own +/// `port_reserver.rs` band too (20000 up to that floor), so a `bind(0)` call in an unrelated +/// process, including that harness's own test servers, is never handed a port this band claims. +const PORT_LOCK_BAND_START: u16 = 15000; +const PORT_LOCK_BAND_SLOTS: u16 = 200; + +/// 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, nothing ever binds the port on this side, so there is no TOCTOU window between +/// picking it and `iggy-server` binding it: the lock, not a socket, is the allocator. +struct PortGuard { + port: u16, + _lock: File, +} + +impl PortGuard { + fn acquire() -> Self { + 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..PORT_LOCK_BAND_SLOTS { + let port = PORT_LOCK_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 [{PORT_LOCK_BAND_START}, {}]", + PORT_LOCK_BAND_START + PORT_LOCK_BAND_SLOTS - 1 + ); + } +} + +/// Builds `iggy-server` (idempotent - a no-op rebuild once already current) and returns its path. +/// +/// Not `assert_cmd::Command::cargo_bin`: that only resolves `CARGO_BIN_EXE_*` for binaries owned +/// by *this* package (confirmed - it fails here with "available binary names are +/// iggy-gateway-kafka"). `iggy-server` belongs to the separate `server` crate, and neither this +/// crate nor `core/integration` (same `Command::cargo_bin` pattern) declares that crate as a +/// dependency just to make its binary buildable. Driving `cargo build` directly sidesteps that +/// entirely - no Cargo.toml dependency edge needed on a crate this one otherwise never touches. +/// +/// Reads the artifact path from `--message-format=json` rather than guessing +/// `target/debug/iggy-server`: a guessed path breaks under `CARGO_TARGET_DIR` (this workspace's +/// own coverage CI sets it), `--release`, or a `--target` triple subdirectory, none of which +/// `cargo build`'s own JSON output leaves to guesswork. +fn iggy_server_binary() -> &'static PathBuf { + static BINARY_PATH: OnceLock<PathBuf> = OnceLock::new(); + BINARY_PATH.get_or_init(|| { + let output = Command::new(env!("CARGO")) + .args([ + "build", + "--package", + "server", + "--bin", + "iggy-server", + "--message-format=json", + ]) + .output() + .expect("run cargo build for iggy-server"); + assert!( + output.status.success(), + "cargo build --package server --bin iggy-server failed:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + + String::from_utf8_lossy(&output.stdout) + .lines() + .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok()) + .find_map(|message| { + if message.get("reason")? == "compiler-artifact" + && message.get("target")?.get("name")? == "iggy-server" + { + message.get("executable")?.as_str().map(PathBuf::from) + } else { + None + } + }) + .expect("cargo build --message-format=json reported no iggy-server executable") + }) +} + +struct TestServer { + child: Child, + address: String, + _port_guard: PortGuard, +} + +impl TestServer { + /// Spawns `iggy-server` with an isolated temp data dir and a locked TCP port, then blocks + /// until a bridge connection succeeds or the startup budget is exhausted. + async fn spawn(data_dir: &std::path::Path) -> 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_SYSTEM_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") + // `--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", "iggy"); + let child = command.spawn().expect("spawn iggy-server"); + + let server = Self { + child, + address, + _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. + async fn wait_ready(&self) { Review Comment: **CRITICAL.** This harness reimplements the existing one, minus the safety parts. `.config/nextest.toml:24-26` says in-repo that nextest ignores `#[serial]`, and no `threads-required` override was added. So up to 4 tests run at once, each spawning a server that inherits `cpu_allocation = "numa:auto"` and `pin_cores = true` (`config.toml:873,883`) - `harness/handle/server.rs:311-319` overrides both because pinned shards of concurrent tests starve each other. `wait_ready` never calls `try_wait()`, so a server that dies at boot gives a 30s spin and "did not become ready", naming the wrong cause. `Drop` SIGKILLs where `graceful_kill` exists. Hits every CI run, x3 with `retries`. Fix: set the two sharding env vars, add `try_wait()`, capture child stdio. A 1-shard server stops covering multi-shard dispatch, which this file covers by accident today. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,310 @@ +// 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 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; + +/// 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. At the default `reconnection_interval` (1s), a fully unreachable address fails in a few +/// seconds rather than hanging. +const RECONNECTION_RETRIES: u32 = 3; + +/// 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`. +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, 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)?; + client.connect().await.map_err(BridgeError::Iggy)?; + info!("Iggy bridge connected to {}", config.address); + + Ok(Self { client, config }) + } + + /// Disconnects the underlying Iggy client. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the client reports a disconnect failure. + pub async fn close(&self) -> Result<(), BridgeError> { + self.client.disconnect().await.map_err(BridgeError::Iggy) + } + + /// 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`]. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] for connectivity/auth/invalid-name failures. Returns + /// [`BridgeError::PartitionCountMismatch`] if the topic already exists with a different + /// partition count than `partition_count`. + pub async fn ensure_stream_and_topic( + &self, + kafka_topic: &str, + partition_count: u32, + ) -> Result<(), BridgeError> { + 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(()) + } + + /// Looks up (or creates) the stream named `stream_name`. + /// + /// `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. + async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> { + let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?; + if let Some(existing) = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + { + debug!("Iggy stream '{stream_name}' already exists"); + return Identifier::numeric(existing.id).map_err(BridgeError::Iggy); + } + + match self.client.create_stream(stream_name).await { + Ok(created) => { + info!("created Iggy stream '{stream_name}'"); + Identifier::numeric(created.id).map_err(BridgeError::Iggy) + } + Err(IggyError::StreamNameAlreadyExists(_)) => { + // Lost a create race - re-verify by name (the identifier this arm already has, + // fixed above) rather than trusting the race outcome alone. + let existing = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + .ok_or_else(|| IggyError::StreamNameNotFound(stream_name.to_string()))?; + Identifier::numeric(existing.id).map_err(BridgeError::Iggy) + } + Err(err) => Err(BridgeError::Iggy(err)), + } + } + + /// Looks up (or creates) the topic named `topic_name` under `stream_id`. + /// + /// `Identifier::named`, not `Identifier::try_from` - see [`Self::ensure_stream`]'s doc + /// comment; the same numeric-name ambiguity applies to topic names. + 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) = self + .client + .get_topic(stream_id, &identifier) + .await + .map_err(BridgeError::Iggy)? + { + 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(()); + } + + let options = TopicCreateOptions { + partitions_count: Some(partition_count), + ..TopicCreateOptions::default() + }; + match self + .client + .create_topic(stream_id, topic_name, &options) + .await + { + Ok(_) => { + info!("created Iggy topic '{topic_name}' with {partition_count} partitions"); + Ok(()) + } + Err(IggyError::TopicNameAlreadyExists(_, _)) => { + // Lost a create race - re-verify by name rather than trusting the race outcome + // alone, the same way ensure_stream's arm does. The winner of the race 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(()). + let existing = self + .client + .get_topic(stream_id, &identifier) + .await + .map_err(BridgeError::Iggy)? + .ok_or_else(|| { + 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(BridgeError::Iggy(err)), + } + } + + /// Returns the high watermark (offset the next produced message would receive) for one + /// partition of the Iggy topic `kafka_topic` maps to. + /// + /// 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. + /// + /// `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` as segments are dropped + /// (`core/partitions/src/iggy_partition.rs::decrement_messages_count`) but never rewinds + /// `current_offset` - a fully-retention-purged partition that has produced messages reports + /// `messages_count == 0` with `current_offset` still at its high value. Treating that as + /// "empty" would report high watermark `0` for a partition that has, in fact, produced past + /// offset `0`; a future `ListOffsets` (`#3537`) LATEST built on this would rewind instead of + /// pointing at the true next-write position. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the mapped stream/topic doesn't exist. Returns + /// [`BridgeError::PartitionOutOfRange`] if `partition` is beyond the topic's partition count. + pub async fn high_watermark( + &self, + kafka_topic: &str, + partition: u32, + ) -> Result<u64, BridgeError> { + let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic); + let stream_id = Identifier::named(&stream_name).map_err(BridgeError::Iggy)?; + let topic_id = Identifier::named(&topic_name).map_err(BridgeError::Iggy)?; + let details = self + .client + .get_topic(&stream_id, &topic_id) + .await + .map_err(BridgeError::Iggy)? + .ok_or_else(|| { + BridgeError::Iggy(IggyError::TopicNameNotFound( + topic_name.clone(), + stream_name.clone(), + )) + })?; + + let partition_details = details + .partitions + .iter() + .find(|p| p.id == partition) + .ok_or_else(|| BridgeError::PartitionOutOfRange { + // The Kafka-side name a caller (a future ListOffsets handler) actually asked + // about, not `topic_name` - a mapping override would otherwise quote the wrong + // (Iggy-side) name back at a Kafka client that never heard of it. + topic: kafka_topic.to_string(), + partition, + partitions_count: details.partitions_count, + })?; + + Ok( + if partition_details.messages_count == 0 && partition_details.current_offset == 0 { Review Comment: This `(0,0)` check conflates five states. `responses.rs:1436` `map_or((0,0,0))` returns zeros for a partition the shard hasn't materialized, so one holding 1M messages reports HW 0. Not fixable client-side: `offset_frontier()` (`iggy_partition.rs:1274-1280`) has the right answer but it isn't on the wire. The code matches `iggy_partition.rs:2681-2683`, so it's defensible; the doc at `:250-263` claims a correctness it doesn't have, and #3537 shouldn't build on it as written. ########## gateways/kafka/src/bridge/iggy_bridge.rs: ########## @@ -0,0 +1,310 @@ +// 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 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; + +/// 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. At the default `reconnection_interval` (1s), a fully unreachable address fails in a few +/// seconds rather than hanging. +const RECONNECTION_RETRIES: u32 = 3; + +/// 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`. +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, 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)?; + client.connect().await.map_err(BridgeError::Iggy)?; + info!("Iggy bridge connected to {}", config.address); + + Ok(Self { client, config }) + } + + /// Disconnects the underlying Iggy client. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the client reports a disconnect failure. + pub async fn close(&self) -> Result<(), BridgeError> { + self.client.disconnect().await.map_err(BridgeError::Iggy) + } + + /// 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`]. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] for connectivity/auth/invalid-name failures. Returns + /// [`BridgeError::PartitionCountMismatch`] if the topic already exists with a different + /// partition count than `partition_count`. + pub async fn ensure_stream_and_topic( + &self, + kafka_topic: &str, + partition_count: u32, + ) -> Result<(), BridgeError> { + 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(()) + } + + /// Looks up (or creates) the stream named `stream_name`. + /// + /// `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. + async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> { + let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?; + if let Some(existing) = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + { + debug!("Iggy stream '{stream_name}' already exists"); + return Identifier::numeric(existing.id).map_err(BridgeError::Iggy); + } + + match self.client.create_stream(stream_name).await { + Ok(created) => { + info!("created Iggy stream '{stream_name}'"); + Identifier::numeric(created.id).map_err(BridgeError::Iggy) + } + Err(IggyError::StreamNameAlreadyExists(_)) => { + // Lost a create race - re-verify by name (the identifier this arm already has, + // fixed above) rather than trusting the race outcome alone. + let existing = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + .ok_or_else(|| IggyError::StreamNameNotFound(stream_name.to_string()))?; + Identifier::numeric(existing.id).map_err(BridgeError::Iggy) + } + Err(err) => Err(BridgeError::Iggy(err)), + } + } + + /// Looks up (or creates) the topic named `topic_name` under `stream_id`. + /// + /// `Identifier::named`, not `Identifier::try_from` - see [`Self::ensure_stream`]'s doc + /// comment; the same numeric-name ambiguity applies to topic names. + 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) = self + .client + .get_topic(stream_id, &identifier) + .await + .map_err(BridgeError::Iggy)? + { + 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(()); + } + + let options = TopicCreateOptions { Review Comment: `TopicCreateOptions::default()` gives `message_expiry: ServerDefault`, which `segment_cleaner.rs:93,98` treats as never expire. Kafka defaults to 7 days. Nothing is frozen on disk (`max_topic_size: ServerDefault` re-resolves at enforcement), but it surprises anyone repointing a Kafka app. ########## gateways/kafka/src/bridge/config.rs: ########## @@ -0,0 +1,204 @@ +// 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; + +const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090"; +/// Matches the Iggy server's own default root user - not a made-up example, the same default +/// every fresh `iggy-server` and every CLI quick-start in this repo uses. +const DEFAULT_IGGY_USERNAME: &str = "iggy"; +const DEFAULT_IGGY_PASSWORD: &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. Mirrors `main.rs`'s + /// `KNOWN_KAFKA_ENV_VARS` guard - add new vars to both, 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 credentials. + /// + /// `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_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()); + let password = std::env::var("IGGY_KAFKA_IGGY_PASSWORD") + .unwrap_or_else(|_| DEFAULT_IGGY_PASSWORD.to_string()); + let stream_env = std::env::var("IGGY_KAFKA_IGGY_STREAM").ok(); + // Caught here, not left to TopicMapping::from_toml_str's own empty check: that + // validation only runs when IGGY_KAFKA_TOPIC_MAP_PATH is set, since the no-file branch + // below builds a TopicMapping directly rather than through the validating constructor. + // An explicitly-empty (not merely absent) IGGY_KAFKA_IGGY_STREAM would otherwise pass + // from_env cleanly and only fail much later, deep in the first ensure_stream_and_topic + // call, as an opaque Identifier::named("") error. + if let Some(ref stream) = stream_env + && stream.trim().is_empty() + { + return Err(BridgeError::InvalidConfig( + "IGGY_KAFKA_IGGY_STREAM must not be empty".to_string(), + )); + } + let topic_map_path = std::env::var("IGGY_KAFKA_TOPIC_MAP_PATH").ok(); + + let topic_mapping = match topic_map_path { + Some(path) => { + if stream_env.is_some() { + warn!( + "IGGY_KAFKA_IGGY_STREAM is set but ignored: IGGY_KAFKA_TOPIC_MAP_PATH's \ + own default_stream takes precedence" + ); + } + TopicMapping::from_file(Path::new(&path))? + } + None => TopicMapping { + default_stream: stream_env.unwrap_or_else(|| "kafka".to_string()), + topics: std::collections::HashMap::new(), + }, + }; + + Ok(Self { + address, + username, + password: SecretString::from(password), + topic_mapping, + }) + } +} + +#[cfg(test)] +mod tests { + use serial_test::serial; + + use super::*; + + fn test_config() -> IggyBridgeConfig { + IggyBridgeConfig { + address: "127.0.0.1:8090".to_string(), + username: "iggy".to_string(), + password: SecretString::from("iggy"), + topic_mapping: TopicMapping { + default_stream: "kafka".to_string(), + topics: std::collections::HashMap::new(), + }, + } + } + + #[test] + fn debug_output_does_not_expose_password() { + // A distinctive value, not the shared fixture's "iggy" - that string also appears as the + // username and inside the struct's own name, which would make a substring check here + // pass trivially regardless of whether the password field itself is actually redacted. + let mut config = test_config(); + config.password = SecretString::from("correct-horse-battery-staple"); + let debug_output = format!("{config:?}"); + assert!( + !debug_output.contains("correct-horse-battery-staple"), + "Debug output must not expose the plaintext password: {debug_output}" + ); + } + + #[test] + #[serial] + fn from_env_rejects_missing_topic_map_file() { + // Safety: single-threaded within this function; no other test in this crate touches + // IGGY_KAFKA_TOPIC_MAP_PATH. + unsafe { Review Comment: `unsafe { set_var }` races a concurrent `getenv`: `server.rs:891` `init_tracing_is_idempotent` calls `EnvFilter::try_from_default_env()`, has no `#[serial]`, same lib test binary. Real under `cargo test`, invisible under nextest. The three `# Safety` comments state the wrong condition - edition 2024's rule is process-wide. ########## gateways/kafka/src/bridge/error.rs: ########## @@ -0,0 +1,185 @@ +// 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 iggy::prelude::IggyError; +use thiserror::Error; + +use crate::protocol::api::{ + ERROR_INVALID_PARTITIONS, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_TOPIC_AUTHORIZATION_FAILED, + ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, +}; + +/// Errors from the `IggyBridge`: connection lifecycle, config, and Iggy SDK calls. +#[derive(Debug, Error)] +pub enum BridgeError { + /// Bridge config is structurally invalid (empty address, missing credentials, malformed + /// topic-mapping TOML) - caught before any connection attempt. + #[error("invalid bridge configuration: {0}")] + InvalidConfig(String), + /// The Iggy client could not connect or authenticate, or a call failed after connecting. + /// Wraps the SDK's own error rather than re-deriving a parallel taxonomy. + #[error("Iggy client error: {0}")] + Iggy(#[from] IggyError), + /// `high_watermark` was asked about a partition index the topic doesn't have. + #[error( + "partition {partition} out of range for topic '{topic}' ({partitions_count} partitions)" + )] + PartitionOutOfRange { + topic: String, + partition: u32, + partitions_count: u32, + }, + /// `ensure_topic` was asked to ensure a topic that already exists with a different partition + /// count. `ensure_topic`'s whole contract is "the topic has `partition_count` partitions + /// afterward" - silently keeping the old count and returning `Ok(())` would let two + /// concurrent callers requesting different counts for the same topic both believe they + /// succeeded. + #[error( + "topic '{topic}' already exists with {existing} partitions, but {requested} were requested" + )] + PartitionCountMismatch { + topic: String, + existing: u32, + requested: u32, + }, +} + +impl BridgeError { + /// Maps this error to the Kafka protocol error code a handler should answer with. + /// + /// Connection-shaped failures reuse `NOT_LEADER_OR_FOLLOWER` (6) - the same retriable code + /// the foundation's Produce/Fetch stubs already send - so a client backs off and retries + /// rather than treating a transient Iggy outage as a permanent failure. Not-found maps to + /// `UNKNOWN_TOPIC_OR_PARTITION` (3). Anything without a closer analogue falls back to + /// `UNKNOWN_SERVER_ERROR` (-1). + #[must_use] + pub const fn to_kafka_error_code(&self) -> i16 { + match self { + Self::Iggy(err) => iggy_error_to_kafka_code(err), + Self::PartitionOutOfRange { .. } => ERROR_UNKNOWN_TOPIC_OR_PARTITION, + Self::PartitionCountMismatch { .. } => ERROR_INVALID_PARTITIONS, Review Comment: `PartitionCountMismatch` maps to `INVALID_PARTITIONS (37)`, whose Kafka text is "Number of partitions is below 1". `TOPIC_ALREADY_EXISTS (36)` fits. ########## gateways/kafka/src/bridge/topic_map.rs: ########## @@ -0,0 +1,203 @@ +// 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; +use std::path::Path; + +use serde::Deserialize; + +use crate::bridge::error::BridgeError; + +/// 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)] +pub struct TopicOverride { + pub stream: String, + pub topic: String, +} + +/// Kafka topic name → Iggy stream/topic mapping, loaded from TOML. +/// +/// 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. +#[derive(Debug, Clone, Deserialize, Default, PartialEq, Eq)] +pub struct TopicMapping { + pub default_stream: String, + #[serde(default)] + pub topics: HashMap<String, TopicOverride>, +} + +impl TopicMapping { + /// Resolves a Kafka topic name to `(iggy_stream, iggy_topic)`. + #[must_use] + pub fn resolve(&self, kafka_topic: &str) -> (String, String) { + self.topics.get(kafka_topic).map_or_else( Review Comment: `resolve` isn't injective: `[topics.orders] stream="kafka", topic="orders_v2"` silently merges Kafka topics `orders` and `orders_v2` into one Iggy topic. That's the main reason anyone writes an override file. ########## gateways/kafka/src/bridge/topic_map.rs: ########## @@ -0,0 +1,203 @@ +// 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; +use std::path::Path; + +use serde::Deserialize; + +use crate::bridge::error::BridgeError; + +/// 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)] +pub struct TopicOverride { + pub stream: String, + pub topic: String, +} + +/// Kafka topic name → Iggy stream/topic mapping, loaded from TOML. +/// +/// 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. +#[derive(Debug, Clone, Deserialize, Default, PartialEq, Eq)] Review Comment: No `#[serde(deny_unknown_fields)]`, no `version` key. A singular `[topic.orders]` typo parses fine, gives an empty map, and everything falls back to the default stream. Also `TopicMapping` derives `Default`, giving `default_stream: ""` - exactly what `from_toml_str:68-72` rejects. -- 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]
