use crate::cluster::{Cluster, ServerNode, ServerType};
use crate::error::{Error, FlussError, Result};
use crate::metadata::{PhysicalTablePath, TableBucket, TablePath};
use crate::proto::MetadataResponse;
use crate::rpc::message::{GetTableRequest, UpdateMetadataRequest};
use crate::rpc::{RpcClient, ServerConnection};
use crate::{PartitionId, TableId};
use log::{info, warn};
use parking_lot::RwLock;
use std::collections::HashSet;
use std::future::Future;
use std::net::{SocketAddr, ToSocketAddrs};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;
const MAX_BOOTSTRAP_RETRIES: usize = 3;
const BOOTSTRAP_RETRY_INTERVAL_MS: u64 = 100;
pub struct Metadata {
cluster: RwLock<Arc<Cluster>>,
unavailable_tablet_server_ids: RwLock<HashSet<i32>>,
connections: Arc<RpcClient>,
bootstrap: Arc<str>,
cluster_version_tx: watch::Sender<u64>,
}
#[derive(Debug)]
struct BootstrapServer {
raw: String,
address: Result<SocketAddr>,
}
impl Metadata {
pub async fn new(bootstrap: &str, connections: Arc<RpcClient>) -> Result<Self> {
let cluster = Self::init_cluster(bootstrap, connections.clone()).await?;
let (cluster_version_tx, _) = watch::channel(0);
Ok(Metadata {
cluster: RwLock::new(Arc::new(cluster)),
unavailable_tablet_server_ids: RwLock::new(HashSet::new()),
connections,
bootstrap: bootstrap.into(),
cluster_version_tx,
})
}
pub fn subscribe_cluster_changes(&self) -> watch::Receiver<u64> {
self.cluster_version_tx.subscribe()
}
fn notify_cluster_changed(&self) {
self.cluster_version_tx
.send_modify(|v| *v = v.wrapping_add(1));
}
fn parse_bootstrap(bootstrap: &str) -> Result<SocketAddr> {
let addrs = bootstrap
.to_socket_addrs()
.map_err(|e| Error::IllegalArgument {
message: format!("Invalid bootstrap address '{bootstrap}': {e}"),
})?;
let mut ipv6_candidate: Option<SocketAddr> = None;
for addr in addrs {
if addr.is_ipv4() {
return Ok(addr);
}
if ipv6_candidate.is_none() {
ipv6_candidate = Some(addr);
}
}
let addr = ipv6_candidate.ok_or_else(|| Error::IllegalArgument {
message: format!("Unable to resolve bootstrap address '{bootstrap}'"),
})?;
Ok(addr)
}
fn parse_bootstrap_servers(bootstrap_servers: &str) -> Result<Vec<BootstrapServer>> {
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 retry_bootstrap<T, Operation, OperationFuture, Disconnect>(
bootstrap: &str,
mut operation: Operation,
mut disconnect: Disconnect,
) -> Result<T>
where
Operation: FnMut() -> OperationFuture,
OperationFuture: Future<Output = Result<T>>,
Disconnect: FnMut(),
{
let mut retry_count = 0;
loop {
match operation().await {
Ok(result) => return Ok(result),
Err(err) if err.is_retriable() && retry_count < MAX_BOOTSTRAP_RETRIES => {
disconnect();
let delay_ms = BOOTSTRAP_RETRY_INTERVAL_MS * (1_u64 << retry_count);
retry_count += 1;
warn!(
"Failed to initialize cluster from bootstrap server '{bootstrap}' \
(retry {retry_count}/{MAX_BOOTSTRAP_RETRIES}). Retrying in \
{delay_ms} ms: {err}"
);
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
}
Err(err) => return Err(err),
}
}
}
async fn init_cluster(bootstrap_servers: &str, connections: Arc<RpcClient>) -> Result<Cluster> {
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,
);
let cluster_result = if bootstrap_count == 1 {
Self::retry_bootstrap(
&bootstrap.raw,
|| Self::fetch_cluster_from_bootstrap(&server_node, connections.clone()),
|| connections.disconnect(server_node.uid()),
)
.await
} else {
Self::fetch_cluster_from_bootstrap(&server_node, connections.clone()).await
};
match cluster_result {
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<String>,
last_connection_error: Option<Error>,
) -> 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("; ")
);
Error::IllegalArgument { message }
}
async fn fetch_cluster_from_bootstrap(
server_node: &ServerNode,
connections: Arc<RpcClient>,
) -> Result<Cluster> {
let con = connections.get_connection(server_node).await?;
let response = con
.request(UpdateMetadataRequest::new(
&HashSet::default(),
&HashSet::new(),
vec![],
))
.await?;
Cluster::from_metadata_response(response, None)
}
pub(crate) async fn reinit_cluster(&self) -> Result<()> {
let cluster = Self::init_cluster(&self.bootstrap, self.connections.clone()).await?;
self.replace_cluster(cluster);
self.notify_cluster_changed();
Ok(())
}
fn replace_cluster(&self, cluster: Cluster) {
let mut guard = self.cluster.write();
self.unavailable_tablet_server_ids
.write()
.retain(|id| cluster.get_tablet_server(*id).is_none());
*guard = Arc::new(cluster);
}
pub fn invalidate_server(&self, server_id: &i32, table_ids: Vec<i64>) {
{
let mut cluster_guard = self.cluster.write();
let updated_cluster = cluster_guard.invalidate_server(server_id, table_ids);
*cluster_guard = Arc::new(updated_cluster);
}
self.notify_cluster_changed();
}
pub fn invalidate_physical_table_meta(
&self,
physical_tables_to_invalid: &HashSet<PhysicalTablePath>,
) {
{
let mut cluster_guard = self.cluster.write();
let updated_cluster =
cluster_guard.invalidate_physical_table_meta(physical_tables_to_invalid);
*cluster_guard = Arc::new(updated_cluster);
}
self.notify_cluster_changed();
}
pub async fn fetch_table_id(&self, table_path: &TablePath) -> Result<Option<TableId>> {
let maybe_server = {
let guard = self.cluster.read();
guard.get_one_available_server().cloned()
};
let server = maybe_server.ok_or_else(|| Error::UnexpectedError {
message: "No available server to fetch table metadata".to_string(),
source: None,
})?;
let conn = self.connections.get_connection(&server).await?;
match conn.request(GetTableRequest::new(table_path)).await {
Ok(response) => Ok(Some(response.table_id)),
Err(e) if e.api_error() == Some(FlussError::TableNotExist) => Ok(None),
Err(e) => Err(e),
}
}
pub fn evict_table_metadata(&self, table_path: &TablePath) {
{
let mut cluster_guard = self.cluster.write();
let updated_cluster = cluster_guard.evict_table(table_path);
*cluster_guard = Arc::new(updated_cluster);
}
self.notify_cluster_changed();
}
pub async fn update(&self, metadata_response: MetadataResponse) -> Result<()> {
let origin_cluster = self.cluster.read().clone();
let new_cluster =
Cluster::from_metadata_response(metadata_response, Some(&origin_cluster))?;
self.replace_cluster(new_cluster);
self.notify_cluster_changed();
Ok(())
}
pub async fn update_tables_metadata(
&self,
table_paths: &HashSet<&TablePath>,
physical_table_paths: &HashSet<&Arc<PhysicalTablePath>>,
partition_ids: Vec<i64>,
) -> Result<()> {
let maybe_server = {
let guard = self.cluster.read();
guard
.get_one_available_server_excluding(&self.unavailable_tablet_server_ids.read())
.cloned()
};
let server = match maybe_server {
Some(s) => s,
None => {
info!(
"No available tablet server to update metadata, attempting to re-initialize cluster using bootstrap server."
);
self.reinit_cluster().await?;
return Ok(());
}
};
let result = async {
let conn = self.connections.get_connection(&server).await?;
let response = conn
.request(UpdateMetadataRequest::new(
table_paths,
physical_table_paths,
partition_ids,
))
.await?;
self.update(response).await
}
.await;
if let Err(err) = &result {
if err.is_retriable() {
self.unavailable_tablet_server_ids
.write()
.insert(server.id());
warn!("Tablet server {server:?} is unavailable for updating metadata: {err}");
}
}
result
}
pub async fn update_table_metadata(&self, table_path: &TablePath) -> Result<()> {
self.update_tables_metadata(&HashSet::from([table_path]), &HashSet::new(), vec![])
.await
}
pub async fn update_physical_table_metadata(
&self,
physical_table_paths: &[Arc<PhysicalTablePath>],
) -> Result<()> {
let mut update_table_paths = HashSet::new();
let mut update_partition_paths = HashSet::new();
for physical_table_path in physical_table_paths {
match physical_table_path.get_partition_name() {
Some(_) => {
update_partition_paths.insert(physical_table_path);
}
None => {
update_table_paths.insert(physical_table_path.get_table_path());
}
}
}
self.update_tables_metadata(&update_table_paths, &update_partition_paths, vec![])
.await
}
pub async fn check_and_update_table_metadata(&self, table_paths: &[TablePath]) -> Result<()> {
let cluster_binding = self.cluster.read().clone();
let need_update_table_paths: HashSet<&TablePath> = table_paths
.iter()
.filter(|table_path| cluster_binding.opt_get_table(table_path).is_none())
.collect();
if !need_update_table_paths.is_empty() {
self.update_tables_metadata(&need_update_table_paths, &HashSet::new(), vec![])
.await?;
}
Ok(())
}
pub async fn check_and_update_partition_metadata_by_ids(
&self,
table_path: &TablePath,
partition_ids: &[PartitionId],
) -> Result<()> {
let cluster_binding = self.cluster.read().clone();
let need_update_partition_ids: Vec<PartitionId> = partition_ids
.iter()
.filter(|partition_id| cluster_binding.get_partition_name(**partition_id).is_none())
.copied()
.collect::<HashSet<_>>()
.into_iter()
.collect();
if !need_update_partition_ids.is_empty() {
self.update_tables_metadata(
&HashSet::from([table_path]),
&HashSet::new(),
need_update_partition_ids,
)
.await?;
}
Ok(())
}
pub async fn check_and_update_partition_metadata(
&self,
physical_table_path: &PhysicalTablePath,
) -> Result<Option<PartitionId>> {
if let Some(id) = self.get_cluster().get_partition_id(physical_table_path) {
return Ok(Some(id));
}
let path = Arc::new(physical_table_path.clone());
match self.update_physical_table_metadata(&[path]).await {
Ok(()) => {}
Err(e) if matches!(e.api_error(), Some(FlussError::PartitionNotExists)) => {
return Ok(None);
}
Err(e) => return Err(e),
}
Ok(self.get_cluster().get_partition_id(physical_table_path))
}
pub(crate) async fn get_connection(
&self,
server_node: &ServerNode,
) -> Result<ServerConnection> {
let result = self.connections.get_connection(server_node).await?;
Ok(result)
}
pub fn get_cluster(&self) -> Arc<Cluster> {
let guard = self.cluster.read();
guard.clone()
}
const MAX_RETRY_TIMES: u8 = 3;
pub async fn leader_for(
&self,
table_path: &TablePath,
table_bucket: &TableBucket,
) -> Result<Option<ServerNode>> {
let leader = self.get_leader_for(table_bucket);
if leader.is_some() {
Ok(leader)
} else {
for _ in 0..Self::MAX_RETRY_TIMES {
if let Some(partition_id) = table_bucket.partition_id() {
self.update_tables_metadata(
&HashSet::from([table_path]),
&HashSet::new(),
vec![partition_id],
)
.await?;
} else {
self.update_tables_metadata(
&HashSet::from([table_path]),
&HashSet::new(),
vec![],
)
.await?;
}
let cluster = self.cluster.read();
let leader = cluster.leader_for(table_bucket);
if leader.is_some() {
return Ok(leader.cloned());
}
}
Ok(None)
}
}
fn get_leader_for(&self, table_bucket: &TableBucket) -> Option<ServerNode> {
let cluster = self.cluster.read();
cluster.leader_for(table_bucket).cloned()
}
}
#[cfg(test)]
impl Metadata {
pub(crate) fn new_for_test(cluster: Arc<Cluster>) -> Self {
let (cluster_version_tx, _) = watch::channel(0);
Metadata {
cluster: RwLock::new(cluster),
unavailable_tablet_server_ids: RwLock::new(HashSet::new()),
connections: Arc::new(RpcClient::new()),
bootstrap: Arc::from(""),
cluster_version_tx,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::ApiError;
use crate::metadata::{TableBucket, TablePath};
use crate::proto::{ApiVersionsResponse, ErrorResponse, PbApiVersion, PbServerNode};
use crate::rpc::ApiKey;
use crate::test_utils::build_cluster_arc;
use prost::Message;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
fn server_metadata(ids: &[i32], port: u16) -> MetadataResponse {
MetadataResponse {
tablet_servers: ids
.iter()
.map(|id| PbServerNode {
node_id: *id,
host: "127.0.0.1".to_string(),
port: i32::from(port),
..Default::default()
})
.collect(),
..Default::default()
}
}
async fn serve_metadata(
listener: TcpListener,
response: std::result::Result<MetadataResponse, FlussError>,
) {
let (mut stream, _) = listener.accept().await.unwrap();
let versions = ApiVersionsResponse {
api_versions: vec![PbApiVersion {
api_key: i32::from(i16::from(ApiKey::MetaData)),
min_version: 0,
max_version: 0,
}],
server_type: Some(ServerType::TabletServer.to_type_id()),
};
for api_key in [ApiKey::ApiVersion, ApiKey::MetaData] {
let len = stream.read_u32().await.unwrap() as usize;
assert!((8..=1024 * 1024).contains(&len));
let mut request = vec![0; len];
stream.read_exact(&mut request).await.unwrap();
assert_eq!(
i16::from_be_bytes([request[0], request[1]]),
i16::from(api_key)
);
let mut reply = vec![0];
reply.extend_from_slice(&request[4..8]);
if api_key == ApiKey::ApiVersion {
versions.encode(&mut reply).unwrap();
} else {
match &response {
Ok(metadata) => metadata.encode(&mut reply).unwrap(),
Err(error) => {
reply[0] = 1;
ErrorResponse {
error_code: error.code(),
error_message: Some("metadata request failed".to_string()),
}
.encode(&mut reply)
.unwrap();
}
}
}
stream.write_u32(reply.len() as u32).await.unwrap();
stream.write_all(&reply).await.unwrap();
}
}
#[tokio::test]
async fn metadata_refresh_recovers_when_all_cached_servers_are_unavailable() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let stale_cluster =
Cluster::from_metadata_response(server_metadata(&[1, 2, 3], 0), None).unwrap();
let mut metadata = Metadata::new_for_test(Arc::new(stale_cluster));
metadata.bootstrap = address.to_string().into();
let mut changes = metadata.subscribe_cluster_changes();
for failed_count in 1..=3 {
let error = metadata
.update_tables_metadata(&HashSet::new(), &HashSet::new(), vec![])
.await
.unwrap_err();
assert!(error.is_retriable());
assert_eq!(
metadata.unavailable_tablet_server_ids.read().len(),
failed_count
);
assert!(!changes.has_changed().unwrap());
}
let server = tokio::spawn(serve_metadata(
listener,
Ok(server_metadata(&[1, 2, 3], address.port())),
));
metadata
.update_tables_metadata(&HashSet::new(), &HashSet::new(), vec![])
.await
.unwrap();
server.await.unwrap();
assert!(metadata.unavailable_tablet_server_ids.read().is_empty());
for id in 1..=3 {
assert_eq!(
metadata.get_cluster().get_tablet_server(id).unwrap().port(),
u32::from(address.port())
);
}
assert!(changes.has_changed().unwrap());
assert_eq!(*changes.borrow_and_update(), 1);
}
#[tokio::test]
async fn metadata_refresh_only_excludes_servers_on_retriable_request_errors() {
for error in [FlussError::RequestTimeOut, FlussError::PartitionNotExists] {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let cluster =
Cluster::from_metadata_response(server_metadata(&[1], port), None).unwrap();
let metadata = Metadata::new_for_test(Arc::new(cluster));
let server = tokio::spawn(serve_metadata(listener, Err(error)));
let result = metadata
.update_tables_metadata(&HashSet::new(), &HashSet::new(), vec![])
.await
.unwrap_err();
server.await.unwrap();
assert_eq!(result.api_error(), Some(error));
assert_eq!(
metadata.unavailable_tablet_server_ids.read().contains(&1),
result.is_retriable()
);
}
}
#[tokio::test]
async fn metadata_refresh_uses_remaining_server_and_clears_recovered_ids() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let mut response = server_metadata(&[1, 2], 0);
response.tablet_servers[1].port = i32::from(port);
let cluster = Cluster::from_metadata_response(response, None).unwrap();
let metadata = Metadata::new_for_test(Arc::new(cluster));
metadata
.unavailable_tablet_server_ids
.write()
.extend([1, 3]);
let server = tokio::spawn(serve_metadata(listener, Ok(server_metadata(&[1, 2], port))));
metadata
.update_tables_metadata(&HashSet::new(), &HashSet::new(), vec![])
.await
.unwrap();
server.await.unwrap();
assert_eq!(
*metadata.unavailable_tablet_server_ids.read(),
HashSet::from([3])
);
assert_eq!(
metadata.get_cluster().get_tablet_server(1).unwrap().port(),
u32::from(port)
);
}
#[tokio::test]
async fn failed_bootstrap_preserves_unavailable_servers_and_cached_metadata() {
let cluster =
Arc::new(Cluster::from_metadata_response(server_metadata(&[1], 0), None).unwrap());
let metadata = Metadata::new_for_test(cluster.clone());
metadata.unavailable_tablet_server_ids.write().insert(1);
let changes = metadata.subscribe_cluster_changes();
let error = metadata
.update_tables_metadata(&HashSet::new(), &HashSet::new(), vec![])
.await
.unwrap_err();
assert!(matches!(error, Error::IllegalArgument { .. }));
assert!(Arc::ptr_eq(&metadata.get_cluster(), &cluster));
assert_eq!(
*metadata.unavailable_tablet_server_ids.read(),
HashSet::from([1])
);
assert!(!changes.has_changed().unwrap());
}
#[tokio::test(start_paused = true)]
async fn bootstrap_retry_succeeds_after_retriable_failures() {
let attempts = Arc::new(AtomicUsize::new(0));
let disconnects = Arc::new(AtomicUsize::new(0));
let started_at = tokio::time::Instant::now();
let result = Metadata::retry_bootstrap(
"localhost:9123",
{
let attempts = attempts.clone();
move || {
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
async move {
if attempt < 2 {
Err(crate::rpc::RpcError::ConnectionError(
"bootstrap is recovering".to_string(),
)
.into())
} else {
Ok(Cluster::default())
}
}
}
},
{
let disconnects = disconnects.clone();
move || {
disconnects.fetch_add(1, Ordering::SeqCst);
}
},
)
.await;
assert!(result.is_ok());
assert_eq!(attempts.load(Ordering::SeqCst), 3);
assert_eq!(disconnects.load(Ordering::SeqCst), 2);
assert_eq!(
tokio::time::Instant::now().duration_since(started_at),
std::time::Duration::from_millis(300)
);
}
#[tokio::test(start_paused = true)]
async fn bootstrap_retry_returns_last_error_after_exhaustion() {
let attempts = Arc::new(AtomicUsize::new(0));
let disconnects = Arc::new(AtomicUsize::new(0));
let started_at = tokio::time::Instant::now();
let result = Metadata::retry_bootstrap(
"localhost:9123",
{
let attempts = attempts.clone();
move || {
attempts.fetch_add(1, Ordering::SeqCst);
async {
Err::<Cluster, Error>(
FlussError::RequestTimeOut
.to_api_error(Some("bootstrap is recovering".to_string()))
.into(),
)
}
}
},
{
let disconnects = disconnects.clone();
move || {
disconnects.fetch_add(1, Ordering::SeqCst);
}
},
)
.await;
let error = match result {
Ok(_) => panic!("retries should be exhausted"),
Err(error) => error,
};
assert_eq!(error.api_error(), Some(FlussError::RequestTimeOut));
assert_eq!(attempts.load(Ordering::SeqCst), 4);
assert_eq!(disconnects.load(Ordering::SeqCst), 3);
assert_eq!(
tokio::time::Instant::now().duration_since(started_at),
std::time::Duration::from_millis(700)
);
}
#[tokio::test(start_paused = true)]
async fn bootstrap_retry_does_not_retry_non_retriable_error() {
let attempts = Arc::new(AtomicUsize::new(0));
let disconnects = Arc::new(AtomicUsize::new(0));
let result = Metadata::retry_bootstrap(
"localhost:9123",
{
let attempts = attempts.clone();
move || {
attempts.fetch_add(1, Ordering::SeqCst);
async {
Err::<Cluster, Error>(Error::IllegalArgument {
message: "invalid bootstrap response".to_string(),
})
}
}
},
{
let disconnects = disconnects.clone();
move || {
disconnects.fetch_add(1, Ordering::SeqCst);
}
},
)
.await;
assert!(matches!(result, Err(Error::IllegalArgument { .. })));
assert_eq!(attempts.load(Ordering::SeqCst), 1);
assert_eq!(disconnects.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn leader_for_returns_server() {
let table_path = TablePath::new("db".to_string(), "tbl".to_string());
let cluster = build_cluster_arc(&table_path, 1, 1);
let metadata = Metadata::new_for_test(cluster);
let leader = metadata
.leader_for(&table_path, &TableBucket::new(1, 0))
.await
.unwrap()
.expect("leader");
assert_eq!(leader.id(), 1);
}
#[test]
fn invalidate_server_removes_leader() {
let table_path = TablePath::new("db".to_string(), "tbl".to_string());
let cluster = build_cluster_arc(&table_path, 1, 1);
let metadata = Metadata::new_for_test(cluster);
metadata.invalidate_server(&1, vec![1]);
let cluster = metadata.get_cluster();
assert!(cluster.get_tablet_server(1).is_none());
}
#[test]
fn evict_table_metadata_purges_cluster() {
let table_path = TablePath::new("db".to_string(), "tbl".to_string());
let cluster = build_cluster_arc(&table_path, 1, 1);
let metadata = Metadata::new_for_test(cluster);
metadata.evict_table_metadata(&table_path);
let cluster = metadata.get_cluster();
assert!(cluster.get_table_id(&table_path).is_none());
assert!(cluster.opt_get_table(&table_path).is_none());
assert!(cluster.leader_for(&TableBucket::new(1, 0)).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() {
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);
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);
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);
let bootstraps = Metadata::parse_bootstrap_servers("localhost").unwrap();
assert!(bootstraps[0].address.is_err());
let bootstraps = Metadata::parse_bootstrap_servers("localhost:99999").unwrap();
assert!(bootstraps[0].address.is_err());
assert!(Metadata::parse_bootstrap_servers("").is_err());
let bootstraps = Metadata::parse_bootstrap_servers("invalid_address").unwrap();
assert!(bootstraps[0].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.invalid: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());
}
}