Struct Client 

pub struct Client { /* private fields */ }

A Kafka client backed by a BrokerPool.

Construct a Client with Client::builder.

A clone of a Client is cheap. The clone shares the underlying BrokerPool through an Arc and copies the connection options by value.

Implementations

impl Client

pub async fn start( bootstrap: String, client_id: String, dns_timeout: Time, socket_connection_setup_timeout: Time, socket_connection_setup_timeout_max: Time, reconnect_backoff: Time, reconnect_backoff_max: Time, connections_max_idle: Time, client_dns_lookup: ClientDnsLookup, send_buffer: Option<ByteSize>, receive_buffer: Option<ByteSize>, request_timeout: Time, dispatch_queue_capacity: usize, frame_max: ByteSize, metadata_recovery_strategy: MetadataRecoveryStrategy, metadata_recovery_rebootstrap_trigger: Time, security: Option<ClientSecurity>, metadata_scope: MetadataScope, telemetry: Option<ClientTelemetryConfig>, ) -> Result<Self, ClientError>

Build a Client pointed at the given bootstrap address.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub fn builder() -> ClientStartBuilder

Build a Client pointed at the given bootstrap address.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

impl Client

pub fn connection_options(&self) -> &ConnectionOptions

The options of every connection that this client opens, such as the socket buffers (send.buffer.bytes, receive.buffer.bytes) that each broker socket gets before it connects.

pub fn metrics(&self) -> Option<&ClientMetrics>

The metric registry of a client that pushes its metrics (KIP-714). A client crate registers its own metrics in it.

pub async fn close_telemetry(&self)

Send the terminating telemetry push (KIP-714), if the client pushes metrics and has a subscription, and wait for it. Kafka’s clients send it from close. Without this call, the push goes out in the background once the last clone of the client drops.

pub async fn client_instance_id( &self, timeout: Time, ) -> Result<Option<Uuid>, ClientError>

The client instance id that the broker assigned for metrics push (KIP-714), waiting up to timeout for the first subscription, as Kafka’s clientInstanceId does. Ok(None) when no subscription has loaded by then, as Kafka returns null.

Errors

Returns ClientError::InvalidArgument for a negative timeout, and ClientError::TelemetryDisabled for a client that does not push metrics.

pub async fn send<R: ProtocolRequest>( &self, req: R, ) -> Result<R::Response, ClientError>

Send a request that has no fixed target broker.

Before the first metadata response, the request goes to a bootstrap address. After it, the request goes to the least loaded known broker, as Kafka’s NetworkClient.leastLoadedNode picks the node.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub async fn send_at_least<R: ProtocolRequest>( &self, req: R, min_version: i16, ) -> Result<R::Response, ClientError>

Send a request, as send routes it, only when the broker and client share a version at or above min_version.

Errors

Returns ClientError::IncompatibleVersion before dispatch when the shared range is below min_version. Other errors match send.

pub async fn send_to_connection<R: ProtocolRequest>( &self, broker_id: Option<i32>, req: R, ) -> Result<R::Response, ClientError>

Send on one fixed pool connection, without selecting another broker.

Some(id) selects that broker’s connection. None selects the distinct bootstrap connection even after cluster metadata has been learned. Callers that account for per-connection request limits can keep this selection fixed until the request completes, then refresh and retry on failure themselves.

Errors

Returns a routing, connection setup, encoding or response error.

pub async fn send_to_connection_no_response<R: ProtocolRequest>( &self, broker_id: Option<i32>, req: R, ) -> Result<(), ClientError>

Send without a response on the fixed connection selected by send_to_connection.

Errors

Returns a routing, connection setup, encoding or writer enqueue error.

pub async fn send_no_response<R: ProtocolRequest>( &self, req: R, ) -> Result<(), ClientError>

Send a request, as send routes it, without registering a pending response. Intended for protocol operations such as Produce acks=0.

Errors

Returns an error if the connection cannot be opened or cannot accept the encoded request.

pub async fn reconnect_bootstrap(&self)

Drop the cached bootstrap connection and refresh the bootstrap address list from the original bootstrap string.

Callers that know their bootstrap request is safe to retry can call this after a transport error and before they send again.

