Struct Producer 

pub struct Producer { /* private fields */ }

Implementations

impl Producer

pub async fn start( bootstrap: String, client_id: Option<String>, compression: Compression, compression_gzip_level: i32, compression_lz4_level: i32, compression_zstd_level: i32, enable_idempotence: Option<bool>, acks: Acks, linger: Duration, batch_size: usize, partitioner_ignore_keys: bool, partitioner_adaptive_partitioning_enable: bool, partitioner_availability_timeout: Duration, partitioner: Option<Arc<dyn Partitioner>>, client_rack: Option<String>, partitioner_rack_aware: bool, dns_timeout: Time, client_dns_lookup: ClientDnsLookup, reconnect_backoff: Time, reconnect_backoff_max: Time, connections_max_idle: Time, socket_connection_setup_timeout: Time, socket_connection_setup_timeout_max: Time, dispatch_queue_capacity: usize, frame_max: ByteSize, send_buffer: Option<ByteSize>, receive_buffer: Option<ByteSize>, request_timeout: Duration, flush_timeout: Duration, retries: i32, retry_backoff: Duration, retry_backoff_max: Duration, delivery_timeout: Option<Duration>, init_retry_timeout: Duration, max_in_flight_per_connection: usize, max_block: Duration, metadata_max_age: Duration, metadata_max_idle: Duration, buffer_memory: usize, max_request_size: usize, metadata_recovery_strategy: MetadataRecoveryStrategy, metadata_recovery_rebootstrap_trigger: Time, transactional_id: Option<String>, transaction_timeout: Option<Duration>, transaction_two_phase_commit_enable: bool, security: Option<ClientSecurity>, enable_metrics_push: bool, ) -> Result<Self, ProducerError>

Build a Producer pointed at the given bootstrap address.

The defaults are Kafka’s: acks=All, a linger of 5 ms, and idempotence on. The builder applies the rules of Kafka’s ProducerConfig. When enable_idempotence is not set, retries=0 or an acks other than All turns idempotence off. When it is set to true, those settings fail with ProducerError::InvalidConfig. An idempotent producer accepts at most 5 in-flight requests per connection, and a transactional_id requires idempotence.

enable_metrics_push (Kafka’s enable.metrics.push, default true) pushes the client metrics that a broker’s client metrics subscription names (KIP-714), with a terminating push on close. When it is false, the producer sends neither GetTelemetrySubscriptions nor PushTelemetry.

send fails a record whose serialized size is larger than max_request_size (default 1 MiB, Kafka’s max.request.size) or larger than buffer_memory with ProducerError::RecordTooLarge, and sends no request for it.

