krishvishal commented on code in PR #4043: URL: https://github.com/apache/iggy/pull/4043#discussion_r3954378584
########## 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( + || (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, if + /// `default_stream` is empty, or if any `[topics.*]` override has an empty `stream`/`topic` - + /// in every case, an empty Iggy stream/topic name would otherwise pass config loading + /// cleanly and only fail much later, deep in `IggyBridge::ensure_stream`/`ensure_topic`, as + /// an opaque `Identifier::named("")` error with no link back to the config entry at fault. + pub fn from_toml_str(raw: &str) -> Result<Self, BridgeError> { + 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() { Review Comment: Validation trims but stores untrimmed (also `config.rs:86`). `default_stream = " kafka "` creates a stream named `" kafka "`. Adding `.trim()` later would silently repoint a live deployment, so reject instead, before this ships. Same code checks `Identifier::named`'s empty case but not its 255-byte case (`identifier/mod.rs:169-171`) - the other half of one condition. ########## 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(_) => { Review Comment: The create path throws away `TopicDetails` and skips the partition-count check done on the other two branches. `topics.rs:97` passes `Some(0)` straight through; `rewrite.rs:218` only bounds the top end. ########## gateways/kafka/src/main.rs: ########## @@ -71,13 +76,20 @@ const KNOWN_KAFKA_ENV_VARS: &[&str] = &[ "IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS", ]; -/// Rejects any `IGGY_KAFKA_*` env var not in [`KNOWN_KAFKA_ENV_VARS`] - a typo (e.g. -/// `IGGY_KAFKA_BIN_ADDR`) would otherwise be silently ignored: `core/configs`' central provider -/// skips the whole `IGGY_KAFKA_` prefix, and this crate's own `env_var()` only ever looks up -/// exact known names, so nothing reads the misspelled var and nothing warns either. +/// Rejects any `IGGY_KAFKA_*` env var not in [`KNOWN_KAFKA_ENV_VARS`] or +/// [`IggyBridgeConfig::KNOWN_ENV_VARS`] - a typo (e.g. `IGGY_KAFKA_BIN_ADDR`) would otherwise be +/// silently ignored: `core/configs`' central provider skips the whole `IGGY_KAFKA_` prefix, and +/// this crate's own `env_var()` only ever looks up exact known names, so nothing reads the +/// misspelled var and nothing warns either. Checking both lists (not `IggyBridgeConfig`'s +/// wiring into `main`) is deliberate: a user who exports a bridge var ahead of +/// `#3535`/`#3536` landing must not see a spurious "unknown env var" rejection for a name this +/// crate already recognizes. fn reject_unknown_kafka_env_vars() -> Result<(), String> { for (key, _) in std::env::vars() { - if key.starts_with("IGGY_KAFKA_") && !KNOWN_KAFKA_ENV_VARS.contains(&key.as_str()) { + if key.starts_with("IGGY_KAFKA_") + && !KNOWN_KAFKA_ENV_VARS.contains(&key.as_str()) + && !IggyBridgeConfig::KNOWN_ENV_VARS.contains(&key.as_str()) Review Comment: Five bridge env vars are allowlisted at startup and read by nothing. The guard's own message says they'd "otherwise be silently ignored", which is what it now permits. One `warn!` fixes it. The anti-desync note at `:64-66` is also meaningless, since checking two lists separately is the same as checking their union. ########## 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 Review Comment: `high_watermark` decodes and sorts the whole partition vector for 16 bytes: at N=1000 that's about 40 KB on the wire and 120 KB allocated per call, plus an O(N) scan of a vector `wire_conversions.rs:174` already sorted. `binary_search_by_key` is free. ########## 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( Review Comment: No caching: 2 round trips steady-state, 4 cold, plus 1 for `high_watermark`, on a path #3535 wants to call per Produce, through one lockstep TCP connection (`tcp_client.rs:1428-1430`). ########## 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; Review Comment: `PortGuard`'s "no TOCTOU window" claim is wrong. flock is advisory, and unlike `port_reserver.rs:115-130` it never reads `/proc/sys/net/ipv4/ip_local_port_range`, so `[15000,15199]` lands inside the ephemeral range on a box tuned `1024 65535`. Sibling test binaries all `bind("127.0.0.1:0")`. ########## 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), Review Comment: `BridgeError::Iggy`'s Display prints empty payloads: `from_repr` fills variants with `Default::default()` and `Identifier::default()` is numeric 0, so you get "for stream with ID: 0" - a made-up id that's also a real one. ########## 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) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if tokio::net::TcpStream::connect(self.address.as_str()) + .await + .is_ok() + { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "iggy-server at {} did not become ready within the startup budget", + self.address + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + fn test_config(&self) -> IggyBridgeConfig { + IggyBridgeConfig { + address: self.address.clone(), + username: "iggy".to_string(), + password: SecretString::from("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(); + } +} + +/// Builds and connects a raw `IggyClient` against `server` - for producing test data directly, +/// independent of the `IggyBridge` under test. Uses the fluent builder, not a hand-built +/// `iggy://user:pass@host` string: `ConnectionString` splits on `@` then `:`, which breaks for +/// any password containing either character. +async fn raw_client(server: &TestServer) -> IggyClient { + let client = IggyClientBuilder::new() + .with_tcp() + .with_server_address(server.address.clone()) + .with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword( + "iggy".to_string(), + SecretString::from("iggy"), + ))) + .build() + .expect("build raw test client"); + client.connect().await.expect("connect raw test client"); + client +} + +#[tokio::test] +#[serial] +async fn ensure_stream_and_topic_is_idempotent_on_repeated_calls() { Review Comment: Coverage gaps: nothing hits `PartitionCountMismatch` on either branch; no test uses a `TopicMapping` override through `IggyBridge` (all use `HashMap::new()`), so the deliberate `kafka_topic` vs `topic_name` choice at `:190,:294` can't fail; `from_env`'s no-file branch and all four defaults never run; and this test checks three `Ok`s while reading back nothing. ########## gateways/kafka/Cargo.toml: ########## @@ -34,11 +34,14 @@ path = "src/main.rs" [dependencies] bytes = { workspace = true } +iggy = { workspace = true } Review Comment: Adding `iggy` makes this crate a reverse-dependency of the SDK, so every `core/sdk/**` PR will run these server-spawning tests from now on. Not this PR - `Cargo.lock` is `infrastructure` in `.config/rail.toml`, forcing full-workspace mode. ########## 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, Review Comment: **CRITICAL** (same match as the `_` arm below, opposite direction). `Unauthenticated` maps to 29, which is `retriable=false`. But `core/common/src/traits/binary_auth.rs:27` shows `ClientState::Connected => Err(IggyError::Unauthenticated)`, i.e. it is the window between the socket coming up and auto-sign-in finishing, not an authorization decision. The SDK itself lists it as retriable at `tcp_client.rs:219`. So a brief reconnect kills a consumer thread with "Not authorized to access topics", sending whoever is on call to audit Kafka ACLs that don't exist in this gateway. The same code also fires for a wrong `IGGY_KAFKA_IGGY_PASSWORD`, which is the gateway's own config fault reported as the client's ACL problem. Suggested: move `Unauthenticated` to `ERROR_NOT_LEADER_OR_FOLLOWER` (6) and keep 29 for `Unauthorized` only. `unauthenticated_maps_to_topic_authorization_failed` asserts the current mapping, so it needs updating alongside. -- 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]