pub fn finalized_features(&self) -> FinalizedFeatures

The finalized feature levels (KIP-584) with the highest epoch that a connection of this client received in ApiVersions. Kafka’s producer reads transaction.version from them. The epoch is UNKNOWN_FINALIZED_FEATURES_EPOCH before a connection to a broker that sends features.

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

Whether the pool knows a dialable address for broker_id.

The pool knows one when refresh_metadata learned it and the port was not 0. A caller can then choose between broker routing and the bootstrap send without a speculative connect.

pub fn broker(&self, broker_id: i32) -> BrokerHandle<'_>

Return a BrokerHandle that routes requests to a specific broker by id.

refresh_metadata must have registered the broker first.

pub fn evict_broker(&self, broker_id: i32)

Drop the pooled connection to broker_id so the next request to it reconnects to its current advertised address.

Call this after a send fails, so a bounced or failed-over broker is not retried over a dead socket.

pub fn metadata_topics(&self) -> &MetadataTopics

The topics that refresh_metadata names. The clones of this client share them.

pub async fn refresh_metadata(&self) -> Result<MetadataResponse, ClientError>

Send the MetadataRequest of metadata_topics, parse the broker list from the response, refresh the pool’s address registry, and return the typed response.

With MetadataScope::AllTopics the request asks for all topics. With MetadataScope::Topics it names the current topics, as Kafka’s producer and consumer do. Use refresh_metadata_with for another request.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub async fn refresh_metadata_with( &self, request: MetadataRequest, ) -> Result<MetadataResponse, ClientError>

Send request, parse the broker list from the response, refresh the pool’s address registry, and return the typed response.

Kafka’s producer names its topics in the request (ProducerMetadata.newMetadataRequestBuilder). A request that names a topic lets the broker create the topic when allow_auto_topic_creation is set, and returns a per-topic error such as TOPIC_AUTHORIZATION_FAILED for it. Every retry and rebootstrap in this call sends the same request.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

pub async fn offset_for_leader_epoch( &self, topic: &str, partition: i32, current_leader_epoch: i32, leader_epoch: i32, ) -> Result<EpochEndOffset, ClientError>

Send a single-partition OffsetForLeaderEpoch (api_key=23) over the bootstrap connection.

This is a thin wrapper over the free offset_for_leader_epoch helper that the consumer’s KIP-320 position-validation pass uses. Client does not otherwise expose its connection, so this method borrows the same bootstrap connection that send uses.

Errors

Returns an error on transport or version-negotiation failure, or when the response does not contain the partition.

pub async fn offset_for_leader_epoch_on( &self, broker_id: i32, topic: &str, partition: i32, current_leader_epoch: i32, leader_epoch: i32, ) -> Result<EpochEndOffset, ClientError>

Send a single-partition OffsetForLeaderEpoch (api_key=23) to a specific broker by id, through BrokerPool::get.

This method mirrors offset_for_leader_epoch but targets the partition leader instead of the bootstrap connection. KIP-320 requires the validation RPC to reach the partition leader, which is the only replica with the authoritative epoch→end-offset history.

The broker must already be in the pool’s registry, which refresh_metadata populates.

Errors

Returns Disconnected if broker_id is not in the registry. Returns an error on transport or version-negotiation failure. Returns an error when the response does not contain the partition.

pub async fn fetch_partition_with_isolation_on( &self, broker_id: i32, fetch: IsolatedFetch<'_>, ) -> Result<Vec<FetchedRecord>, ClientError>

Fetch one partition from a specific broker with the requested isolation level. The broker address must have been learned through refresh_metadata.

Errors

Returns a connection, protocol, or broker error. A failed connection to a known broker is evicted, metadata is refreshed, and the request is retried once against the broker’s current advertised address.

pub fn close(self)

Close the client and all pooled connections.

Trait Implementations

impl Clone for Client

fn clone(&self) -> Client

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl TelemetryConnections for Client

fn open_connection( &self, preferred: Option<i32>, ) -> Option<(i32, Arc<Connection>)>

An open connection with its broker id, preferring the broker of preferred while it stays open. Per KIP-714 a client keeps pushing to the same broker. None when no connection is open.