Struct Cluster 

pub struct Cluster<T: Transport> { /* private fields */ }

Connections plus the cluster map.

Implementations

impl<T: Transport> Cluster<T>

pub async fn connect( transport: T, bootstrap: &[String], client_id: &str, ) -> Result<Self>

Connect to the first reachable bootstrap address and load metadata.

Errors

If no bootstrap address answers.

pub fn set_credentials(&mut self, credentials: Credentials)

Authenticate every connection with credentials.

Must be set before the first request; connections already open are not re-authenticated, because Kafka has no way to do so.

pub fn set_request_timeout(&mut self, timeout: Duration)

How long a request may take before its connection is dropped as broken.

pub fn set_allow_auto_topic_creation(&mut self, allow: bool)

Whether metadata requests may create missing topics. See the field.

pub fn metadata(&self) -> &Metadata

The cluster map, for callers that want to inspect leadership.

pub async fn refresh_metadata(&mut self, topic: &str) -> Result<()>

Ask any broker for topic’s metadata and merge it in.

Errors

If no broker answers, or the topic carries an error code.

pub fn set_metadata_max_age(&mut self, age: Duration)

How long a topic’s metadata may go unrefreshed before a lookup re-reads it. See the refreshed_at field.

pub fn is_metadata_stale(&self, topic: &str) -> bool

Whether this topic’s metadata is older than the maximum age. A topic never read is stale.

pub async fn refresh_if_stale( &mut self, topic: &str, ) -> Result<Option<(i32, i32)>>

Re-read this topic if it is stale, and report (before, after) if its partition count grew.

Only growth: Kafka has no operation that removes partitions from a live topic, so a smaller number is a transient answer — a broker that has not caught up, a topic mid-creation — and acting on it would move every key twice. A topic seen for the first time reports no growth either; there is nothing to have grown from.

Errors

If the refresh fails.

pub fn forget_topic(&mut self, topic: &str)

Forget a topic that has been deleted — its count and every leader.

pub async fn refresh_cluster(&mut self) -> Result<()>

Refresh brokers and the controller without naming a topic.

An empty topic list is not the same as no topic list: None asks for every topic in the cluster, which on a large cluster is a large answer for information this does not want.

Errors

If no broker answers.

pub async fn controller_addr(&mut self) -> Result<String>

The controller’s address, refreshing if it is unknown.

Errors

If metadata cannot be refreshed, or no controller is elected — which happens briefly during a controller election and is a wait, not a failure, so callers retry it.

pub fn invalidate_controller(&mut self)

Forget which broker is the controller, after it said it is not.

pub async fn leader_addr( &mut self, topic: &str, partition: i32, ) -> Result<String>

The address of topic/partition’s leader, refreshing if unknown.

Errors

If metadata cannot be refreshed, or the partition still has no leader afterwards — which is what a partition mid-election looks like, and is Error::NoLeader so a caller can back off and try again rather than treat it as fatal.

pub async fn partition_count(&mut self, topic: &str) -> Result<i32>

How many partitions topic has, refreshing metadata if the client has not seen it yet.

Errors

If metadata cannot be refreshed, or the topic has no partitions after it.

pub fn invalidate(&mut self, topic: &str, partition: i32)

Forget one partition’s leader after the broker said it is not the leader. Per partition on purpose — see barnabas_core::metadata.

pub fn connection_count(&self) -> usize

Number of open connections. Exposed because connection count is a real cost of the per-core design, and something a caller may want to watch.