diff --git a/fluss-rust/crates/fluss/src/client/metadata.rs b/fluss-rust/crates/fluss/src/client/metadata.rs index 20a8a15944..8340fdd52c 100644 --- a/fluss-rust/crates/fluss/src/client/metadata.rs +++ b/fluss-rust/crates/fluss/src/client/metadata.rs @@ -22,7 +22,7 @@ use crate::metadata::{PhysicalTablePath, TableBucket, TablePath}; use crate::proto::MetadataResponse; use crate::rpc::message::UpdateMetadataRequest; use crate::rpc::{RpcClient, ServerConnection}; -use log::info; +use log::{info, warn}; use parking_lot::RwLock; use std::collections::HashSet; use std::net::{SocketAddr, ToSocketAddrs}; @@ -36,6 +36,12 @@ pub struct Metadata { cluster_version_tx: watch::Sender, } +#[derive(Debug)] +struct BootstrapServer { + raw: String, + address: Result, +} + impl Metadata { pub async fn new(bootstrap: &str, connections: Arc) -> Result { let cluster = Self::init_cluster(bootstrap, connections.clone()).await?; @@ -57,12 +63,12 @@ impl Metadata { .send_modify(|v| *v = v.wrapping_add(1)); } - fn parse_bootstrap(boot_strap: &str) -> Result { + fn parse_bootstrap(bootstrap: &str) -> Result { // Resolve all socket addresses and deterministically choose one. - let addrs = boot_strap + let addrs = bootstrap .to_socket_addrs() .map_err(|e| Error::IllegalArgument { - message: format!("Invalid bootstrap address '{boot_strap}': {e}"), + message: format!("Invalid bootstrap address '{bootstrap}': {e}"), })?; // Prefer IPv4 addresses; if none are available, fall back to the first IPv6. @@ -77,20 +83,108 @@ impl Metadata { } let addr = ipv6_candidate.ok_or_else(|| Error::IllegalArgument { - message: format!("Unable to resolve bootstrap address '{boot_strap}'"), + message: format!("Unable to resolve bootstrap address '{bootstrap}'"), })?; Ok(addr) } - async fn init_cluster(boot_strap: &str, connections: Arc) -> Result { - let socket_address = Self::parse_bootstrap(boot_strap)?; - let server_node = ServerNode::new( - -1, - socket_address.ip().to_string(), - socket_address.port() as u32, - ServerType::Unknown, + fn parse_bootstrap_servers(bootstrap_servers: &str) -> Result> { + let mut bootstraps = Vec::new(); + + for bootstrap in bootstrap_servers.split(',') { + let bootstrap = bootstrap.trim(); + if bootstrap.is_empty() { + continue; + } + bootstraps.push(BootstrapServer { + raw: bootstrap.to_string(), + address: Self::parse_bootstrap(bootstrap), + }); + } + + if bootstraps.is_empty() { + return Err(Error::IllegalArgument { + message: "No bootstrap servers configured".to_string(), + }); + } + + Ok(bootstraps) + } + + async fn init_cluster(bootstrap_servers: &str, connections: Arc) -> Result { + let bootstraps = Self::parse_bootstrap_servers(bootstrap_servers)?; + let bootstrap_count = bootstraps.len(); + let mut errors = Vec::new(); + let mut last_connection_error = None; + + for (index, bootstrap) in bootstraps.into_iter().enumerate() { + let socket_address = match bootstrap.address { + Ok(socket_address) => socket_address, + Err(err) => { + if index + 1 < bootstrap_count { + warn!( + "Failed to resolve bootstrap server '{}', trying the next server: {err}", + bootstrap.raw + ); + } + errors.push(format!("{}: {err}", bootstrap.raw)); + continue; + } + }; + let server_node = ServerNode::new( + -1, + socket_address.ip().to_string(), + socket_address.port() as u32, + ServerType::Unknown, + ); + + match Self::fetch_cluster_from_bootstrap(&server_node, connections.clone()).await { + Ok(cluster) => return Ok(cluster), + Err(err) => { + if index + 1 < bootstrap_count { + warn!( + "Failed to initialize cluster from bootstrap server '{}', trying the next server: {err}", + bootstrap.raw + ); + } + errors.push(format!("{}: {err}", bootstrap.raw)); + last_connection_error = Some(err); + } + } + } + + Err(Self::bootstrap_initialization_error( + bootstrap_servers, + errors, + last_connection_error, + )) + } + + fn bootstrap_initialization_error( + bootstrap_servers: &str, + errors: Vec, + last_connection_error: Option, + ) -> Error { + if let Some(error) = last_connection_error { + warn!( + "Unable to initialize cluster from bootstrap servers '{bootstrap_servers}': {}", + errors.join("; ") + ); + return error; + } + + let message = format!( + "Unable to initialize cluster from bootstrap servers '{bootstrap_servers}': {}", + errors.join("; ") ); - let con = connections.get_connection(&server_node).await?; + Error::IllegalArgument { message } + } + + async fn fetch_cluster_from_bootstrap( + server_node: &ServerNode, + connections: Arc, + ) -> Result { + let con = connections.get_connection(server_node).await?; let response = con .request(UpdateMetadataRequest::new( @@ -315,6 +409,7 @@ impl Metadata { #[cfg(test)] mod tests { use super::*; + use crate::error::ApiError; use crate::metadata::{TableBucket, TablePath}; use crate::test_utils::build_cluster_arc; @@ -341,30 +436,102 @@ mod tests { assert!(cluster.get_tablet_server(1).is_none()); } + #[test] + fn bootstrap_failure_preserves_last_connection_error() { + let authentication_error = Error::FlussAPIError { + api_error: ApiError { + code: FlussError::AuthenticateException.code(), + message: "Authentication failed".to_string(), + }, + }; + + let error = Metadata::bootstrap_initialization_error( + "127.0.0.1:9123", + vec!["127.0.0.1:9123: Authentication failed".to_string()], + Some(authentication_error), + ); + + assert_eq!(error.api_error(), Some(FlussError::AuthenticateException)); + } + #[test] fn parse_bootstrap_variants() { // valid IP - let addr = Metadata::parse_bootstrap("127.0.0.1:8080").unwrap(); + let bootstraps = Metadata::parse_bootstrap_servers("127.0.0.1:8080").unwrap(); + let addr = bootstraps[0].address.as_ref().unwrap(); + assert_eq!(addr.ip().to_string(), "127.0.0.1"); assert_eq!(addr.port(), 8080); // valid hostname - let addr = Metadata::parse_bootstrap("localhost:9090").unwrap(); + let bootstraps = Metadata::parse_bootstrap_servers("localhost:9090").unwrap(); + let addr = bootstraps[0].address.as_ref().unwrap(); + assert_eq!(addr.ip().to_string(), "127.0.0.1"); assert_eq!(addr.port(), 9090); // valid IPv6 address - let addr = Metadata::parse_bootstrap("[::1]:8080").unwrap(); + let bootstraps = Metadata::parse_bootstrap_servers("[::1]:8080").unwrap(); + let addr = bootstraps[0].address.as_ref().unwrap(); + assert_eq!(addr.ip().to_string(), "::1"); assert_eq!(addr.port(), 8080); - // invalid input: missing port - assert!(Metadata::parse_bootstrap("localhost").is_err()); + // invalid input: missing port is preserved as an entry-level parse error + let bootstraps = Metadata::parse_bootstrap_servers("localhost").unwrap(); + assert!(bootstraps[0].address.is_err()); - // invalid input: out-of-range port - assert!(Metadata::parse_bootstrap("localhost:99999").is_err()); + // invalid input: out-of-range port is preserved as an entry-level parse error + let bootstraps = Metadata::parse_bootstrap_servers("localhost:99999").unwrap(); + assert!(bootstraps[0].address.is_err()); // invalid input: empty string - assert!(Metadata::parse_bootstrap("").is_err()); + assert!(Metadata::parse_bootstrap_servers("").is_err()); + + // invalid input: nonsensical address is preserved as an entry-level parse error + let bootstraps = Metadata::parse_bootstrap_servers("invalid_address").unwrap(); + assert!(bootstraps[0].address.is_err()); + } - // invalid input: nonsensical address - assert!(Metadata::parse_bootstrap("invalid_address").is_err()); + #[test] + fn parse_bootstrap_accepts_comma_separated_servers() { + let bootstraps = + Metadata::parse_bootstrap_servers("127.0.0.1:8080, localhost:9090, [::1]:7070") + .unwrap(); + + assert_eq!(bootstraps.len(), 3); + let first = bootstraps[0].address.as_ref().unwrap(); + assert_eq!(first.ip().to_string(), "127.0.0.1"); + assert_eq!(first.port(), 8080); + let second = bootstraps[1].address.as_ref().unwrap(); + assert_eq!(second.ip().to_string(), "127.0.0.1"); + assert_eq!(second.port(), 9090); + let third = bootstraps[2].address.as_ref().unwrap(); + assert_eq!(third.ip().to_string(), "::1"); + assert_eq!(third.port(), 7070); + } + + #[test] + fn parse_bootstrap_preserves_later_entries_when_one_entry_is_unresolvable() { + let bootstraps = + Metadata::parse_bootstrap_servers("invalid_address:8080, 127.0.0.1:9090").unwrap(); + + assert_eq!(bootstraps.len(), 2); + assert!(bootstraps[0].address.is_err()); + let addr = bootstraps[1].address.as_ref().unwrap(); + assert_eq!(addr.ip().to_string(), "127.0.0.1"); + assert_eq!(addr.port(), 9090); + } + + #[test] + fn parse_bootstrap_ignores_empty_comma_separated_entries() { + let bootstraps = + Metadata::parse_bootstrap_servers("127.0.0.1:8080, , localhost:9090,").unwrap(); + + assert_eq!(bootstraps.len(), 2); + assert_eq!(bootstraps[0].raw, "127.0.0.1:8080"); + assert_eq!(bootstraps[1].raw, "localhost:9090"); + } + + #[test] + fn parse_bootstrap_rejects_config_without_servers() { + assert!(Metadata::parse_bootstrap_servers(" , ").is_err()); } } diff --git a/fluss-rust/crates/fluss/tests/integration/admin.rs b/fluss-rust/crates/fluss/tests/integration/admin.rs index dfb00b0dd7..b587fdac50 100644 --- a/fluss-rust/crates/fluss/tests/integration/admin.rs +++ b/fluss-rust/crates/fluss/tests/integration/admin.rs @@ -556,6 +556,19 @@ mod admin_test { assert!(has_tablet, "Expected a tablet server node"); } + #[tokio::test] + async fn test_bootstrap_server_failover() { + let cluster = get_shared_cluster(); + let bootstrap_servers = format!("127.0.0.1:1,{}", cluster.plaintext_bootstrap_servers()); + + FlussConnection::new(Config { + bootstrap_servers, + ..Default::default() + }) + .await + .expect("should fall back to the second bootstrap server"); + } + #[tokio::test] async fn test_error_table_not_partitioned() { let cluster = get_shared_cluster();