compression_gzip_level, compression_lz4_level and compression_zstd_level take Kafka’s defaults (-1, 9 and 3) and ranges (KIP-390), and build fails with ProducerError::InvalidConfig for a level out of range. Each batch of a gzip, lz4 or zstd producer is compressed at the level of its codec. The lz4 codec has no high compression mode, so every lz4 level gives the same output (krabka-io/krabka-protocol#27).

When client_id is not set, the client id is producer-<transactional_id>, or producer-<n> with a process-wide sequence number.

Errors

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

pub fn builder() -> ProducerStartBuilder

Build a Producer pointed at the given bootstrap address.

The defaults are Kafka’s: acks=All, a linger of 5 ms, and idempotence on. The builder applies the rules of Kafka’s ProducerConfig. When enable_idempotence is not set, retries=0 or an acks other than All turns idempotence off. When it is set to true, those settings fail with ProducerError::InvalidConfig. An idempotent producer accepts at most 5 in-flight requests per connection, and a transactional_id requires idempotence.

enable_metrics_push (Kafka’s enable.metrics.push, default true) pushes the client metrics that a broker’s client metrics subscription names (KIP-714), with a terminating push on close. When it is false, the producer sends neither GetTelemetrySubscriptions nor PushTelemetry.

send fails a record whose serialized size is larger than max_request_size (default 1 MiB, Kafka’s max.request.size) or larger than buffer_memory with ProducerError::RecordTooLarge, and sends no request for it.

compression_gzip_level, compression_lz4_level and compression_zstd_level take Kafka’s defaults (-1, 9 and 3) and ranges (KIP-390), and build fails with ProducerError::InvalidConfig for a level out of range. Each batch of a gzip, lz4 or zstd producer is compressed at the level of its codec. The lz4 codec has no high compression mode, so every lz4 level gives the same output (krabka-io/krabka-protocol#27).

When client_id is not set, the client id is producer-<transactional_id>, or producer-<n> with a process-wide sequence number.

Errors

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

impl Producer

pub fn producer_id(&self) -> i64

pub fn producer_epoch(&self) -> i16

pub async fn transactional_identity(&self) -> Option<(i64, i16)>

Give the identity that the transaction coordinator minted for this producer’s transactional.id.

Producer::init_transactions sets this pair, and the broker fences a write that carries an older one. It is the fencing token of the transactional flow.

This is not the pair that Producer::producer_id and Producer::producer_epoch report. Those come from the idempotent InitProducerId that the builder sends with no transactional id, and the coordinator never advances them.

Returns None before init_transactions runs, and for a producer that carries no transactional_id.

pub async fn begin_transaction(&self) -> Result<Transaction<'_>, ProducerError>

Begin a new transaction, returning a borrowed guard that must be finished with Transaction::commit or Transaction::abort.

The caller must call this after init_transactions has completed, and before any transactional send call. It transitions the producer from Ready to InTransaction.

Errors

• ProducerError::NotTransactional — transactional_id was not set.
• ProducerError::InvalidTransactionState — the producer is not in the Ready state. For example, the caller has not yet called init_transactions, or a transaction is already in flight.

pub async fn begin_transaction_owned( self: Arc<Self>, ) -> Result<OwnedTransaction, ProducerError>

Begin a new transaction, returning an owning guard that must be finished with OwnedTransaction::commit or OwnedTransaction::abort.

The semantics are identical to begin_transaction, but the returned guard owns an Arc<Producer> instead of borrowing &self. Use it when the guard must survive across an owned or 'static boundary that a borrow cannot, for example when it is stored behind a dyn Trait object. It mirrors tokio::sync::Mutex::lock_owned.

Errors

Same as begin_transaction.

pub async fn prepare_transaction( &self, ) -> Result<PreparedTransactionState, ProducerError>

Flushes the current transaction and locks it against further writes.

The returned identity has a stable Kafka-compatible string form and can be persisted by an external transaction coordinator. After this method succeeds, only commit, abort, or complete_transaction may finish the transaction.

Errors

• ProducerError::NotTransactional — transactional_id was not set.
• ProducerError::InvalidTransactionState — 2PC is disabled or no transaction is currently in flight.
• ProducerError::FlushTimeout — pending records did not flush before the configured deadline; the transaction remains open and can retry.

pub async fn complete_transaction( &self, prepared: PreparedTransactionState, ) -> Result<(), ProducerError>

Completes the current prepared transaction.

A token matching the prepared transaction commits it. A stale, empty, or otherwise different token aborts it, so recovery never commits work that belongs to another producer incarnation.

Errors

Returns ProducerError::InvalidTransactionState unless a local prepare_transaction or init_transactions_with_keep_prepared established a prepared transaction. Other failures are the same as committing or aborting a normal transaction.

pub async fn init_transactions(&self) -> Result<(), ProducerError>

Initialize the transactional producer.

The caller must call this before any transactional operation. It discovers the transaction coordinator with FindCoordinator, opens a dedicated connection to it, and calls InitProducerId to get a fenced (producer_id, producer_epoch) pair.

Like Kafka’s initTransactions, it blocks while the coordinator is not ready. A lookup that answers a retriable code, such as COORDINATOR_NOT_AVAILABLE from a broker that is still creating __transaction_state, or that fails in transport, goes out again. The producer-ID initialization retry timeout bounds the lookup and InitProducerId together.

Errors

• ProducerError::NotTransactional — transactional_id was not set.
• ProducerError::InvalidTransactionState — called while a transaction is in flight.
• ProducerError::FencedProducer — the broker returned INVALID_PRODUCER_EPOCH (47) or PRODUCER_FENCED (90), now or in an earlier transactional operation.
• ProducerError::FatalTransactionError — the coordinator answered with a fatal code, now or in an earlier transactional operation. A FindCoordinator answer of TRANSACTIONAL_ID_AUTHORIZATION_FAILED, or of a code that Kafka’s FindCoordinatorHandler does not expect, is fatal. Kafka’s initTransactions does not clear a fatal error either: close the producer.
• ProducerError::Server — an abortable code (TRANSACTIONAL_ID_AUTHORIZATION_FAILED or CLUSTER_AUTHORIZATION_FAILED from InitProducerId, or TRANSACTION_ABORTABLE), or a retriable code at the retry deadline. The call can be made again.
• ProducerError::Client — transport-level failure, at once when it repeats on every attempt, such as a failed authentication, or at the retry deadline.

pub async fn init_transactions_with_keep_prepared( &self, keep_prepared: bool, ) -> Result<(), ProducerError>

Initializes the producer and optionally preserves an ongoing prepared transaction from an earlier producer incarnation.

When keep_prepared is true and the broker reports an ongoing transaction, the producer enters the prepared state. The only allowed transaction-ending operations are commit, abort, and complete_transaction. The broker returns both identities: the original identity becomes the persisted comparison token, while the newly staged identity is used for the eventual EndTxn.

Errors

The errors are the same as init_transactions.

pub async fn send_offsets_to_transaction( &self, offsets: impl IntoIterator<Item = ((String, i32), i64)>, group_meta: &ConsumerGroupMetadata, ) -> Result<(), ProducerError>

Enroll a consumer group’s offsets in the current transaction, and fence zombie producers with the supplied ConsumerGroupMetadata, as KIP-447 defines.

It first waits for metadata that names the topics of offsets, as Kafka’s producer does, so that TxnOffsetCommit can name each topic by its id. Then it does two broker round-trips:

1. AddOffsetsToTxn to the transaction coordinator. This registers the group offset commit as part of the ongoing transaction.
2. TxnOffsetCommit to the group coordinator. This commits the actual offsets transactionally. It carries the generation, member and instance of group_meta, so the coordinator can fence stale producers.

Each request goes out again after a transport loss and after a retriable code, until the producer-ID initialization retry timeout ends. A code that shows the coordinator moved also finds the coordinator again. The rules follow Kafka’s AddOffsetsToTxnHandler and TxnOffsetCommitHandler.

Errors

• ProducerError::NotTransactional — transactional_id was not set.
• ProducerError::InvalidTransactionState — there is no cached transaction coordinator. Call init_transactions first.
• ProducerError::FencedProducer — the coordinator answered INVALID_PRODUCER_EPOCH (47) or PRODUCER_FENCED (90).
• ProducerError::Server — any other broker error code. A code that Kafka calls abortable also makes every later commit fail, so the application must abort the transaction.
• ProducerError::MetadataTimeout — no metadata response named a topic of offsets within max_block. Kafka’s sendOffsetsToTransaction throws TimeoutException.
• ProducerError::Client — transport-level failure.

pub async fn send( &self, record: ProducerRecord, ) -> Result<RecordMetadata, ProducerError>

Send a record and wait for its delivery result.

Use Self::enqueue to queue multiple records before waiting for delivery. Once queued, dropping this future does not cancel the record.

Errors

Returns enqueue or delivery failures as ProducerError.

pub async fn enqueue( &self, record: ProducerRecord, ) -> Result<DeliveryHandle, ProducerError>

Enqueue a record and return a future that resolves when the broker acks, or when the producer fences or closes.

A record with a negative partition or timestamp fails at once with ProducerError::InvalidPartition or ProducerError::InvalidTimestamp, and the producer sends nothing.

A transactional producer fails the record at once, and sends nothing, when its transaction holds an error: the abortable error until the application aborts, and the fatal error for good. Kafka’s KafkaProducer.doSend fails the same way in TransactionManager.maybeAddPartition.

Waits for metadata and buffer space for at most max_block before queueing. Dropping the returned handle does not cancel delivery.

Errors

Returns validation, producer state, metadata, or buffer allocation failures before queueing. Delivery errors are returned by the handle.

pub async fn close(self) -> Result<(), ProducerError>

Errors

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

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

The client instance id that the broker assigned for metrics push (KIP-714), as Kafka’s KafkaProducer.clientInstanceId returns it.

The call waits up to timeout for the first GetTelemetrySubscriptions response. It returns Ok(None) when none has come by then, as Kafka returns null. A zero timeout does not wait.

Errors

Returns ClientError::TelemetryDisabled when enable_metrics_push is false, and ClientError::InvalidArgument for a negative timeout, as Kafka throws an IllegalStateException and an IllegalArgumentException.

pub async fn flush(&self) -> Result<(), ProducerError>

Errors

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

Trait Implementations

impl Debug for Producer

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.