Struct BrokerPool 

pub struct BrokerPool<C: BrokerConnector = TcpConnector> { /* private fields */ }

Pool of Arc<Connection> keyed by broker id.

The pool opens a connection lazily on first use and caches it afterwards. A closed connection is replaced on the next use. After a failure or a disconnect, the pool waits Kafka’s reconnect backoff before it dials that broker again. Concurrent users of one broker share one connection attempt.

Implementations

impl BrokerPool<TcpConnector>

pub fn new(bootstrap: Vec<SocketAddr>, options: ConnectionOptions) -> Self

Create a new pool with the given bootstrap addresses and connection options.

pub fn new_with_server_names( bootstrap: Vec<(SocketAddr, String)>, options: ConnectionOptions, ) -> Self

Create a pool whose bootstrap TLS names retain their pre-DNS hostnames.

impl<C: BrokerConnector> BrokerPool<C>

pub async fn get(&self, broker_id: i32) -> Result<Arc<C::Conn>, ClientError>

Get-or-connect to a specific broker id. The pool must have already learned the (id, address) mapping with refresh_brokers.

When the broker is in its reconnect backoff, this method waits for the backoff to end before it dials. One call makes at most one connection attempt, to the next address of the broker.

Errors

Returns ClientError::Disconnected for a broker id that the pool does not know, the resolve error, or the dial error.

pub fn evict(&self, broker_id: i32)

Drop the cached connection to broker_id, if there is one, so the next get reconnects.

Call this after a send fails. A bounced or failed-over broker must not be retried over its dead, cached socket. As Kafka does after a disconnect, the next connection waits the reconnect backoff and resolves the host again.

pub fn evict_bootstrap(&self)

Drop the cached bootstrap connection so the next bootstrap_connection re-iterates the bootstrap addresses and reconnects to a live broker.

The pool keys the bootstrap connection by the synthetic id -1, which no real broker id matches, so evict can never reach it.

pub fn replace_bootstrap(&self, bootstrap: Vec<SocketAddr>)

Replace the bootstrap address list and drop the cached bootstrap connection so the next bootstrap send dials the fresh addresses.

pub fn replace_bootstrap_with_server_names( &self, bootstrap: Vec<(SocketAddr, String)>, )

Replace bootstrap addresses while retaining their TLS server names.

The bootstrap node keeps its reconnect backoff and failure counts, so a retry loop that calls this after each failure still backs off. Until the next refresh_brokers with brokers, least_loaded uses the bootstrap connection.

pub fn prefer_bootstrap(&self)

Make least_loaded use the bootstrap connection until the next refresh_brokers with brokers.

pub fn rebootstrap(&self, bootstrap: Vec<SocketAddr>)

Replace bootstrap addresses and discard every connection and advertised broker address learned from stale metadata.

pub fn rebootstrap_with_server_names( &self, bootstrap: Vec<(SocketAddr, String)>, )

Replace all broker state while retaining bootstrap TLS server names.

pub async fn bootstrap_connection(&self) -> Result<Arc<C::Conn>, ClientError>

Get-or-connect to the first reachable bootstrap address. The bootstrap connection is cached under the synthetic broker id -1.

One call dials each bootstrap address at most once, starting after the address that failed last. When every address fails, the next call waits the reconnect backoff first.

Errors

Returns the error of the last address, an authentication failure at once, or ClientError::Disconnected for an empty address list.

pub async fn least_loaded(&self) -> Result<Arc<C::Conn>, ClientError>

Get-or-connect to the least loaded known broker, as Kafka’s NetworkClient.leastLoadedNode picks the node for a request with no fixed target. Before the pool knows a broker, this is the bootstrap connection.

The order of preference:

1. An open connection with no request in flight.
2. The open connection with the fewest requests in flight.
3. A broker out of its reconnect backoff, the one with the oldest connection attempt first, and a broker never dialed before all.
4. The broker whose reconnect backoff ends first.

Errors

Returns the error of get or bootstrap_connection.

pub fn open_connection( &self, preferred: Option<i32>, ) -> Option<(i32, Arc<C::Conn>)>

An open connection with its id, without a new connection attempt: the one of preferred while it is open, or else the one with the fewest requests in flight. The bootstrap connection has id -1.

pub async fn refresh_brokers(&self, brokers: &[BrokerInfo])

Update the (id, host, port) registry from a list of brokers, which usually comes from a MetadataResponse. This method opens no new connections, and it does not resolve the hosts: the pool resolves a host each time it needs new addresses, as Kafka does.

This method skips brokers that advertise port 0, because that is not a dialable address. It shows up for in-process test brokers whose advertised port never got rewritten to the real bound port. When the registry leaves such an entry out, get reports Disconnected for that id. The caller can then fall back to the bootstrap connection instead of trying a doomed host:0 connect.

A broker whose host or port changed loses its cached connection and addresses.

pub fn knows_broker(&self, broker_id: i32) -> bool

Whether the registry knows a dialable address for this broker id. It knows one when refresh_brokers learned it and the port was not 0. A caller can use this to choose between a route to a specific broker and a fallback to the bootstrap connection, without a speculative connect.

pub fn broker_ids(&self) -> Vec<i32>

Return the broker ids from the most recent usable metadata.

The deterministic order keeps metadata failover predictable. Callers use this list to exhaust the last-known cluster before deciding that a fresh bootstrap is required.

pub fn broker_ids_by_load(&self) -> Vec<i32>

broker_ids, with the brokers that have an open connection first, fewest requests in flight first, then the others in id order.

Kafka sends a metadata request to NetworkClient.leastLoadedNode. A broker answers the requests of one connection in order, so a request queued behind a parked long-poll Fetch waits for up to the fetch’s max_wait_ms. A caller that walks every broker until one answers, as the metadata refresh does, walks them in this order.

pub fn finalized_features(&self) -> FinalizedFeatures

The finalized feature levels with the highest epoch that any connection of the pool received, as Kafka’s ApiVersions.getMaxFinalizedFeaturesInfo gives them.

pub fn close_all(self)

Close every open connection in the pool. Consumes the pool.