feat(gateways): Iggy Kafka bridge core (#3533)

gateways/kafka/Cargo.toml

@@ -34,11 +34,14 @@ path = "src/main.rs"
 
 [dependencies]
 bytes = { workspace = true }
+iggy = { workspace = true }
 # Broker-role only: decodes requests and encodes responses. Default features also pull in
 # client-role codec paths and compression codecs (gzip/lz4/snappy/zstd) this gateway never
 # uses, since RecordBatch payloads stay opaque `Bytes` here.
 kafka-protocol = { workspace = true }
 libc = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
 socket2 = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = [
@@ -51,11 +54,14 @@ tokio = { workspace = true, features = [
     "signal",
 ] }
 tokio-util = { workspace = true, features = ["rt"] }
+toml = { workspace = true }
 tracing = { workspace = true }
 tracing-appender = { workspace = true }
 tracing-subscriber = { workspace = true }
 
 [dev-dependencies]
+serial_test = { workspace = true }
+tempfile = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "time"] }
 
 [lints.clippy]

gateways/kafka/README.md

@@ -58,6 +58,62 @@ 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 the high watermark 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` | `iggy` | Iggy password |
+| `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 connection retries a fixed, bounded number of times (not the Iggy SDK client's own default of
+unlimited retries, one dial per second, forever) so a bridge call fails within a few seconds
+against an unreachable Iggy instead of blocking the calling task indefinitely - see
+`IggyBridgeConfig::connection_string`'s doc comment.
+
+### 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"
+```
+
+Point `IGGY_KAFKA_TOPIC_MAP_PATH` at the file to load it; topics not listed under `[topics.*]`
+still fall back to the default rule.
+
+### Provisioning and idempotency
+
+`ensure_stream_and_topic(kafka_topic, partition_count)` creates the mapped Iggy stream and topic
+if either is missing, and is a no-op if both already exist - safe to call on every Produce/Fetch
+for a topic once the handler wiring lands. A `NameAlreadyExists` race against a concurrent caller
+is treated as success, not an error: the goal is "it exists," not "this call created it."
+
+### Error mapping
+
+`BridgeError::to_kafka_error_code()` maps Iggy failures to Kafka wire error codes - stream/topic
+not found → `UNKNOWN_TOPIC_OR_PARTITION` (3), auth/credential failures →
+`TOPIC_AUTHORIZATION_FAILED` (29), 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), anything
+else → `UNKNOWN_SERVER_ERROR` (-1).
+
 ## Wire fixture tool
 
 See [tools/kafka-tool/README.md](tools/kafka-tool/README.md).

gateways/kafka/docs/SCOPE.md

@@ -97,14 +97,22 @@ Full reference for future phases: [`kafka_api_keys_reference.md`](kafka_api_keys
 
 Items from the [hybrid architecture review](https://github.com/apache/iggy/discussions/3252) and maintainer feedback. **Not part of #3421.**
 
-### Phase 2 — Iggy bridge (new issue)
-
-- [ ] Add `bridge/` module (`iggy_bridge`): Produce → `send_messages`, Fetch → `poll_messages`
+### Phase 2 — Iggy bridge
+
+[#3533](https://github.com/apache/iggy/issues/3533) landed the bridge module itself; the items
+below it are still open for the issues that build on top of it.
+
+- [x] Add `bridge/` module (`iggy_bridge`) - connection lifecycle, topic mapping, provisioning,
+      high watermark, error mapping. See [README.md](../README.md#iggy-bridge-3533). Produce →
+      `send_messages` / Fetch → `poll_messages` handler wiring itself is
+      [#3535](https://github.com/apache/iggy/issues/3535)/[#3536](https://github.com/apache/iggy/issues/3536),
+      not part of `bridge/`'s own scope.
+- [x] Idempotent `ensure_stream_and_topic()` (create-if-not-exists) - `src/bridge/iggy_bridge.rs`,
+      exercised end-to-end in `tests/bridge_iggy_integration_tests.rs`.
 - [ ] Document partition mapping in `docs/BRIDGE_MAPPING.md`:
   - Iggy partitions are **0-based** (same as Kafka) — direct `partition_id` mapping, no offset conversion
   - Iggy **consumer groups exist** — map Kafka group APIs to Iggy consumer group APIs
   - Use `Partitioning::balanced()` only when Kafka sends `partition == -1`; otherwise use request partition ID
-- [ ] Idempotent `ensure_stream_and_topic()` (create-if-not-exists)
 - [ ] Real Metadata topology (brokers, partitions, leaders) backed by Iggy state
 
 ### `kafka-protocol` crate adoption — superseded, done differently

gateways/kafka/src/bridge/config.rs

@@ -0,0 +1,184 @@
+// 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::{ExposeSecret, SecretString};
+
+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.
+    ///
+    /// # 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 default_stream =
+            std::env::var("IGGY_KAFKA_IGGY_STREAM").unwrap_or_else(|_| "kafka".to_string());
+
+        let topic_mapping = match std::env::var("IGGY_KAFKA_TOPIC_MAP_PATH") {
+            Ok(path) => TopicMapping::from_file(Path::new(&path))?,
+            Err(_) => TopicMapping {
+                default_stream,
+                topics: std::collections::HashMap::new(),
+            },
+        };
+
+        Ok(Self {
+            address,
+            username,
+            password: SecretString::from(password),
+            topic_mapping,
+        })
+    }
+
+    /// Builds the `iggy://` connection string the SDK's `IggyClientBuilder::from_connection_string`
+    /// expects, embedding credentials. Never pass the result to a `tracing`/`format!` call that
+    /// might reach a log line - it exposes `password` in full, unlike this struct's own `Debug`.
+    ///
+    /// Pins `reconnection_retries` to [`RECONNECTION_RETRIES`] rather than 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.
+    #[must_use]
+    pub fn connection_string(&self) -> String {
+        format!(
+            "iggy://{}:{}@{}?reconnection_retries={RECONNECTION_RETRIES}",
+            self.username,
+            self.password.expose_secret(),
+            self.address
+        )
+    }
+}
+
+/// Passes attempted, after the first, before `IggyBridge::connect` gives up and returns
+/// `Err` - see [`IggyBridgeConfig::connection_string`]'s doc comment for why this is bounded
+/// at all. At the default `reconnection_interval` (1s), a fully unreachable address fails in a
+/// few seconds rather than hanging.
+const RECONNECTION_RETRIES: u32 = 3;
+
+#[cfg(test)]
+mod tests {
+    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 connection_string_embeds_credentials_and_address() {
+        let config = test_config();
+        assert_eq!(
+            config.connection_string(),
+            "iggy://iggy:iggy@127.0.0.1:8090?reconnection_retries=3"
+        );
+    }
+
+    #[test]
+    fn debug_output_does_not_expose_password() {
+        let config = test_config();
+        let debug_output = format!("{config:?}");
+        assert!(
+            !debug_output.contains("iggy://iggy:iggy"),
+            "Debug output must not expose the plaintext password: {debug_output}"
+        );
+    }
+
+    /// Every var `from_env` actually reads must be declared, or a future rename here silently
+    /// desyncs from the allowlist (as opposed to `KNOWN_ENV_VARS` listing a var this module
+    /// never reads, which the compiler can't catch either but is far less consequential).
+    #[test]
+    fn known_env_vars_covers_every_var_from_env_reads() {
+        for var in [
+            "IGGY_KAFKA_IGGY_ADDR",
+            "IGGY_KAFKA_IGGY_USERNAME",
+            "IGGY_KAFKA_IGGY_PASSWORD",
+            "IGGY_KAFKA_IGGY_STREAM",
+            "IGGY_KAFKA_TOPIC_MAP_PATH",
+        ] {
+            assert!(
+                IggyBridgeConfig::KNOWN_ENV_VARS.contains(&var),
+                "{var} read by from_env() but missing from KNOWN_ENV_VARS"
+            );
+        }
+    }
+
+    #[test]
+    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 {
+            std::env::set_var("IGGY_KAFKA_TOPIC_MAP_PATH", "/nonexistent/topic_map.toml");
+        }
+        let result = IggyBridgeConfig::from_env();
+        unsafe {
+            std::env::remove_var("IGGY_KAFKA_TOPIC_MAP_PATH");
+        }
+        assert!(matches!(result, Err(BridgeError::InvalidConfig(_))));
+    }
+}