From 041bd7ec3bbf944b66c3d23c4f2d19339302930c Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Thu, 3 Sep 2026 06:00:28 +0530 Subject: [PATCH 1/2] Initial version of bridge implementation --- Cargo.lock | 6 + gateways/kafka/Cargo.toml | 6 + gateways/kafka/README.md | 56 +++++ gateways/kafka/docs/SCOPE.md | 16 +- gateways/kafka/src/bridge/config.rs | 184 ++++++++++++++ gateways/kafka/src/bridge/error.rs | 137 +++++++++++ gateways/kafka/src/bridge/iggy_bridge.rs | 204 ++++++++++++++++ gateways/kafka/src/bridge/mod.rs | 32 +++ gateways/kafka/src/bridge/topic_map.rs | 166 +++++++++++++ gateways/kafka/src/lib.rs | 1 + gateways/kafka/src/main.rs | 10 + .../tests/bridge_iggy_integration_tests.rs | 227 ++++++++++++++++++ 12 files changed, 1041 insertions(+), 4 deletions(-) create mode 100644 gateways/kafka/src/bridge/config.rs create mode 100644 gateways/kafka/src/bridge/error.rs create mode 100644 gateways/kafka/src/bridge/iggy_bridge.rs create mode 100644 gateways/kafka/src/bridge/mod.rs create mode 100644 gateways/kafka/src/bridge/topic_map.rs create mode 100644 gateways/kafka/tests/bridge_iggy_integration_tests.rs diff --git a/Cargo.lock b/Cargo.lock index c6aa065741..a9f4ed0202 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6843,12 +6843,18 @@ name = "iggy-gateway-kafka" version = "0.1.0" dependencies = [ "bytes", + "iggy", "kafka-protocol", "libc", + "secrecy", + "serde", + "serial_test", "socket2 0.6.5", + "tempfile", "thiserror 2.0.19", "tokio", "tokio-util", + "toml 1.1.3+spec-1.1.0", "tracing", "tracing-appender", "tracing-subscriber", diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml index 648cd70385..211c3e75ed 100644 --- a/gateways/kafka/Cargo.toml +++ b/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 = { version = "0.17", default-features = false, features = ["broker"] } 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] diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 14984f44f0..9b8a8b943c 100644 --- a/gateways/kafka/README.md +++ b/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). diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index 0b510a5fb5..d343c45cd9 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/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 diff --git a/gateways/kafka/src/bridge/config.rs b/gateways/kafka/src/bridge/config.rs new file mode 100644 index 0000000000..7bf13af587 --- /dev/null +++ b/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 { + 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(_)))); + } +} diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs new file mode 100644 index 0000000000..d830b47786 --- /dev/null +++ b/gateways/kafka/src/bridge/error.rs @@ -0,0 +1,137 @@ +// 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_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION, +}; + +/// Kafka's generic `UNKNOWN_SERVER_ERROR` (`-1`). Not in `protocol::api`'s `ERROR_*` set - that +/// table only lists codes the foundation's stub responses actually send; this is the bridge's own +/// catch-all for an `IggyError` variant with no closer Kafka analogue. +const ERROR_UNKNOWN_SERVER_ERROR: i16 = -1; +/// Kafka's `TOPIC_AUTHORIZATION_FAILED`. Closest fit for an Iggy permission/credential rejection - +/// there is no bridge-side SASL exchange yet (`#3549`), so `SASL_AUTHENTICATION_FAILED` would +/// misstate the failure point. +const ERROR_TOPIC_AUTHORIZATION_FAILED: i16 = 29; + +/// 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, + }, +} + +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::InvalidConfig(_) => ERROR_UNSUPPORTED_VERSION, + } + } +} + +/// `InvalidConfig` maps to `ERROR_UNSUPPORTED_VERSION` only because no closer code exists in the +/// foundation's table for "this gateway is misconfigured" - it is never actually sent for a +/// version mismatch. Kept private so that association can change without touching call sites. +const fn iggy_error_to_kafka_code(err: &IggyError) -> i16 { + match err { + IggyError::StreamIdNotFound(_) | IggyError::TopicIdNotFound(_, _) => { + ERROR_UNKNOWN_TOPIC_OR_PARTITION + } + IggyError::Unauthenticated + | IggyError::InvalidCredentials + | IggyError::InvalidUsername + | IggyError::InvalidPassword => ERROR_TOPIC_AUTHORIZATION_FAILED, + IggyError::Disconnected | IggyError::CannotEstablishConnection => { + ERROR_NOT_LEADER_OR_FOLLOWER + } + _ => ERROR_UNKNOWN_SERVER_ERROR, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use iggy::prelude::Identifier; + + #[test] + fn stream_not_found_maps_to_unknown_topic_or_partition() { + let err = BridgeError::Iggy(IggyError::StreamIdNotFound(Identifier::numeric(1).unwrap())); + assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_TOPIC_OR_PARTITION); + } + + #[test] + fn unauthenticated_maps_to_topic_authorization_failed() { + let err = BridgeError::Iggy(IggyError::Unauthenticated); + assert_eq!(err.to_kafka_error_code(), ERROR_TOPIC_AUTHORIZATION_FAILED); + } + + #[test] + fn disconnected_maps_to_not_leader_or_follower_for_retry() { + let err = BridgeError::Iggy(IggyError::Disconnected); + assert_eq!(err.to_kafka_error_code(), ERROR_NOT_LEADER_OR_FOLLOWER); + } + + #[test] + fn cannot_establish_connection_maps_to_not_leader_or_follower_for_retry() { + let err = BridgeError::Iggy(IggyError::CannotEstablishConnection); + assert_eq!(err.to_kafka_error_code(), ERROR_NOT_LEADER_OR_FOLLOWER); + } + + #[test] + fn unmatched_iggy_error_falls_back_to_unknown_server_error() { + let err = BridgeError::Iggy(IggyError::InvalidConfiguration); + assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_SERVER_ERROR); + } + + #[test] + fn partition_out_of_range_maps_to_unknown_topic_or_partition() { + let err = BridgeError::PartitionOutOfRange { + topic: "t".to_string(), + partition: 5, + partitions_count: 2, + }; + assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_TOPIC_OR_PARTITION); + } +} diff --git a/gateways/kafka/src/bridge/iggy_bridge.rs b/gateways/kafka/src/bridge/iggy_bridge.rs new file mode 100644 index 0000000000..830311e9d0 --- /dev/null +++ b/gateways/kafka/src/bridge/iggy_bridge.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 iggy::prelude::{ + Client, Identifier, IggyClient, IggyClientBuilder, IggyError, StreamClient, TopicClient, + TopicCreateOptions, +}; +use tracing::{debug, info}; + +use crate::bridge::config::IggyBridgeConfig; +use crate::bridge::error::BridgeError; + +/// 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. + /// + /// # Errors + /// + /// Returns [`BridgeError::InvalidConfig`] if `config.address` is empty. Returns + /// [`BridgeError::Iggy`] if the connection string 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 { + if config.address.trim().is_empty() { + return Err(BridgeError::InvalidConfig( + "Iggy address must not be empty".to_string(), + )); + } + + let client = IggyClientBuilder::from_connection_string(&config.connection_string()) + .map_err(BridgeError::Iggy)? + .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: a `get` before each `create` means calling this twice for the same topic is a + /// no-op the second time. 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. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] for any Iggy failure other than the + /// already-exists race described above (auth, connectivity, invalid name). + 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, partition_count) + .await?; + Ok(()) + } + + async fn ensure_stream(&self, stream_name: &str) -> Result { + let identifier = Identifier::try_from(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 - the stream exists now regardless of who created it. + let existing = self + .client + .get_stream(&identifier) + .await + .map_err(BridgeError::Iggy)? + .ok_or(IggyError::StreamIdNotFound(identifier))?; + Identifier::numeric(existing.id).map_err(BridgeError::Iggy) + } + Err(err) => Err(BridgeError::Iggy(err)), + } + } + + async fn ensure_topic( + &self, + stream_id: &Identifier, + topic_name: &str, + partition_count: u32, + ) -> Result<(), BridgeError> { + let identifier = Identifier::try_from(topic_name).map_err(BridgeError::Iggy)?; + if self + .client + .get_topic(stream_id, &identifier) + .await + .map_err(BridgeError::Iggy)? + .is_some() + { + debug!("Iggy topic '{topic_name}' already exists"); + 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(()) + } + // Lost a create race - the topic exists now regardless of who created it. + Err(IggyError::TopicNameAlreadyExists(_, _)) => Ok(()), + Err(err) => Err(BridgeError::Iggy(err)), + } + } + + /// Returns the high watermark (offset of the next message to be written) for one partition of + /// an Iggy topic. + /// + /// Maps directly from `Partition::current_offset` - Iggy partitions start a fresh partition's + /// counter at `0` and advance it past each written message's own offset, the same "next free + /// offset" convention Kafka's high watermark uses; there is no unit conversion needed for + /// `ListOffsets` (`#3537`) to build on this. + /// + /// # Errors + /// + /// Returns [`BridgeError::Iggy`] if the stream/topic doesn't exist. Returns + /// [`BridgeError::PartitionOutOfRange`] if `partition` is beyond the topic's partition count. + pub async fn high_watermark( + &self, + stream: &str, + topic: &str, + partition: u32, + ) -> Result { + let stream_id = Identifier::try_from(stream).map_err(BridgeError::Iggy)?; + let topic_id = Identifier::try_from(topic).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::TopicIdNotFound(topic_id, stream_id)))?; + + details + .partitions + .iter() + .find(|p| p.id == partition) + .map(|p| p.current_offset) + .ok_or(BridgeError::PartitionOutOfRange { + topic: topic.to_string(), + partition, + partitions_count: details.partitions_count, + }) + } +} diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs new file mode 100644 index 0000000000..af29dbb01a --- /dev/null +++ b/gateways/kafka/src/bridge/mod.rs @@ -0,0 +1,32 @@ +// 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. + +//! Iggy SDK integration layer (`#3533`). +//! +//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning and +//! high-watermark lookups, and translates Iggy errors to Kafka wire error codes. Not yet wired +//! into the live Produce/Fetch dispatch path - that lands with `#3535`/`#3536`. + +pub mod config; +pub mod error; +pub mod iggy_bridge; +pub mod topic_map; + +pub use config::IggyBridgeConfig; +pub use error::BridgeError; +pub use iggy_bridge::IggyBridge; +pub use topic_map::{TopicMapping, TopicOverride}; diff --git a/gateways/kafka/src/bridge/topic_map.rs b/gateways/kafka/src/bridge/topic_map.rs new file mode 100644 index 0000000000..2cb11060db --- /dev/null +++ b/gateways/kafka/src/bridge/topic_map.rs @@ -0,0 +1,166 @@ +// 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, +} + +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( + || (self.default_stream.clone(), kafka_topic.to_string()), + |over| (over.stream.clone(), over.topic.clone()), + ) + } + + /// Parses a `TopicMapping` from a TOML document. + /// + /// # Errors + /// + /// Returns [`BridgeError::InvalidConfig`] if `raw` is not valid TOML for this shape, or if + /// `default_stream` is empty (every unmapped Kafka topic would otherwise resolve to an empty + /// stream name, which the Iggy SDK rejects only once a request is actually made). + pub fn from_toml_str(raw: &str) -> Result { + let mapping: Self = toml::from_str(raw) + .map_err(|e| BridgeError::InvalidConfig(format!("invalid topic mapping TOML: {e}")))?; + if mapping.default_stream.trim().is_empty() { + return Err(BridgeError::InvalidConfig( + "topic mapping's default_stream must not be empty".to_string(), + )); + } + Ok(mapping) + } + + /// Reads and parses a `TopicMapping` TOML file. + /// + /// # Errors + /// + /// Returns [`BridgeError::InvalidConfig`] if the file cannot be read, or on the same + /// conditions as [`from_toml_str`](Self::from_toml_str). + pub fn from_file(path: &Path) -> Result { + let raw = std::fs::read_to_string(path).map_err(|e| { + BridgeError::InvalidConfig(format!( + "failed to read topic mapping file '{}': {e}", + path.display() + )) + })?; + Self::from_toml_str(&raw) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn given_no_override_should_resolve_to_default_stream_and_same_topic_name() { + let mapping = TopicMapping { + default_stream: "kafka".to_string(), + topics: HashMap::new(), + }; + assert_eq!( + mapping.resolve("orders"), + ("kafka".to_string(), "orders".to_string()) + ); + } + + #[test] + fn given_override_should_resolve_to_mapped_stream_and_topic() { + let mut topics = HashMap::new(); + topics.insert( + "orders".to_string(), + TopicOverride { + stream: "billing".to_string(), + topic: "orders_v2".to_string(), + }, + ); + let mapping = TopicMapping { + default_stream: "kafka".to_string(), + topics, + }; + assert_eq!( + mapping.resolve("orders"), + ("billing".to_string(), "orders_v2".to_string()) + ); + assert_eq!( + mapping.resolve("payments"), + ("kafka".to_string(), "payments".to_string()) + ); + } + + #[test] + fn from_toml_str_parses_default_stream_and_overrides() { + let toml = r#" + default_stream = "kafka" + + [topics.orders] + stream = "billing" + topic = "orders_v2" + "#; + let mapping = TopicMapping::from_toml_str(toml).unwrap(); + assert_eq!(mapping.default_stream, "kafka"); + assert_eq!( + mapping.resolve("orders"), + ("billing".to_string(), "orders_v2".to_string()) + ); + } + + #[test] + fn from_toml_str_rejects_empty_default_stream() { + let toml = r#"default_stream = """#; + let err = TopicMapping::from_toml_str(toml).unwrap_err(); + assert!(matches!(err, BridgeError::InvalidConfig(_))); + } + + #[test] + fn from_toml_str_rejects_malformed_toml() { + let err = TopicMapping::from_toml_str("not valid toml {{{").unwrap_err(); + assert!(matches!(err, BridgeError::InvalidConfig(_))); + } + + #[test] + fn from_file_rejects_missing_file() { + let err = TopicMapping::from_file(Path::new("/nonexistent/topic_map.toml")).unwrap_err(); + assert!(matches!(err, BridgeError::InvalidConfig(_))); + } +} diff --git a/gateways/kafka/src/lib.rs b/gateways/kafka/src/lib.rs index ce77d86c25..5ccd935ae1 100644 --- a/gateways/kafka/src/lib.rs +++ b/gateways/kafka/src/lib.rs @@ -17,6 +17,7 @@ //! Kafka wire protocol gateway foundation for Apache Iggy. +pub mod bridge; pub mod error; pub mod protocol; pub mod server; diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs index 9886d9fd53..e9c2f0cbf8 100644 --- a/gateways/kafka/src/main.rs +++ b/gateways/kafka/src/main.rs @@ -59,6 +59,11 @@ async fn main() -> Result<(), Box> { /// away the central provider's typo-detection for this whole namespace - a misspelled key here /// would otherwise silently no-op instead of surfacing anywhere. `reject_unknown_kafka_env_vars` /// is this crate's own replacement for that lost check. +/// +/// Includes `bridge::config::IggyBridgeConfig::KNOWN_ENV_VARS` even though nothing in `main` +/// reads them yet (`IggyBridge` isn't wired into the live dispatch path until `#3535`/`#3536`) - +/// a user who exports them ahead of that wiring landing must not see a spurious "unknown env var" +/// rejection for a name this crate already recognizes. const KNOWN_KAFKA_ENV_VARS: &[&str] = &[ "IGGY_KAFKA_BIND_ADDR", "IGGY_KAFKA_ADVERTISED_HOST", @@ -69,6 +74,11 @@ const KNOWN_KAFKA_ENV_VARS: &[&str] = &[ "IGGY_KAFKA_READ_TIMEOUT_SECS", "IGGY_KAFKA_WRITE_TIMEOUT_SECS", "IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS", + "IGGY_KAFKA_IGGY_ADDR", + "IGGY_KAFKA_IGGY_USERNAME", + "IGGY_KAFKA_IGGY_PASSWORD", + "IGGY_KAFKA_IGGY_STREAM", + "IGGY_KAFKA_TOPIC_MAP_PATH", ]; /// Rejects any `IGGY_KAFKA_*` env var not in [`KNOWN_KAFKA_ENV_VARS`] - a typo (e.g. diff --git a/gateways/kafka/tests/bridge_iggy_integration_tests.rs b/gateways/kafka/tests/bridge_iggy_integration_tests.rs new file mode 100644 index 0000000000..259e5dbaae --- /dev/null +++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs @@ -0,0 +1,227 @@ +// 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::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command}; +use std::sync::OnceLock; +use std::time::Duration; + +use secrecy::SecretString; +use serial_test::serial; + +use iggy_gateway_kafka::bridge::{BridgeError, IggyBridge, IggyBridgeConfig, TopicMapping}; + +/// Picks a free TCP port by binding to `127.0.0.1:0` and immediately releasing it. Small TOCTOU +/// window between release and `iggy-server` binding the same port - acceptable for a test helper, +/// same tradeoff `core/integration`'s own `port_reserver.rs` makes. +fn free_port() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + listener.local_addr().expect("local addr").port() +} + +/// 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. +fn iggy_server_binary() -> &'static PathBuf { + static BINARY_PATH: OnceLock = OnceLock::new(); + BINARY_PATH.get_or_init(|| { + let status = Command::new(env!("CARGO")) + .args(["build", "--package", "server", "--bin", "iggy-server"]) + .status() + .expect("run cargo build for iggy-server"); + assert!( + status.success(), + "cargo build --package server --bin iggy-server failed" + ); + + // CARGO_MANIFEST_DIR is gateways/kafka; the workspace root (and its target/ dir) is two + // levels up. + let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR")); + let workspace_root = manifest_dir + .parent() + .and_then(Path::parent) + .expect("gateways/kafka is two levels under the workspace root"); + workspace_root.join("target/debug/iggy-server") + }) +} + +struct TestServer { + child: Child, + address: String, +} + +impl TestServer { + /// Spawns `iggy-server` with an isolated temp data dir and an ephemeral 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 = free_port(); + let address = format!("127.0.0.1:{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") + // `--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 }; + server.wait_ready().await; + server + } + + /// Retries a full bridge connect (not just a TCP connect) so the wait covers the server + /// actually being ready to authenticate, not just its listener socket being open. + async fn wait_ready(&self) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if IggyBridge::connect(self.test_config()).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("iggy"), + topic_mapping: TopicMapping { + default_stream: "kafka".to_string(), + topics: HashMap::new(), + }, + } + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +#[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"); +} + +#[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 checking the watermark"); + + let watermark = bridge + .high_watermark("kafka", "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" + ); +} + +#[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("kafka", "orders", 5) + .await + .expect_err("partition 5 does not exist on a 1-partition topic"); + assert!(matches!(err, BridgeError::PartitionOutOfRange { .. })); +} + +/// Acceptance criterion: "no panics on Iggy unreachable at handler boundary." Connects to a port +/// nothing is listening on and asserts a plain `Err`, not a panic - the strongest way to fail this +/// assertion is exactly the failure mode being guarded against. +#[tokio::test] +async fn connect_to_unreachable_iggy_returns_err_not_panic() { + let port = free_port(); // reserved, then immediately released, nothing binds it + let config = IggyBridgeConfig { + address: format!("127.0.0.1:{port}"), + username: "iggy".to_string(), + password: SecretString::from("iggy"), + topic_mapping: TopicMapping { + default_stream: "kafka".to_string(), + topics: HashMap::new(), + }, + }; + + let result = IggyBridge::connect(config).await; + assert!(matches!(result, Err(BridgeError::Iggy(_)))); +} From 37aae647e3301a5d1728d16abd2ad75e36279721 Mon Sep 17 00:00:00 2001 From: ryerraguntla Date: Sat, 5 Sep 2026 00:30:04 -0400 Subject: [PATCH 2/2] Update config.rs --- gateways/kafka/src/bridge/config.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gateways/kafka/src/bridge/config.rs b/gateways/kafka/src/bridge/config.rs index 7bf13af587..9c20c550d4 100644 --- a/gateways/kafka/src/bridge/config.rs +++ b/gateways/kafka/src/bridge/config.rs @@ -91,7 +91,7 @@ impl IggyBridgeConfig { /// 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 + /// 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