You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/10/27 07:15:33 UTC
(iggy) branch iggy-address-list created (now 52449c21a)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch iggy-address-list
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 52449c21a feat(sdk): accept a list of server addresses for TCP clients
This branch includes the following new commits:
new 52449c21a feat(sdk): accept a list of server addresses for TCP clients
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(sdk): accept a list of server addresses for TCP clients
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch iggy-address-list
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 52449c21af0540b00be007abd7050dd17228efe4
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Mon Oct 26 14:31:49 2026 +0530
feat(sdk): accept a list of server addresses for TCP clients
A new TCP client reached a cluster only through its one configured node. It now tries each address of a comma list, then the nodes of the last roster.
---
.../configuration/auth_config/connection_string.rs | 66 +++++++++--
.../auth_config/connection_string_options.rs | 4 +
.../configuration/tcp_config/tcp_client_config.rs | 14 ++-
.../tcp_config/tcp_client_config_builder.rs | 35 +++++-
.../tcp_config/tcp_connection_string_options.rs | 2 +
core/sdk/README.md | 2 +-
core/sdk/src/driver.rs | 6 +-
core/sdk/src/links/fake_cluster.rs | 3 +-
core/sdk/src/links/replicas.rs | 123 ++++++++++++++++++---
core/sdk/src/quic/quic_client.rs | 2 +-
core/sdk/src/tcp/tcp_client.rs | 8 +-
core/sdk/src/websocket/websocket_client.rs | 2 +-
gateways/kafka/README.md | 7 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 44 ++++++--
15 files changed, 270 insertions(+), 50 deletions(-)
diff --git a/core/common/src/types/configuration/auth_config/connection_string.rs b/core/common/src/types/configuration/auth_config/connection_string.rs
index d1bdac1b7..72c4b149c 100644
--- a/core/common/src/types/configuration/auth_config/connection_string.rs
+++ b/core/common/src/types/configuration/auth_config/connection_string.rs
@@ -82,16 +82,10 @@ impl<T: ConnectionStringOptions + Default> ConnectionString<T> {
}
let server_address = server_and_options[0];
- if server_address.starts_with('[') {
- validate_server_address(server_address)
- .map_err(|_| IggyError::InvalidConnectionString)?;
+ if T::ADDRESS_LIST {
+ server_address.split(',').try_for_each(validate_address)?;
} else {
- let (host, port) = server_address
- .split_once(':')
- .ok_or(IggyError::InvalidConnectionString)?;
- if host.is_empty() || !port.parse::<u16>().is_ok_and(|port| port != 0) {
- return Err(IggyError::InvalidConnectionString);
- }
+ validate_address(server_address)?;
}
let connection_string_options;
@@ -129,6 +123,21 @@ impl<T: ConnectionStringOptions + Default> FromStr for ConnectionString<T> {
}
}
+/// Checks one `host:port` or `[ipv6]:port` of a connection string.
+fn validate_address(server_address: &str) -> Result<(), IggyError> {
+ if server_address.starts_with('[') {
+ return validate_server_address(server_address)
+ .map_err(|_| IggyError::InvalidConnectionString);
+ }
+ let (host, port) = server_address
+ .split_once(':')
+ .ok_or(IggyError::InvalidConnectionString)?;
+ if host.is_empty() || !port.parse::<u16>().is_ok_and(|port| port != 0) {
+ return Err(IggyError::InvalidConnectionString);
+ }
+ Ok(())
+}
+
/// ConnectionStringUtils is a utility struct for connection strings.
pub struct ConnectionStringUtils;
@@ -180,6 +189,45 @@ mod tests {
}
}
+ #[test]
+ fn should_succeed_with_a_list_of_tcp_addresses() {
+ let value = "iggy://user:secret@10.0.0.1:8090,[2001:db8::1]:8090,node-3:8090";
+ let connection_string = ConnectionString::<TcpConnectionStringOptions>::new(value).unwrap();
+ let config = TcpClientConfig::from(connection_string);
+ assert_eq!(
+ config.server_addresses(),
+ ["10.0.0.1:8090", "[2001:db8::1]:8090", "node-3:8090"]
+ );
+ }
+
+ #[test]
+ fn should_fail_with_a_list_that_holds_a_bad_address() {
+ for addresses in [
+ "10.0.0.1:8090,",
+ "10.0.0.1:8090,node-3",
+ "10.0.0.1:8090,,node-3:8090",
+ ] {
+ let value = format!("iggy://user:secret@{addresses}");
+ assert!(
+ matches!(
+ ConnectionString::<TcpConnectionStringOptions>::new(&value),
+ Err(IggyError::InvalidConnectionString)
+ ),
+ "{addresses}"
+ );
+ }
+ }
+
+ /// Only TCP clients dial a list.
+ #[test]
+ fn should_fail_with_a_list_of_quic_addresses() {
+ let value = "iggy+quic://user:secret@10.0.0.1:8080,10.0.0.2:8080";
+ assert!(matches!(
+ ConnectionString::<QuicConnectionStringOptions>::new(value),
+ Err(IggyError::InvalidConnectionString)
+ ));
+ }
+
#[test]
fn should_succeed_with_ipv6_address() {
for server_address in [
diff --git a/core/common/src/types/configuration/auth_config/connection_string_options.rs b/core/common/src/types/configuration/auth_config/connection_string_options.rs
index 778a76e5e..43f8e8751 100644
--- a/core/common/src/types/configuration/auth_config/connection_string_options.rs
+++ b/core/common/src/types/configuration/auth_config/connection_string_options.rs
@@ -18,6 +18,10 @@
use crate::{IggyError, NonZeroIggyDuration};
pub trait ConnectionStringOptions {
+ /// Whether the address of a connection string can list several nodes of one cluster,
+ /// separated by commas.
+ const ADDRESS_LIST: bool = false;
+
fn retries(&self) -> Option<u32>;
fn heartbeat_interval(&self) -> NonZeroIggyDuration;
diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
index f91f0eea1..8ba1359d3 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
@@ -24,7 +24,9 @@ use std::str::FromStr;
/// Configuration for the TCP client.
#[derive(Debug, Clone)]
pub struct TcpClientConfig {
- /// The address of the Iggy server.
+ /// The address of the Iggy server. To reach a cluster while one of its nodes is down, list
+ /// the addresses of several nodes, separated by commas. A new client tries each address in
+ /// order, and then the nodes of the last roster that it read.
pub server_address: String,
/// Whether to use TLS when connecting to the server.
pub tls_enabled: bool,
@@ -64,6 +66,16 @@ impl Default for TcpClientConfig {
}
}
+impl TcpClientConfig {
+ /// Each address of `server_address`, in order.
+ pub fn server_addresses(&self) -> Vec<String> {
+ self.server_address
+ .split(',')
+ .map(|address| address.trim().to_owned())
+ .collect()
+ }
+}
+
impl From<ConnectionString<TcpConnectionStringOptions>> for TcpClientConfig {
fn from(connection_string: ConnectionString<TcpConnectionStringOptions>) -> Self {
TcpClientConfig {
diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
index 7b063bb85..6052a0144 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
@@ -38,7 +38,8 @@ impl TcpClientConfigBuilder {
TcpClientConfigBuilder::default()
}
- /// Sets the server address for the TCP client.
+ /// Sets the server address for the TCP client, or a comma-separated list of the addresses of
+ /// several nodes of one cluster.
pub fn with_server_address(mut self, server_address: String) -> Self {
self.config.server_address = server_address;
self
@@ -112,8 +113,11 @@ impl TcpClientConfigBuilder {
/// Builds the TCP client configuration.
pub fn build(mut self) -> Result<TcpClientConfig, IggyError> {
- self.config.server_address = self.config.server_address.trim().to_owned();
- validate_server_address(&self.config.server_address)?;
+ let addresses = self.config.server_addresses();
+ for address in &addresses {
+ validate_server_address(address)?;
+ }
+ self.config.server_address = addresses.join(",");
Ok(self.config)
}
@@ -151,6 +155,31 @@ mod tests {
));
}
+ #[test]
+ fn list_of_addresses_should_succeed_with_each_trimmed() {
+ let config = builder_with_address(" 10.0.0.1:8090 , [::1]:8090,node-3:8090 ")
+ .build()
+ .unwrap();
+ assert_eq!(
+ config.server_address,
+ "10.0.0.1:8090,[::1]:8090,node-3:8090"
+ );
+ assert_eq!(
+ config.server_addresses(),
+ ["10.0.0.1:8090", "[::1]:8090", "node-3:8090"]
+ );
+ }
+
+ #[test]
+ fn list_with_an_empty_address_should_fail() {
+ for addresses in ["10.0.0.1:8090,,node-3:8090", "10.0.0.1:8090,"] {
+ assert!(
+ builder_with_address(addresses).build().is_err(),
+ "{addresses}"
+ );
+ }
+ }
+
#[test]
fn valid_dns_name_should_succeed() {
let builder = builder_with_address("localhost:8080");
diff --git a/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
index f6e19b866..85a1c514d 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs
@@ -59,6 +59,8 @@ impl TcpConnectionStringOptions {
}
impl ConnectionStringOptions for TcpConnectionStringOptions {
+ const ADDRESS_LIST: bool = true;
+
fn retries(&self) -> Option<u32> {
self.reconnection.max_retries
}
diff --git a/core/sdk/README.md b/core/sdk/README.md
index 2aa916385..bfbaad82f 100644
--- a/core/sdk/README.md
+++ b/core/sdk/README.md
@@ -27,7 +27,7 @@ Official Rust client SDK for [Apache Iggy](https://iggy.apache.org), the persist
- **Transports**: TCP (custom binary), QUIC, HTTP, WebSocket. One unified `IggyClient` API across all four.
- **TLS**: TCP and WebSocket expose TLS connection-string options; QUIC always uses TLS; HTTP uses an HTTPS URL configured through the builder.
-- **Connection strings**: `iggy://` (TCP default), `iggy+tcp://`, `iggy+quic://`, `iggy+http://`, `iggy+ws://`. Binary transports apply credentials on `connect()`; HTTP requires an explicit login. Option keys and reconnection support differ by transport.
+- **Connection strings**: `iggy://` (TCP default), `iggy+tcp://`, `iggy+quic://`, `iggy+http://`, `iggy+ws://`. Binary transports apply credentials on `connect()`; HTTP requires an explicit login. Option keys and reconnection support differ by transport. A TCP client can list several nodes of one cluster, separated by commas, as in `iggy://iggy:iggy@node-1:8090,node-2:8090`: a new client tries each in order, then the nodes of the last roster that it read.
- **Authentication**: username/password and Personal Access Tokens (PAT).
- **Async, non-blocking** client built on Tokio with custom zero-copy (de)serialization.
- **High-level builders** on `IggyClient`: `producer(stream, topic)`, `consumer(name, stream, topic, partition)`, and `consumer_group(name, stream, topic)`.
diff --git a/core/sdk/src/driver.rs b/core/sdk/src/driver.rs
index 3a61faa67..b37b7dc6f 100644
--- a/core/sdk/src/driver.rs
+++ b/core/sdk/src/driver.rs
@@ -2249,7 +2249,7 @@ mod tests {
));
fixture.driver = OperationDriver::new(ReplicaLinks::new(
fixture.cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
session,
));
@@ -3378,7 +3378,7 @@ mod tests {
) -> OperationDriver {
let session = Arc::new(SessionOwner::new(auto_login));
session.remember(credentials, USER_ID);
- let links = ReplicaLinks::new(cluster.dialer(), endpoint(0), policy(), session);
+ let links = ReplicaLinks::new(cluster.dialer(), vec![endpoint(0)], policy(), session);
links.connect().await.unwrap();
eventually("every replica is bound", || {
links.ready_replicas().len() == usize::from(REPLICAS)
@@ -3636,7 +3636,7 @@ mod tests {
session.bind(FIRST_SESSION).unwrap();
f.driver = OperationDriver::new(ReplicaLinks::new(
f.cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
session,
));
diff --git a/core/sdk/src/links/fake_cluster.rs b/core/sdk/src/links/fake_cluster.rs
index 1358b3148..466fa5d15 100644
--- a/core/sdk/src/links/fake_cluster.rs
+++ b/core/sdk/src/links/fake_cluster.rs
@@ -525,7 +525,8 @@ pub(crate) fn links(
) -> ReplicaLinks {
let credentials = Credentials::UsernamePassword("iggy".to_owned(), SecretString::from("iggy"));
let session = Arc::new(SessionOwner::new(AutoLogin::Enabled(credentials)));
- ReplicaLinks::new(cluster.dialer(), configured.to_owned(), policy, session)
+ let configured = configured.split(',').map(str::to_owned).collect();
+ ReplicaLinks::new(cluster.dialer(), configured, policy, session)
}
/// Connects, and waits until `count` slots serve the session.
diff --git a/core/sdk/src/links/replicas.rs b/core/sdk/src/links/replicas.rs
index aebe4608f..e8c8a7d3c 100644
--- a/core/sdk/src/links/replicas.rs
+++ b/core/sdk/src/links/replicas.rs
@@ -141,7 +141,9 @@ pub(crate) struct ReplicaLinks {
#[derive(Debug)]
struct LinksShared {
dialer: Dialer,
- configured_endpoint: String,
+ /// The endpoints of the client configuration, in order. A sweep dials
+ /// them before the roster.
+ configured_endpoints: Vec<String>,
policy: ReconnectPolicy,
session: Arc<SessionOwner>,
/// Never held across an await. The epoch of the session is read before
@@ -168,7 +170,7 @@ struct Table {
bootstrap: Option<Bootstrap>,
metadata_primary: Option<u8>,
/// The endpoints of the last roster. A sweep dials them after the
- /// configured endpoint.
+ /// configured endpoints.
roster: Vec<String>,
}
@@ -229,7 +231,7 @@ impl From<IggyError> for SignInFailure {
impl ReplicaLinks {
pub(crate) fn new(
dialer: Dialer,
- configured_endpoint: String,
+ configured_endpoints: Vec<String>,
policy: ReconnectPolicy,
session: Arc<SessionOwner>,
) -> Self {
@@ -245,7 +247,7 @@ impl ReplicaLinks {
Self {
shared: Arc::new(LinksShared {
dialer,
- configured_endpoint,
+ configured_endpoints,
policy,
session,
table: Mutex::new(table),
@@ -472,7 +474,7 @@ impl ReplicaLinks {
}
/// The endpoint of the metadata primary, else of the bootstrap link, else
- /// the configured one.
+ /// the first configured one.
pub(crate) fn connection_address(&self) -> String {
let table = self.shared.table();
table
@@ -483,8 +485,9 @@ impl ReplicaLinks {
.bootstrap
.as_ref()
.map(|bootstrap| &bootstrap.endpoint))
- .unwrap_or(&self.shared.configured_endpoint)
- .clone()
+ .or(self.shared.configured_endpoints.first())
+ .cloned()
+ .unwrap_or_default()
}
}
@@ -633,11 +636,11 @@ impl LinksShared {
Ok(())
}
- /// The configured endpoint, then the endpoints of the last roster, each
+ /// The configured endpoints, then the endpoints of the last roster, each
/// spelling once. Spellings need no name lookup, which can stall a sweep.
fn candidates(&self, roster: &[String]) -> Vec<String> {
- let mut candidates = vec![self.configured_endpoint.clone()];
- for endpoint in roster {
+ let mut candidates: Vec<String> = Vec::new();
+ for endpoint in self.configured_endpoints.iter().chain(roster) {
if !candidates
.iter()
.any(|known| is_same_spelling(known, endpoint))
@@ -648,6 +651,22 @@ impl LinksShared {
candidates
}
+ /// The configured endpoint that `link` was dialed at, else the first
+ /// configured one.
+ fn configured_endpoint_of(&self, link: &Arc<Link>) -> String {
+ let dialed = self.table().endpoint_of(link).map(str::to_owned);
+ let configured = &self.configured_endpoints;
+ dialed
+ .and_then(|dialed| {
+ configured
+ .iter()
+ .find(|endpoint| is_same_spelling(endpoint, &dialed))
+ })
+ .or(configured.first())
+ .cloned()
+ .unwrap_or_default()
+ }
+
async fn try_endpoint(
self: &Arc<Self>,
endpoint: &str,
@@ -964,7 +983,7 @@ impl LinksShared {
/// Rebuilds the slots from the roster that `via` carried. A removed
/// replica or a moved endpoint retires its link, so the requests that
/// hold it finish first. The slot of a one-node roster dials the
- /// configured endpoint.
+ /// configured endpoint that `via` was dialed at, else the first one.
async fn apply_roster(
self: &Arc<Self>,
roster: &ClusterMetadata,
@@ -989,12 +1008,12 @@ impl LinksShared {
}
}
// A single node advertises its own view of its address, such as a
- // bind address or a container name. Only the configured endpoint is
+ // bind address or a container name. Only a configured endpoint is
// known to reach it.
if wanted.len() == 1
&& let Some(endpoint) = wanted.values_mut().next()
{
- endpoint.clone_from(&self.configured_endpoint);
+ *endpoint = self.configured_endpoint_of(via);
}
let primary = roster
.nodes
@@ -1330,6 +1349,22 @@ impl Table {
self.links().find(|ready| Arc::ptr_eq(&ready.link, link))
}
+ /// The endpoint that `link` was dialed at, while a slot or the bootstrap
+ /// holds it.
+ fn endpoint_of(&self, link: &Arc<Link>) -> Option<&str> {
+ let holds = |ready: &ReadyLink| Arc::ptr_eq(&ready.link, link);
+ self.slots
+ .values()
+ .find(|slot| slot.link.as_ref().is_some_and(holds))
+ .map(|slot| slot.endpoint.as_str())
+ .or_else(|| {
+ self.bootstrap
+ .as_ref()
+ .filter(|bootstrap| holds(&bootstrap.link))
+ .map(|bootstrap| bootstrap.endpoint.as_str())
+ })
+ }
+
fn find_mut(&mut self, link: &Arc<Link>) -> Option<&mut ReadyLink> {
self.slots
.values_mut()
@@ -1586,6 +1621,66 @@ mod tests {
assert_eq!(links.connection_address(), MAPPED);
}
+ /// The first pass of the sweep dials every configured endpoint.
+ #[tokio::test(start_paused = true)]
+ async fn given_configured_endpoints_when_the_first_is_down_should_connect_through_the_next() {
+ let cluster = FakeCluster::new(3);
+ cluster.down.lock().unwrap().insert(endpoint(0));
+ let links = links(
+ &cluster,
+ &format!("{},{}", endpoint(0), endpoint(1)),
+ one_retry(),
+ );
+
+ links.connect().await.unwrap();
+
+ let dialed: Vec<String> = cluster
+ .dials
+ .lock()
+ .unwrap()
+ .iter()
+ .map(|(endpoint, _)| endpoint.clone())
+ .take(2)
+ .collect();
+ assert_eq!(dialed, [endpoint(0), endpoint(1)]);
+ eventually("the live replicas are ready", || {
+ links.ready_replicas() == [1, 2]
+ })
+ .await;
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_single_node_roster_should_redial_the_configured_endpoint_that_reached_it() {
+ const GONE: &str = "127.0.0.1:18091";
+ const MAPPED: &str = "127.0.0.1:18090";
+ let cluster = FakeCluster::new(1);
+ cluster
+ .down
+ .lock()
+ .unwrap()
+ .extend([endpoint(0), GONE.to_owned()]);
+ let links = links(&cluster, &format!("{GONE},{MAPPED}"), one_retry());
+ links.connect().await.unwrap();
+
+ cluster.drop_connections(MAPPED);
+ eventually(
+ "the slot redials the endpoint that reached the node",
+ || cluster.dials_to(MAPPED).len() == 2 && links.ready_replicas() == [0],
+ )
+ .await;
+ assert_eq!(slot_endpoints(&links), [(0, MAPPED.to_owned())]);
+ assert_eq!(links.connection_address(), MAPPED);
+ }
+
+ /// A policy that gives up after one retry, so a client that cannot
+ /// connect fails the test instead of retrying for ever.
+ fn one_retry() -> ReconnectPolicy {
+ ReconnectPolicy {
+ max_retries: Some(1),
+ ..policy()
+ }
+ }
+
#[tokio::test(start_paused = true)]
async fn given_a_bootstrap_endpoint_outside_the_roster_when_the_primary_slot_is_ready_should_close_the_bootstrap_link()
{
@@ -2034,7 +2129,7 @@ mod tests {
session.remember(credentials.clone(), 1);
let links = ReplicaLinks::new(
cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
Arc::clone(&session),
);
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index d161de71b..8d0e7c204 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -96,7 +96,7 @@ impl QuicClient {
endpoint: OnceCell::new(),
client_config,
},
- config.server_address.clone(),
+ vec![config.server_address.clone()],
ReconnectPolicy::from(&config.reconnection),
session,
);
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 25dd57033..776488d4d 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -92,7 +92,7 @@ impl TcpClient {
));
let links = ReplicaLinks::new(
Dialer::Tcp(Arc::clone(&config)),
- config.server_address.clone(),
+ config.server_addresses(),
ReconnectPolicy::from(&config.reconnection),
session,
);
@@ -156,7 +156,7 @@ mod tests {
let session = Arc::new(SessionOwner::new(AutoLogin::Disabled));
TcpClient::from_links(ReplicaLinks::new(
cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
session,
))
@@ -942,7 +942,7 @@ mod tests {
let session = Arc::new(SessionOwner::new(AutoLogin::Enabled(credentials)));
let client = TcpClient::from_links(ReplicaLinks::new(
cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
session,
));
@@ -987,7 +987,7 @@ mod tests {
let client = || {
TcpClient::from_links(ReplicaLinks::new(
cluster.dialer(),
- endpoint(0),
+ vec![endpoint(0)],
policy(),
Arc::new(SessionOwner::new(AutoLogin::Enabled(credentials.clone()))),
))
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 75ad7302a..fe178b5d5 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -73,7 +73,7 @@ impl WebSocketClient {
));
let links = ReplicaLinks::new(
Dialer::WebSocket(Arc::clone(&config)),
- config.server_address.clone(),
+ vec![config.server_address.clone()],
ReconnectPolicy::from(&config.reconnection),
session,
);
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 692aa6a2c..187b686ac 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -402,7 +402,7 @@ check's and the bridge's alike, and are refused while neither SASL nor the bridg
| Variable | Default | Description |
| --- | --- | --- |
| `IGGY_KAFKA_IGGY_TLS_ENABLED` | `false` | Encrypt the gateway's connections to Iggy (`true` or `false`, nothing else). Required if the Iggy server only accepts TLS, otherwise the bridge fails to connect and every verification fails as unreachable |
-| `IGGY_KAFKA_IGGY_TLS_DOMAIN` | derived from the address | Name checked against the Iggy server certificate |
+| `IGGY_KAFKA_IGGY_TLS_DOMAIN` | derived from each address | Name checked against the certificate of every Iggy node |
| `IGGY_KAFKA_IGGY_TLS_CA_FILE` | SDK bundled roots | PEM roots to trust. Note the SDK does not use the system trust store |
Client side, for example with `kcat`:
@@ -583,7 +583,7 @@ Limits:
| Variable | Default | Description |
| --- | --- | --- |
-| `IGGY_KAFKA_IGGY_ADDR` | `127.0.0.1:8090` | Address of the Iggy server to bridge to |
+| `IGGY_KAFKA_IGGY_ADDR` | `127.0.0.1:8090` | Address of the Iggy server to bridge to and to check SASL logins against. For a cluster, list several nodes, separated by commas: each new Iggy client tries them in order, so new producers and logins go on while one node is down |
| `IGGY_KAFKA_IGGY_USERNAME` | `iggy` | Iggy username |
| `IGGY_KAFKA_IGGY_PASSWORD` | none - **required** | Iggy password. No default: `iggy-server` only uses the well-known `iggy`/`iggy` root credentials when started with `--with-default-root-credentials` (dev-only). Otherwise the operator sets the root password, so a hardcoded default here could never be right and would invite running as root unnoticed |
| `IGGY_KAFKA_IGGY_STREAM` | `kafka` | Default Iggy stream for a Kafka topic with no explicit mapping override |
@@ -596,7 +596,8 @@ The initial connect retries a fixed, bounded number of times (`RECONNECTION_RETR
Iggy SDK client's own default of unlimited retries, one dial per second, forever), and the whole
attempt - retries included - is capped at `REQUEST_TIMEOUT` (15s) wall-clock, so `IggyBridge::connect`
fails in bounded time whether the address refuses the connection or silently drops it, instead of
-blocking the calling task indefinitely. Every other bridge call (`ensure_stream_and_topic`,
+blocking the calling task indefinitely. Each of those tries dials every address of
+`IGGY_KAFKA_IGGY_ADDR` in turn. Every other bridge call (`ensure_stream_and_topic`,
`high_watermark(s)`, `probe`, `close`) carries the same `REQUEST_TIMEOUT` for the same
reason: the SDK reconnects internally, mid-call, on a transport error, through the same
undead-lined dial path - a bridge call made well after the initial connect can still hit this if
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index b32f374b0..b23110852 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -79,7 +79,7 @@ file under `tests/` anymore.
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
| [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, t [...]
-| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. `kafka-cluster cluster-id` prints the same id across a ga [...]
+| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. With a stopped node first in `IGGY_KAFKA_IGGY_ADDR`, kcat [...]
| [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) |
| [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) |
| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index ed3f86941..cbf02c079 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -349,13 +349,18 @@ async fn start_bridged_gateway_with(
topics: &[(&str, u32)],
gateway_config: GatewayConfig,
) -> BridgedGateway {
- start_gateway(server, topics, false, gateway_config).await
+ start_gateway(&server.address, topics, false, gateway_config).await
}
/// A bridge to `server` as the root user, with the default topic mapping.
async fn root_bridge(server: &TestServer) -> IggyBridge {
+ root_bridge_at(&server.address).await
+}
+
+/// [`root_bridge`] for `IGGY_KAFKA_IGGY_ADDR` set to `iggy_address`.
+async fn root_bridge_at(iggy_address: &str) -> IggyBridge {
let config = IggyBridgeConfig {
- address: server.address.clone(),
+ address: iggy_address.to_owned(),
username: ROOT_USER.to_string(),
password: SecretString::from(ROOT_PASSWORD.to_string()),
topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new())
@@ -369,15 +374,16 @@ async fn root_bridge(server: &TestServer) -> IggyBridge {
.expect("the bridge connects to a ready server")
}
-/// A bridged gateway on `gateway_config`, with SASL on when `sasl` is set, creating each
-/// `(topic, partitions)` first. Its listeners are replaced by one on an ephemeral port.
+/// A gateway bridged to `iggy_address` on `gateway_config`, with SASL on when `sasl` is set,
+/// creating each `(topic, partitions)` first. Its listeners are replaced by one on an ephemeral
+/// port.
async fn start_gateway(
- server: &TestServer,
+ iggy_address: &str,
topics: &[(&str, u32)],
sasl: bool,
gateway_config: GatewayConfig,
) -> BridgedGateway {
- let bridge = root_bridge(server).await;
+ let bridge = root_bridge_at(iggy_address).await;
for &(topic, partitions) in topics {
bridge
.ensure_stream_and_topic(topic, partitions)
@@ -395,7 +401,7 @@ async fn start_gateway(
let mut gateway = KafkaGateway::new(gateway_config).with_bridge(Some(Arc::new(bridge)));
if sasl {
gateway =
- gateway.with_authenticator(Arc::new(IggyAuthenticator::new(server.address.clone())));
+ gateway.with_authenticator(Arc::new(IggyAuthenticator::new(iggy_address.to_owned())));
}
let (shutdown, receiver) = broadcast::channel(1);
let task = tokio::spawn(async move {
@@ -719,7 +725,13 @@ async fn bridged_stack() -> Option<(TestServer, SocketAddr)> {
/// serves it.
async fn sasl_bridged_stack() -> Option<(TestServer, SocketAddr)> {
let server = ready_server()?;
- let gateway = start_gateway(&server, &[(TOPIC, 1)], true, GatewayConfig::default()).await;
+ let gateway = start_gateway(
+ &server.address,
+ &[(TOPIC, 1)],
+ true,
+ GatewayConfig::default(),
+ )
+ .await;
// Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange.
std::mem::forget(gateway.shutdown);
Some((server, gateway.addr))
@@ -1392,6 +1404,22 @@ async fn given_sasl_and_the_bridge_when_kcat_produces_and_consumes_should_read_b
assert_read_in_order(&output, &RECORDS);
}
+/// `IGGY_KAFKA_IGGY_ADDR` can list several Iggy nodes. With the first one stopped, the bridge, each
+/// SASL login and each new producer start on the next one.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_a_stopped_first_iggy_node_when_kcat_signs_in_and_produces_should_use_the_next_one() {
+ let Some(server) = ready_server() else {
+ return;
+ };
+ // Nothing listens on a released port, so a connect to it is refused at once.
+ let iggy_address = format!("127.0.0.1:{},{}", free_port(), server.address);
+ let gateway = start_gateway(&iggy_address, &[(TOPIC, 1)], true, GatewayConfig::default()).await;
+ let auth = sasl_plain_args(ROOT_USER, ROOT_PASSWORD);
+ kcat_produce(gateway.addr, TOPIC, 0, &RECORDS, &auth).expect_ran("kcat produce over SASL");
+ let output = kcat_consume(gateway.addr, &auth).expect_ran("kcat consume over SASL");
+ assert_read_in_order(&output, &RECORDS);
+}
+
/// The other half of the same path: a client the gateway refuses must not reach the bridge. kcat
/// exits non-zero here by design, so only its output and what the partition holds are asserted.
///