Crate krabka_client_producer

Expand description

Idempotent producer client for Apache Kafka in Rust.

It builds on [krabka_client_core] for transport, and adds full idempotent-producer semantics: InitProducerId on connect, per-batch (producer_id, producer_epoch, base_sequence), and retries that re-frame the same RecordBatch so the broker’s dedup catches them.

It also supports transactional, exactly-once production: init_transactions, begin_transaction, which returns a Transaction guard whose commit/abort finishes it, and send_offsets_to_transaction for the consume-process-produce pattern of KIP-447.

§Quick start

use std::time::Duration;

use bytes::Bytes;
use krabka_client_producer::{Acks, Compression, Producer, ProducerRecord};

let producer = Producer::builder()
    .bootstrap("localhost:9092")
    .compression(Compression::Lz4)
    .acks(Acks::All)
    .linger(Duration::from_millis(5))
    .build()
    .await?;

let metadata = producer
    .send(ProducerRecord {
        topic: "my-topic".into(),
        value: Some(Bytes::from("hello")),
        ..Default::default()
    })
    .await?;

producer.flush().await?;
producer.close().await?;

§Pipelined delivery

Use Producer::enqueue to queue records without waiting for each acknowledgement. Each DeliveryHandle returns one producer result.

let mut deliveries = Vec::new();
for record in records {
    deliveries.push(producer.enqueue(record).await?);
}
for delivery in deliveries {
    let metadata = delivery.await?;
}

§Capabilities and boundaries

This crate owns producer-facing semantics: batching, compression, idempotence, retries, per-record partition overrides, transactional RPCs, and send_offsets_to_transaction for consume-process-produce flows. The built-in partitioner follows Kafka’s: keyed records hash to a partition, and keyless records use an adaptive sticky partition (KIP-480, KIP-794). Set ProducerRecord::partition to pin an individual record. Serialization is deliberately owned by the caller: key and value are raw Bytes, so a schema-registry or serde integration can sit on top without constraining the producer API.

Structs§

CompressionLevels: Validated compression levels of the producer (KIP-390).
ConsumerGroupMetadata: The identity that a consumer presents to a transactional producer for KIP-447 offset-commit fencing.
DeliveryHandle: The delivery result of a record queued by crate::Producer::enqueue.
EndTransactionError: Error returned by Transaction::commit, abort, and the OwnedTransaction equivalents.
Header
OwnedTransaction: Same contract as Transaction, but owns an Arc<Producer> instead of borrowing it.
PartitionInfo: One partition of a topic, as Kafka’s PartitionInfo describes it to a custom Partitioner.
PreparedTransactionState: Stable identity of a transaction prepared for external two-phase commit.
PreparedTransactionStateParseError: Error returned when a prepared transaction token is malformed.
Producer
ProducerFlushTimeout: Validated deadline for flushing all buffered and in-flight records.
ProducerRecord
ProducerRetryPolicy: Validated producer retry and transaction timing.
ProducerThroughputPolicy: Validated producer batching and compression policy.
RecordMetadata: What the producer learned about a record that the broker acknowledged.
Transaction: An open transaction, borrowing the Producer that opened it.

Enums§

Acks
Compression
ProducerError
RecordSizeLimit: The producer limit that a record is larger than.

Constants§

DEFAULT_PRODUCER_ACKS: Default producer acknowledgement mode. Kafka’s acks default is all.
DEFAULT_PRODUCER_BATCH_BYTES: Default producer batch size in bytes.
DEFAULT_PRODUCER_BUFFER_MEMORY: Default producer buffer memory in bytes.
DEFAULT_PRODUCER_COMPRESSION: Default producer compression.
DEFAULT_PRODUCER_COMPRESSION_GZIP_LEVEL: Default gzip compression level. Kafka’s compression.gzip.level default is CompressionType.GZIP.defaultLevel(), which is Deflater.DEFAULT_COMPRESSION (-1).
DEFAULT_PRODUCER_COMPRESSION_LZ4_LEVEL: Default lz4 compression level. Kafka’s compression.lz4.level default is 9.
DEFAULT_PRODUCER_COMPRESSION_ZSTD_LEVEL: Default zstd compression level. Kafka’s compression.zstd.level default is 3.
DEFAULT_PRODUCER_DELIVERY_TIMEOUT: Default producer delivery timeout. Kafka’s delivery.timeout.ms default is 120000.
DEFAULT_PRODUCER_FLUSH_TIMEOUT: Default deadline for flushing all buffered and in-flight records.
DEFAULT_PRODUCER_INIT_RETRY_TIMEOUT: Default producer-ID initialization retry timeout.
DEFAULT_PRODUCER_LINGER: Default delay before sending a partial producer batch.
DEFAULT_PRODUCER_MAX_BLOCK: Default longest time that send waits for the metadata of its topic.
DEFAULT_PRODUCER_MAX_IN_FLIGHT: Default cross-partition in-flight request limit.
DEFAULT_PRODUCER_MAX_REQUEST_SIZE: Default largest serialized record that send accepts, in bytes.
DEFAULT_PRODUCER_METADATA_MAX_AGE: Kafka’s producer metadata.max.age.ms default (5 minutes).
DEFAULT_PRODUCER_METADATA_MAX_IDLE: Kafka’s producer metadata.max.idle.ms default (5 minutes).
DEFAULT_PRODUCER_PARTITIONER_ADAPTIVE_PARTITIONING_ENABLE: Default of partitioner_adaptive_partitioning_enable. Kafka’s partitioner.adaptive.partitioning.enable default is true.
DEFAULT_PRODUCER_PARTITIONER_AVAILABILITY_TIMEOUT: Default of partitioner_availability_timeout. Kafka’s partitioner.availability.timeout.ms default is 0, which turns the check off.
DEFAULT_PRODUCER_PARTITIONER_IGNORE_KEYS: Default of partitioner_ignore_keys. Kafka’s partitioner.ignore.keys default is false.
DEFAULT_PRODUCER_PARTITIONER_RACK_AWARE: Default of partitioner_rack_aware. Kafka’s partitioner.rack.aware default is false.
DEFAULT_PRODUCER_RECEIVE_BUFFER: Kafka’s producer receive.buffer.bytes default (32 KiB), half the consumer and admin default.
DEFAULT_PRODUCER_REQUEST_TIMEOUT: Default producer request timeout.
DEFAULT_PRODUCER_RETRIES: Default retries after a batch’s initial send.
DEFAULT_PRODUCER_RETRY_BACKOFF: Default producer retry backoff.
DEFAULT_PRODUCER_RETRY_BACKOFF_MAX: Default upper bound of the exponential retry backoff. Kafka’s retry.backoff.max.ms default is 1000.
DEFAULT_PRODUCER_SEND_BUFFER: Kafka’s producer send.buffer.bytes default (128 KiB).
DEFAULT_PRODUCER_TRANSACTION_TIMEOUT: Default transaction timeout.
MIN_PRODUCER_METADATA_MAX_IDLE: Kafka’s lower bound of metadata.max.idle.ms (ProducerConfig atLeast(5000)).

Traits§

Partitioner: A caller-supplied partitioner.class hook (Kafka’s Partitioner interface).

Functions§

partition_for_key: Pick the partition that Kafka picks for key.