Struct BrokerTransport 

pub struct BrokerTransport { /* private fields */ }

The CoordinationTransport that talks to a Kafka cluster.

Build one with BrokerTransport::builder. One transport serves every role of one process. It opens the connections it needs on first use, and BrokerTransport::close shuts them down.

The module documentation states the parts of the design that the call sites do not show.

Implementations

impl BrokerTransport

pub async fn start( bootstrap: String, client_id: String, lease_duration: Time, request_timeout: Time, fetch_max_wait: Time, fetch_partition_max: ByteSize, topic_partitions: i32, topic_replication: i32, security: Option<ClientSecurity>, ) -> Result<Self, CoordinationError>

Builds a transport against the cluster at bootstrap.

bootstrap takes one host:port, or several separated by commas. lease_duration becomes the transaction timeout of every role, so the transaction coordinator aborts the open transaction of a dead holder after that extent.

The call opens the admin connection. It does not create the topic and it does not build a producer. The first call that needs the topic creates it.

Errors

Returns CoordinationError::InvalidConfig when bootstrap holds no address, when an address does not resolve, or when the partition count or the replication factor is not positive. Returns CoordinationError::Admin when no bootstrap address accepts the admin connection.

pub fn builder() -> BrokerTransportStartBuilder

Builds a transport against the cluster at bootstrap.

bootstrap takes one host:port, or several separated by commas. lease_duration becomes the transaction timeout of every role, so the transaction coordinator aborts the open transaction of a dead holder after that extent.

The call opens the admin connection. It does not create the topic and it does not build a producer. The first call that needs the topic creates it.

Errors

Returns CoordinationError::InvalidConfig when bootstrap holds no address, when an address does not resolve, or when the partition count or the replication factor is not positive. Returns CoordinationError::Admin when no bootstrap address accepts the admin connection.

impl BrokerTransport

pub async fn bound_producer(&self, role: &Role) -> Option<Arc<Producer>>

The producer that CoordinationTransport::acquire_epoch bound to the epoch of role.

The result is None when this process holds no epoch for role. A leadership handle gives the producer to the caller, and the broker fences every write the caller makes with it.

pub async fn held_token(&self, role: &Role) -> Option<FencingToken>

The epoch that CoordinationTransport::acquire_epoch minted for role in this process.

The result is None when this process holds no epoch for role. The value is what this process believes. The cluster is the authority, and CoordinationTransport::describe asks it.

pub async fn close(self)

Closes every producer and every connection this transport opened.

Each producer runs a sender task, and this call stops those tasks. Drop every leadership handle before this call. A producer that a handle still shares stays open, and this call reports it in a warning.

Trait Implementations

impl CoordinationTransport for BrokerTransport

fn acquire_epoch<'life0, 'life1, 'async_trait>( &'life0 self, role: &'life1 Role, ) -> Pin<Box<dyn Future<Output = Result<FencingToken, CoordinationError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mints the epoch of role and fences the member that held it.

The call builds the transactional producer of the role and calls init_transactions, which sends InitProducerId to the transaction coordinator. The transport keeps that producer, because it stays bound to the epoch the coordinator minted.

The call then reads the minted pair out of that producer with Producer::transactional_identity. It does not ask the coordinator again. A second request would report whatever the coordinator holds at that moment, so a competitor that mints in between would hand this member the token of the competitor. Producer::producer_id is a different pair again: it reports the build-time InitProducerId, which carries no transactional.id, and the coordinator never advances it.

Errors

Returns CoordinationError::Fenced when a later member already took role while this call ran. Returns CoordinationError::Producer when the coordinator refuses the call or returns no identity, and CoordinationError::Client when a connection fails.

fn read_role_records<'life0, 'life1, 'async_trait>( &'life0 self, role: &'life1 Role, ) -> Pin<Box<dyn Future<Output = Result<RoleRecords, CoordinationError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Reads the partition of role and returns its records in offset order.

The scan runs at isolation level 1, so an aborted lease write is invisible. It starts at the first offset the partition still holds, and it stops when the cursor reaches the last stable offset that the cluster reported at the start of the scan. The scan does not wait for a record that a writer appends while it runs.

The scan drops a record whose key does not decode, and a record whose key names another role. Both belong to a writer that this role does not own.

Errors

Returns CoordinationError::Client when the metadata read, the offset lookup, or a fetch fails, and when the cluster reports no leader for the partition of role. Returns CoordinationError::Record when the value of a record of role does not decode. Returns CoordinationError::InvalidConfig when the topic reports no partition.

fn register<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, role: &'life1 Role, member: &'life2 MemberId, ) -> Pin<Box<dyn Future<Output = Result<(), CoordinationError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Appends the registration of member to the partition of role.

The append travels on the plain producer and sits outside a transaction, because a candidate holds no epoch. The offset the broker gives it is the join sequence of the member.

Errors

Returns CoordinationError::Producer when the append fails, and CoordinationError::Client when a connection fails. This method never reports CoordinationError::Fenced, because its producer holds no epoch for the role.

fn write_lease<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, role: &'life1 Role, token: FencingToken, lease: &'life2 Lease, ) -> Pin<Box<dyn Future<Output = Result<(), CoordinationError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Writes the lease of role in a transaction under token.

The write travels on the producer that CoordinationTransport::acquire_epoch bound to token. The broker checks the epoch of that producer, so a superseded holder cannot append a lease. That check is how a deposed holder learns that it lost the role.

A failed send aborts the transaction, so the partial write stays invisible to a committed read.

Errors

Returns CoordinationError::NotHeld when this process holds no epoch for role, and when token is not the epoch it holds. Returns CoordinationError::Fenced when the broker rejects the epoch, and CoordinationError::Producer for every other failure of the write.

fn describe<'life0, 'life1, 'async_trait>( &'life0 self, role: &'life1 Role, ) -> Pin<Box<dyn Future<Output = Result<Option<FencingToken>, CoordinationError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Asks the transaction coordinator which epoch holds role now.

The call sends DescribeTransactions (KIP-664). It joins no group and it takes no epoch, so any process makes it. The result is None when the coordinator answers TRANSACTIONAL_ID_NOT_FOUND, which says that no member has ever held role.

Errors

Returns CoordinationError::Admin when the coordinator lookup fails, or when the coordinator reports a fault other than an unknown role.