Producer

Batching, compression, partitioning, idempotence and exactly-once transactions.

Overview

The krafka producer is an async-native, high-performance message producer for Apache Kafka. Key features include:

• Async/await API with Tokio
• Automatic batching for throughput
• Multiple compression codecs (gzip, snappy, lz4, zstd)
• Flexible partitioning strategies
• Automatic metadata refresh
• Interceptor hooks for observability

Basic Usage

use krafka::producer::Producer;
use krafka::error::Result;

#[tokio::main]
async fn main() -> Result<()> {
    let producer = Producer::builder()
        .bootstrap_servers("localhost:9092")
        .build()
        .await?;

    // Simple send
    producer.send("topic", None, Some(b"value")).await?;

    // Send with key (for partitioning)
    producer.send("topic", Some(b"key"), Some(b"value")).await?;

    producer.close().await;
    Ok(())
}

Authentication

Connect to secured Kafka clusters using SASL or TLS:

use krafka::producer::Producer;

// SASL/SCRAM-SHA-256
let producer = Producer::builder()
    .bootstrap_servers("broker:9093")
    .sasl_scram_sha256("username", "password")
    .build()
    .await?;

// AWS MSK IAM
use krafka::auth::AuthConfig;
let auth = AuthConfig::aws_msk_iam("access_key", "secret_key", "us-east-1");
let producer = Producer::builder()
    .bootstrap_servers("broker:9094")
    .auth(auth)
    .build()
    .await?;

See the Authentication Guide for all supported mechanisms.

Producer Configuration

Acknowledgments

Control durability vs. latency with the acks setting:

use krafka::producer::{Producer, Acks};

// Fire and forget (lowest latency, risk of data loss)
let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::None)
    .build()
    .await?;

// Wait for leader (balanced)
let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::Leader)
    .build()
    .await?;

// Wait for all in-sync replicas (highest durability)
let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::All)
    .build()
    .await?;

Compression

Choose the right compression codec for your workload:

use krafka::producer::Producer;
use krafka::protocol::Compression;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .compression(Compression::Lz4)  // Fast compression
    .build()
    .await?;

The default compression convenience feature enables the pure-Rust codecs: gzip, snappy, and LZ4. Zstd remains available through the explicit zstd or compression-all feature because it requires a C toolchain via zstd-sys.

To trim binary size further, disable defaults and select only the codecs you need:

# Only the codecs you need. `--no-default-features` also drops the default
# `ring` TLS backend, so a crypto backend must be named explicitly.
cargo add krafka --no-default-features --features lz4,ring

# Or every codec, including zstd:
cargo add krafka --features compression-all

Compression level

Gzip and Zstd accept a level. Snappy has none in its format, and krafka encodes LZ4 with lz4_flex, whose frame encoder exposes none.

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .compression(Compression::Zstd)
    .compression_level(Some(1))   // favour throughput over ratio
    .build()
    .await?;

Setting a level alongside a codec that takes none is rejected at build time, as is a level outside the codec's range, and per-topic codec overrides are validated against it too. Neither case is silently ignored: a tuning knob that quietly does nothing is how a deployment ships believing it was tuned.

The level applies to the plain producer and to the TransactionalProducer alike, enforced by one shared validator, so a codec check cannot exist on one producer and not the other.

Higher is not better. zstd's output size is not monotonic in level — the match-finding strategy changes as levels rise, and on realistic record payloads level 3 can be larger than level 1. Above roughly level 9 the CPU cost climbs much faster than the byte savings, so on a throughput-bound producer the high levels are usually a net loss. Measure against your own payloads.

Batching

Batching improves throughput by combining multiple messages:

use krafka::producer::Producer;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .batch_size(65536)                      // Max bytes per batch (64KB)
    .linger(Duration::from_millis(5))       // Wait up to 5ms for more messages
    .build()
    .await?;

What linger actually controls

Every send goes through the record accumulator — there is no separate unbatched path, at any linger setting. Records are accumulated per partition and a batch is dispatched when:

• it reaches batch_size bytes, or
• the linger window expires, or
• the partition has no batch in flight (this is the linger = 0 case).

linger is therefore "how long may a batch wait for company", not "may this producer batch at all". At linger = 0 the first record goes out immediately — nothing is on the wire, so there is nothing to wait for — and the records that arrive during that round trip coalesce into the next batch, which is dispatched the instant the acknowledgement lands. You pay no added latency and still get batching under load. This is what linger.ms = 0 means in the Java client, and it is worth an order of magnitude: 200 concurrent sends to one partition leave krafka as 3 Produce requests, not 200.

Raise linger when you want to trade a bounded amount of latency for larger batches even when the producer is not saturated — a bursty, low-rate publisher will not fill a batch on its own.

One batch per partition on the wire

krafka keeps exactly one batch per partition in flight, and batches take their turn in the order the accumulator sealed them. Different partitions proceed concurrently and are never serialised against each other.

That is a stronger guarantee than the Java client's, and it is why krafka has no max.in.flight.requests.per.connection knob to get wrong: sequence order and wire order cannot diverge, so idempotent production needs no "≤ 5 in flight" rule, and a retry cannot reorder a partition. The per-connection in-flight ceiling that does exist is a transport concern — see TransportConfig::max_in_flight_requests.

Note: batch_size must be at least 1. Setting batch_size to 0 will cause the builder to return a configuration error.

Request Size Cap

Use max_request_size when you want the producer to fail locally before sending a Produce request frame larger than your broker or network budget:

use krafka::producer::Producer;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .max_request_size(1 * 1024 * 1024)      // 1 MiB encoded Produce frame cap
    .build()
    .await?;

The producer encodes the final request using the negotiated Produce API version and rejects frames that exceed max_request_size before any broker I/O. The default is 100 MiB, matching Kafka's protocol request-size ceiling. Leave some headroom between batch_size and max_request_size for request headers and topic names; the builder rejects configurations where batch_size > max_request_size. The same knob is available on TransactionalProducer::builder().

// High-throughput configuration
let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .batch_size(131072)                     // 128KB batches
    .linger(Duration::from_millis(10))      // Wait up to 10ms
    .compression(Compression::Lz4)          // Fast compression
    .build()
    .await?;

// Low-latency configuration (this is also the default)
let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .linger(Duration::ZERO)                 // Never wait; still coalesces under load
    .build()
    .await?;

Memory Backpressure

The producer limits memory usage to prevent unbounded growth under high load:

use krafka::producer::Producer;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .buffer_memory(64 * 1024 * 1024)        // 64MB buffer limit
    .max_block(Duration::from_secs(30))     // Total time send() may block
    .build()
    .await?;

Once a record is admitted it holds a share of the producer memory budget until it is acknowledged or fails. If memory is unavailable, send() blocks the caller before returning an error — the Kafka Java client's max.block.ms behaviour, which prevents both OOM conditions and unnecessary record loss under bursty load. max_block is one budget for the whole call, so time already spent resolving the topic is deducted from this wait.

That wait is charged against delivery_timeout: the clock starts when you call send(), not when the record reaches a batch, so a record that spent 30 s blocked on backpressure does not then get a full fresh delivery budget.

Flushing

Call flush() whenever you need a durability barrier over records that have already been handed to the producer:

// Send multiple records
for i in 0..100 {
    producer.send("topic", Some(format!("key-{}", i).as_bytes()), Some(b"value")).await?;
}

// Ensure all records are sent before closing
producer.flush().await?;
producer.close().await;

Tombstones and Compacted Topics

On a cleanup.policy=compact topic, a record with a null value is a tombstone: it marks its key for deletion. A null value is not an empty one — the record format encodes null as a -1 length prefix and zero-length as 0, and compaction deletes on the first while keeping the second. krafka models the distinction as Option<Bytes> on both sides of the wire.

use krafka::producer::ProducerRecord;

// A tombstone needs a key: a null value on a keyless record deletes nothing.
producer
    .send_record(ProducerRecord::tombstone("users", "user-42"))
    .await?;

// The same thing, without building a record.
producer.send("users", Some(b"user-42"), None).await?;

// A zero-length value is NOT a tombstone — compaction keeps this record.
producer.send("users", Some(b"user-42"), Some(b"")).await?;

without_value() turns an existing record into one, keeping its key and headers; with_value() reverses that. is_tombstone() reports whether a record has a key and no value, using the same rule as ConsumerRecord::is_tombstone().

Two things worth knowing:

• A configured value_serializer is skipped for a tombstone, and a key_serializer for a null key. Framing a null value would emit a short record that compaction reads as ordinary data. See Serializer.
• The tombstone must share a partition with the records it retires, since compaction runs per partition. The default partitioner hashes the key, so reusing the key is enough — do not pin partition on one and not the other.

Null header values

Header values carry the same distinction, as Vec<(String, Option<Bytes>)>:

use krafka::producer::ProducerRecord;

let record = ProducerRecord::new("events", b"payload".to_vec())
    .with_header("X-Source", &b"api"[..])   // an ordinary header value
    .with_null_header("X-Flag");            // null, not zero-length

On the read side, ConsumerRecord::is_tombstone() classifies a record and CompactedTable applies the semantics — the key is removed from the table and reported as a TableChange with is_delete().

Partitioning

Default Partitioner

The default partitioner uses murmur2 hashing (Java-compatible) for keyed messages and round-robin for null keys:

// Messages with the same key go to the same partition
producer.send("topic", Some(b"user-123"), Some(b"event1")).await?;
producer.send("topic", Some(b"user-123"), Some(b"event2")).await?;  // Same partition

// Messages without keys are distributed round-robin
producer.send("topic", None, Some(b"event")).await?;

Custom Partitioners

krafka provides several built-in partitioners:

use krafka::producer::{
    DefaultPartitioner,
    RoundRobinPartitioner,
    StickyPartitioner,
    HashPartitioner,
};

// Round-robin: ignores keys, distributes evenly
let partitioner = RoundRobinPartitioner::new();

// Sticky: sticks to one partition, auto-advances after batch_threshold records (default 100)
let partitioner = StickyPartitioner::new();

// Sticky with custom batch threshold
let partitioner = StickyPartitioner::with_batch_threshold(500);

// Hash: uses Rust's default hasher instead of murmur2
let partitioner = HashPartitioner::new();

Implementing Custom Partitioners

use krafka::producer::Partitioner;
use krafka::PartitionId;

struct RegionPartitioner {
    region_to_partition: std::collections::HashMap<String, PartitionId>,
}

impl Partitioner for RegionPartitioner {
    fn partition(
        &self,
        topic: &str,
        key: Option<&[u8]>,
        partition_count: usize,
    ) -> PartitionId {
        if let Some(key) = key {
            if let Ok(region) = std::str::from_utf8(key) {
                if let Some(&partition) = self.region_to_partition.get(region) {
                    return partition % partition_count as i32;
                }
            }
        }
        // Fallback to first partition
        0
    }
}

Metadata Topic Cache TTL

krafka caches topic metadata between refreshes. During a partial refresh — one that names specific topics — entries that have been idle for longer than the TTL are evicted, so topic churn does not grow the cache indefinitely. The default is 5 minutes, matching Java's metadata.max.idle.ms.

Idle means nothing has addressed the topic: producing to it, resolving a leader for it, asking for its partition count, or naming it in a metadata refresh all reset the timer. A topic whose metadata is still current survives regardless.

use krafka::producer::Producer;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .metadata_topic_cache_ttl(Duration::from_secs(600))
    .build()
    .await?;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .disable_metadata_topic_cache_ttl()
    .build()
    .await?;

A full metadata refresh still replaces the cache unconditionally.

Topic Resolution

send() to a topic the cache does not hold fetches metadata for it and retries until it resolves or the max_block budget expires — the equivalent of KafkaProducer.waitOnMetadata. A topic that was never fetched, one evicted as idle, and one still being created all resolve this way.

A topic the cluster will not resolve within max_block fails with the broker's own reason:

use krafka::error::{ErrorCode, KrafkaError};
use krafka::producer::Producer;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .build()
    .await?;

match producer.send("maybe-missing", None, Some(b"v")).await {
    Ok(metadata) => println!("wrote to partition {}", metadata.partition),
    Err(KrafkaError::Broker { code: ErrorCode::TopicAuthorizationFailed, .. }) => {
        // The topic exists; this principal may not write to it.
    }
    Err(KrafkaError::Broker { code: ErrorCode::UnknownTopicOrPartition, .. }) => {
        // The cluster does not have this topic.
    }
    Err(e) => eprintln!("send failed: {e}"),
}

max_block is one budget for the whole call: time spent resolving the topic is deducted from the wait for buffer memory, so send() never blocks longer than max_block in total.

partitions_for — the equivalent of KafkaProducer.partitionsFor — inspects a topic directly, fetching on a cache miss under the same budget:

use krafka::producer::Producer;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .build()
    .await?;

for partition in producer.partitions_for("events").await? {
    println!("partition {} led by {}", partition.partition, partition.leader);
}

An explicit partition is range-checked too: partition 7 of a 2-partition topic is rejected by send().

Letting the broker create the topic

allow_auto_create_topics sets allow.auto.create.topics on the metadata requests the send path issues, so the broker creates a missing topic on demand:

use krafka::producer::Producer;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .allow_auto_create_topics(true)
    .build()
    .await?;

The broker must also run with auto.create.topics.enable=true; the client flag only says it is willing.

Off by default, unlike the Java producer, which always asks for auto-creation. A typo'd topic name that silently materialises a real topic reports nothing until the traffic is found missing from the topic it was meant for. Turn it on for development and test clusters.

The flag lives on the metadata cache, so a client sharing a KrafkaClient's metadata inherits that client's setting instead.

Error Handling

Record Validation

Before sending, each ProducerRecord is validated against Kafka wire-format limits:

• Topic name: max 32,767 bytes (i16 limit)
• Key: max 2,147,483,647 bytes (i32 limit)
• Value: max 2,147,483,647 bytes (i32 limit)
• Header keys: max 2,147,483,647 bytes (i32 limit)
• Header values: max 2,147,483,647 bytes (i32 limit)

Oversized data returns a descriptive KrafkaError::protocol error instead of panicking.

Built-in Retry

The producer automatically retries transient failures (e.g., NotLeaderForPartition, network timeouts) using the configured retry policy. On each retriable error, the producer refreshes metadata to discover the new partition leader before retrying with exponential backoff.

Configure retries via the builder:

use krafka::producer::Producer;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .retries(5)                                      // Max retry attempts; defaults to u32::MAX
    .retry_backoff(Duration::from_millis(100))        // Initial backoff
    .build()
    .await?;

// send() automatically retries on transient failures
producer.send("topic", None, Some(b"value")).await?;

Delivery Timeout

The delivery_timeout setting (analogous to the Java client's delivery.timeout.ms) caps the total time from when a record enters the producer to when it must be acknowledged. This includes time spent in the accumulator's linger window, backpressure waits, and all retry attempts.

use krafka::producer::Producer;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .delivery_timeout(Duration::from_secs(120))  // Total delivery budget
    .linger(Duration::from_millis(5))             // Batching window
    .retries(u32::MAX)                            // Retry until timeout
    .build()
    .await?;

The producer defaults to delivery_timeout = 120s and retries = u32::MAX, so transient failures are retried until the delivery budget is exhausted. Backoff durations are clamped to the remaining budget so the producer does not overshoot. If the budget is exhausted, the send fails immediately regardless of the remaining retry count.

Note: By default linger is 0, so a record never waits for a batching window and the delivery timeout is essentially backpressure + network + retry time. With linger > 0, add the maximum linger window to the budget.

Manual Retry

For additional retry control beyond the built-in behavior, handle errors explicitly:

use krafka::producer::Producer;
use krafka::error::{KrafkaError, Result};

async fn send_with_retry(
    producer: &Producer,
    topic: &str,
    key: Option<&[u8]>,
    value: &[u8],
    max_retries: u32,
) -> Result<()> {
    let mut attempts = 0;
    
    loop {
        match producer.send(topic, key, Some(value)).await {
            Ok(metadata) => {
                println!("Sent to {}:{}", metadata.partition, metadata.offset);
                return Ok(());
            }
            Err(e) if e.is_retriable() && attempts < max_retries => {
                println!("Send failed (attempt {}): {}", attempts + 1, e);
                attempts += 1;
                tokio::time::sleep(std::time::Duration::from_millis(100 * attempts as u64)).await;
            }
            Err(e) => return Err(e),
        }
    }
}

Using RetryPolicy

For more sophisticated retry handling with exponential backoff:

use krafka::producer::{Producer, RetryPolicy, RetryContext};
use krafka::error::Result;

async fn send_with_policy(
    producer: &Producer,
    topic: &str,
    value: &[u8],
) -> Result<()> {
    let policy = RetryPolicy::new()
        .with_max_retries(5)
        .with_initial_backoff(std::time::Duration::from_millis(100))
        .with_max_backoff(std::time::Duration::from_secs(10))
        .with_backoff_multiplier(2.0)
        .with_jitter_factor(0.1);  // Add 10% jitter to prevent thundering herd
    
    let mut ctx = RetryContext::new(policy, "send_message");
    
    loop {
        match producer.send(topic, None, Some(value)).await {
            Ok(metadata) => {
                ctx.record_success();
                return Ok(());
            }
            Err(e) => {
                if let Some(backoff) = ctx.record_failure(&e) {
                    ctx.wait(backoff).await;
                } else {
                    return Err(e);
                }
            }
        }
    }
}

Performance Tips

High Throughput

For maximum throughput:

use krafka::producer::{Producer, Acks};
use krafka::protocol::Compression;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::Leader)                     // Don't wait for all replicas
    .compression(Compression::Lz4)           // Fast compression
    .batch_size(1048576)                     // 1MB batches
    .linger(Duration::from_millis(10))       // Allow batching
    .build()
    .await?;

Low Latency

For minimum latency:

use krafka::producer::{Producer, Acks};
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::None)                        // Don't wait for acks
    .batch_size(1)                           // No batching
    .linger(Duration::ZERO)                  // Send immediately
    .build()
    .await?;

acks=0 also gives up quota feedback. The broker sends no response, so there is no throttle_time_ms to read (KIP-219) and a producer sending only acks=0 traffic never learns it is being throttled. A throttle learned from any other API on the same connection is still honoured, but a pure acks=0 client keeps writing at full rate until the broker mutes the channel itself. Combined with the loss of delivery confirmation, acks=0 gives up more than durability alone — prefer acks=1 unless you have measured that the difference matters.

Durability

For maximum durability:

use krafka::producer::{Producer, Acks};
use krafka::protocol::Compression;
use std::time::Duration;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .retries(10)                             // Retry on failure
    .build()
    .await?;

Idempotent by default (KIP-679): Since Kafka 3.0, idempotent production is the default. The regular Producer now obtains a Producer ID via InitProducerId at startup, tracks sequence numbers per partition, and de-duplicates retries automatically. acks = All is required when idempotent is enabled. Unlike the Java client and librdkafka there is no in-flight-request limit to observe: krafka keeps exactly one batch per partition on the wire, so sequence order and wire order cannot diverge and KIP-679's "≤ 5 in flight" rule has nothing to protect. The InitProducerId call retries on retriable errors (e.g. CoordinatorLoadInProgress) with exponential backoff, rotating through available brokers on each attempt.

Error handling:

• OutOfOrderSequenceNumber triggers a sequence reset and batch rebuild before retrying.
• DuplicateSequenceNumber is treated as success (broker already committed the batch; idempotent dedup worked). The returned offset is -1 since the broker does not echo the original offset for duplicates.
• Multi-record batches acknowledge the last sequence (base + count − 1), matching the Kafka Java client's ProducerBatch.lastSequence() semantics.

For cross-session exactly-once semantics (transactions), use TransactionalProducer.

Concurrency control

There is nothing to configure for per-partition ordering: the accumulator's dispatch FIFO already permits exactly one batch per partition on the wire, and batches take their turn in seal order. A retry cannot reorder a partition, and an idempotent producer's sequence order always matches its wire order.

Concurrency across partitions is bounded in two independent places:

• Per connection — TransportConfig::max_in_flight_requests caps how many requests may be outstanding on a single broker socket.
• Per producer — the accumulator caps how many batch-send tasks run at once, so an overlapping burst of linger waves cannot spawn unboundedly.

Both are transport concerns. Neither affects ordering, because ordering is not bought with concurrency limits here.

Graceful Shutdown

Always close producers properly to flush pending messages. The close() method is a barrier over all started sends, not just batches still resident in the accumulator. It blocks new sends, waits for buffered and already-in-flight work to finish, then tears down connections. Calling close() more than once is a no-op:

use krafka::producer::Producer;

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .build()
    .await?;

// ... send messages ...

// Flush and close — waits for all in-flight batches to complete
producer.flush().await?;
producer.close().await;

If you need a bounded shutdown window, use close_with_timeout() instead. On timeout, krafka tears down the connection pool and returns a timeout error, causing any remaining in-flight work to fail fast instead of hanging shutdown indefinitely:

use std::time::Duration;

producer.close_with_timeout(Duration::from_secs(10)).await?;

Transactional Producer

For exactly-once semantics across multiple partitions and topics, use the TransactionalProducer. This is the recommended approach for idempotent and exactly-once production.

The transactional producer:

• Automatically obtains a Producer ID (PID) and epoch from the broker via InitProducerId
• Sets producer_id, producer_epoch, and base_sequence on every record batch
• Marks batches as transactional (attribute bit 0x10)
• Tracks sequence numbers per topic-partition for idempotent delivery

Basic Usage

use krafka::producer::TransactionalProducer;
use krafka::error::Result;

#[tokio::main]
async fn main() -> Result<()> {
    // Create transactional producer with unique ID
    let producer = TransactionalProducer::builder()
        .bootstrap_servers("localhost:9092")
        .transactional_id("my-unique-transaction-id")
        .build()
        .await?;

    // Initialize transactions (once per producer)
    producer.init_transactions().await?;

    // Start transaction
    producer.begin_transaction()?;

    // Send messages atomically
    producer.send("topic-a", Some(b"key1"), Some(b"value1")).await?;
    producer.send("topic-b", Some(b"key2"), Some(b"value2")).await?;

    // Commit transaction (all or nothing)
    producer.commit_transaction().await?;

    Ok(())
}

Configuration

TransactionalProducerBuilder mirrors ProducerBuilder setter for setter: compression and compression levels, delivery timeout, interceptors, a dead-letter queue, a state store, with_client, the metadata cache TTLs, and the synchronous build_config() terminal. tests/builder_surface.rs asserts that at compile time, so the two builders cannot drift apart.

use krafka::producer::TransactionalProducer;
use krafka::protocol::Compression;
use std::time::Duration;

let producer = TransactionalProducer::builder()
    .bootstrap_servers("localhost:9092")
    .transactional_id("order-processor-1")
    .client_id("my-app")
    .transaction_timeout(Duration::from_secs(60))          // coordinator's deadline
    .request_timeout(Duration::from_secs(30))
    .delivery_timeout(Duration::from_secs(45))             // bound on one batch in flight
    .compression(Compression::Zstd)
    .compression_level(Some(1))
    .build()
    .await?;

Two setters are deliberately absent, because the transactional protocol fixes both:

Delivery timeout

delivery_timeout bounds how long one batch may spend in flight, including batching, retries and backoff. It matters more here than on the plain producer: a batch that keeps retrying holds the transaction open, and an open transaction blocks every read_committed consumer at its first offset.

Keep it at or below transaction_timeout — the coordinator aborts at that point regardless. build() and build_config() warn when the two disagree.

Validating without a broker

build_config() runs exactly the checks build() runs and returns the validated TransactionalProducerConfig without connecting — for a validate-config subcommand, a startup check, or a unit test:

let config = TransactionalProducer::builder()
    .bootstrap_servers("localhost:9092")
    .transactional_id("order-processor-1")
    .compression(Compression::Zstd)
    .compression_level(Some(1))
    .build_config()?;      // no cluster required

assert_eq!(config.compression_level(), Some(1));

Flushing

flush() dispatches every buffered record and waits for the in-flight sends to complete. You do not need it before commit_transaction(), which flushes first and must — a commit marker written while records were still buffered would leave them outside the transaction they were sent in.

It is there for two other reasons: forcing buffered records onto the wire mid-transaction so their failures surface with the record's context rather than at commit time, and writing code generic over "a producer" without special-casing which of the two you hold.

Unlike Producer::flush, it does not make the records visible to a read_committed consumer — only commit_transaction() does.

Authentication

Connect a transactional producer to secured Kafka clusters:

use krafka::producer::TransactionalProducer;

// SASL/SCRAM-SHA-256 over cleartext (development only)
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9093")
    .transactional_id("my-txn-id")
    .sasl_scram_sha256("username", "password")
    .build()
    .await?;

// SASL_SSL + SCRAM-SHA-512 — what a managed cluster almost always wants
use krafka::auth::{AuthConfig, TlsConfig};
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9093")
    .transactional_id("my-txn-id")
    .auth(AuthConfig::sasl_scram_sha512_ssl("username", "password", TlsConfig::new()))
    .build()
    .await?;

// Or use AuthConfig for advanced auth (e.g., AWS MSK IAM)
use krafka::auth::AuthConfig;
let auth = AuthConfig::aws_msk_iam("access_key", "secret_key", "us-east-1");
let producer = TransactionalProducer::builder()
    .bootstrap_servers("broker:9094")
    .transactional_id("my-txn-id")
    .auth(auth)
    .build()
    .await?;

See the Authentication Guide for all supported mechanisms.

Transaction Lifecycle

1. Initialize: Call init_transactions() once when producer starts
2. Begin: Call begin_transaction() to start a new transaction
3. Send: Send messages with send() or send_record()
4. End: Call commit_transaction() or abort_transaction()
5. Close: Call close() when done — aborts any active transaction and cleans up resources

// Error handling with abort
producer.begin_transaction()?;

match do_work(&producer).await {
    Ok(()) => producer.commit_transaction().await?,
    Err(e) => {
        producer.abort_transaction().await?;
        return Err(e);
    }
}

// When finished with the producer, always close it
producer.close().await;

send_offsets_to_transaction must complete inside the transaction. A commit waits for an offset commit that is already in flight, so the EndTxn marker is never written around one. An offset commit started after the commit has begun is refused with the same Committing error as a send. Both orderings are safe; there is no arrangement in which the offsets land outside the transaction.

A commit closes the transaction before it drains it. The moment commit_transaction() is entered it transitions out of InTransaction, so any concurrent send() from another task is refused with an InvalidState error naming the Committing state. This is deliberate: a record admitted after the drain had begun would still be buffered when EndTxn went out, and would land in the next transaction — vanishing if that one aborted. If you share a TransactionalProducer across tasks, treat that error as "the transaction closed under me" and retry the record in the next one.

Never abort after a commit times out. If commit_transaction() fails with a timeout or a connection loss, the coordinator may already have committed — the response was simply lost. Aborting then is the KAFKA-17754 trigger: the delayed EndTxn can be applied to a later transaction and tear it. The Java client's documentation recommends aborting in this case; that advice predates KAFKA-17754 and krafka deliberately does not follow it.

krafka enforces this rather than relying on you to remember it. A commit whose outcome is unknown moves the producer to TransactionState::CommitIndeterminate, from which:

• abort_transaction() returns an error explaining why, instead of performing an abort that could silently corrupt data;
• close() leaves the transaction alone rather than auto-aborting it, and logs that it did so;
• commit_transaction() may be retried — EndTxn is idempotent for the same producer id and epoch, so a duplicate commit either lands or is recognised by the coordinator as the one it already applied.

If you cannot retry, drop the producer. The coordinator resolves the transaction on its own via transaction.timeout.ms.

A commit that fails with a broker error code is different: the coordinator answered and declined, so the transaction is definitively still open and the producer returns to InTransaction, where aborting is safe.

Graceful Shutdown (Transactional)

Always close transactional producers properly. The close() method:

• Blocks new sends and waits for already-started transactional produce requests to finish
• Aborts any active transaction to avoid dangling open transactions on the broker
• Transitions the producer to FatalError state, preventing further use
• Closes the underlying connection pool
• Is idempotent — calling it more than once is a no-op

// Graceful shutdown
producer.close().await;
// Producer is no longer usable after close()

For bounded shutdown windows, close_with_timeout() provides the same semantics with an explicit deadline:

use std::time::Duration;

producer.close_with_timeout(Duration::from_secs(10)).await?;

Built-in Retry Logic

The transactional producer automatically retries sends on transient failures:

• Uses the shared RetryPolicy (default: 3 retries, exponential backoff with jitter)
• Metadata is refreshed on transient errors before retrying
• OutOfOrderSequenceNumber errors trigger a sequence number reset and batch rebuild with a fresh sequence before retrying
• Sequence numbers and the batch are allocated once and reused across normal retries to maintain idempotent semantics
• Non-retriable errors (auth failures, invalid topics) fail immediately

Coordinator Re-discovery

All coordinator RPCs (InitProducerId, AddPartitionsToTxn, AddOffsetsToTxn, EndTxn) automatically handle coordinator failover:

• On NotCoordinator, CoordinatorNotAvailable, or CoordinatorLoadInProgress the cached coordinator is invalidated and a fresh FindCoordinator is issued before retrying.
• Network and timeout errors to the coordinator trigger the same invalidation + re-discovery flow.
• The retry uses the producer's RetryPolicy for exponential backoff between attempts.
• Fatal errors (TransactionCoordinatorFenced, ProducerFenced, InvalidProducerEpoch, InvalidTxnState) are never retried.
• If no coordinator is cached (e.g. after invalidation), coordinator_connection() auto-discovers one transparently before returning the connection.

KIP-890 Epoch Bumping (Kafka 3.7+)

Kafka 3.7+ brokers implement KIP-890 epoch bumping: after every successful EndTxn (commit or abort) the broker increments the producer epoch and returns the new ProducerId and ProducerEpoch in the EndTxn v4+ response. krafka reads these fields and automatically applies them to the local identity, so subsequent AddPartitionsToTxn requests use the correct epoch.

For brokers that do not support EndTxn v4+ (Kafka < 3.7), the response omits these fields and krafka continues with the unchanged epoch — the pre-KIP-890 protocol is used transparently.

Negotiated transaction version

krafka negotiates one protocol level for the cluster and reports it at init_transactions():

Two rules govern the negotiation, and both exist because getting them wrong is silent:

• The level alone is not evidence. Finalized features are cluster-wide metadata and can be observed before every broker has restarted into a build that serves the matching API versions. krafka additionally requires the API versions each level depends on — Produce, TxnOffsetCommit and EndTxn for TV2, InitProducerId v6 for TV3. A broker that cannot encode enable2Pc does not reject it; the field is simply absent, and the coordinator applies transaction.max.timeout.ms to a transaction the caller believes is exempt.
• The cluster level is the minimum across brokers. One lagging broker during a rolling upgrade holds the whole cluster at the level it can serve, so 2PC never turns on before every broker can honour it.

two_phase_commit(true) on a cluster below TV3 fails at init_transactions() with a message naming the feature level, the API version and the ACL required — rather than surfacing the broker's bare UNSUPPORTED_VERSION.

Persisting Producer State

ProducerStateStore is a hook for saving and restoring the producer's identity — its producer ID, epoch and per-partition sequence numbers — across restarts. Attach one with state_store() on either producer builder:

use krafka::producer::{ProducerIdentitySnapshot, ProducerStateStore, TransactionalProducer};

struct FileStateStore {
    path: std::path::PathBuf,
}

impl ProducerStateStore for FileStateStore {
    async fn load(&self) -> krafka::Result<Option<ProducerIdentitySnapshot>> {
        // Read and deserialise the snapshot; `Ok(None)` on first run.
        Ok(None)
    }

    async fn store(&self, snapshot: &ProducerIdentitySnapshot) -> krafka::Result<()> {
        // Persist it. Errors are logged at WARN and never fail the send.
        Ok(())
    }
}

let producer = TransactionalProducer::builder()
    .bootstrap_servers("localhost:9092")
    .transactional_id("orders-processor-1")
    .state_store(FileStateStore { path: "/var/lib/app/producer.json".into() })
    .build()
    .await?;

load() is called once during build(); store() is called after each successful batch acknowledgement.

A restored snapshot is only honoured when it is safe to honour. krafka applies it only if the stored producer_id and producer_epoch match what the broker returned from InitProducerId. For a plain idempotent producer that can never happen — the broker issues a fresh PID with epoch 0 on every call — so restored sequences are ignored and the store is useful only for observability. It carries real weight for a transactional producer with a stable transactional.id, where the broker may hand back the same PID with a bumped epoch.

Timestamps

Both Producer and TransactionalProducer propagate the timestamp field from ProducerRecord to the Kafka record batch. If set, the timestamp is used as the base_timestamp of the record batch:

use krafka::producer::ProducerRecord;

let mut record = ProducerRecord::new("my-topic", b"value".to_vec());
record.timestamp = Some(1700000000000); // epoch millis
producer.send_record(record).await?;

Note: If timestamp is not set, the broker defaults apply (typically LogAppendTime or CreateTime depending on topic configuration).

Consume-Transform-Produce (Exactly-Once)

For read-process-write patterns with exactly-once guarantees:

use krafka::producer::TransactionalProducer;
use std::collections::HashMap;

// Commit consumer offsets atomically with produce
producer.begin_transaction()?;

// Process records and produce output
for record in consumer_records {
    let output = transform(&record)?;
    producer.send("output-topic", record.key, Some(&output)).await?;
}

// Commit offsets as part of the transaction.
//
// KIP-447: pass the consumer's live group metadata so the group coordinator
// can fence a zombie committer. Re-read it every transaction — the generation
// changes on every rebalance, and a cached value defeats the fencing.
let offsets = [TopicPartitionOffset::new(topic, partition, next_offset)];
let group_metadata = consumer.group_metadata().await?;
producer.send_offsets_to_transaction(&offsets, &group_metadata).await?;

// Atomic commit of messages and offsets
producer.commit_transaction().await?;

Transaction States

The producer maintains a state machine with atomic CAS (compare-and-swap) transitions for thread safety:

Note: State transitions are protected by atomic compare-and-swap operations, preventing race conditions when multiple tasks interact with the transactional producer concurrently.

Two-phase commit (KIP-939)

Requires the unstable-protocol feature (InitProducerId v6), broker transaction.version 3, and both WRITE and TWO_PHASE_COMMIT on the transactional-id resource.

Kafka transactions are atomic within Kafka. They are not atomic with anything else — so a service that must write to Kafka and a database, either both or neither, has no way to express that with commit_transaction() alone. KIP-939 supplies the missing half: an external coordinator (a database, an XA manager, a workflow engine) owns the commit decision, and Kafka's side is held in doubt until that decision arrives.

The obstacle is transaction.max.timeout.ms. Ordinarily the coordinator aborts a transaction that stays open too long — which is exactly right when Kafka owns the decision, and exactly wrong when it does not. two_phase_commit(true) sends enable2Pc on InitProducerId, and the broker then never times these transactions out.

use krafka::producer::{PreparedTxnState, TransactionOutcome, TransactionalProducer};

let producer = TransactionalProducer::builder()
    .bootstrap_servers("localhost:9092")
    .transactional_id("orders-sink")
    .two_phase_commit(true)   // contradicts transaction_timeout; setting both is an error
    .build()
    .await?;

producer.init_transactions().await?;
producer.begin_transaction()?;
producer.send("orders", None, Some(b"...")).await?;

// Prepare: flush everything, then stop accepting records. Sends no request —
// the prepare *is* the flush, and the coordinator was already told to hold.
let prepared: PreparedTxnState = producer.prepare_transaction().await?;

// Store it in the SAME external transaction as the rest of your work.
db.execute("INSERT INTO kafka_prepared (id, state) VALUES ($1, $2)",
           &[&"orders-sink", &prepared.to_string()])?;
db.commit()?;

producer.commit_transaction().await?;

Recovery is the point of all this. If the process dies between the flush and the database commit, the replacement asks the coordinator what it is still holding and compares:

let producer = TransactionalProducer::builder()
    .bootstrap_servers("localhost:9092")
    .transactional_id("orders-sink")
    .two_phase_commit(true)
    .build()
    .await?;

// Unlike init_transactions(), this does NOT abort what the previous
// incarnation left open.
if let Some(_ongoing) = producer.init_transactions_keeping_prepared().await? {
    let stored: PreparedTxnState = db
        .query_one("SELECT state FROM kafka_prepared WHERE id = $1", &[&"orders-sink"])?
        .get::<_, String>(0)
        .parse()?;

    match producer.complete_transaction(stored).await? {
        // The stored state names the transaction still open: the prepare was
        // durably recorded, so the external side committed and this must match.
        TransactionOutcome::Committed => println!("recovered and committed"),
        // It names an older one: the prepare never got recorded, the external
        // side rolled back, and this must abort. A *normal* outcome of a crash
        // in the window, not an error.
        TransactionOutcome::Aborted => println!("recovered and aborted"),
    }
}

PreparedTxnState renders as producer_id:epoch through Display and parses back through FromStr, so storing it needs no bespoke serialisation.

A prepared transaction with no stored state cannot be resolved by anything except a human. It sits in doubt indefinitely — that is what disabling the timeout buys — and blocks read_committed consumers on its partitions the whole time. Write the state durably before you report the prepare as successful, and treat a prepared transaction you cannot match as an incident.

Producer Interceptors

Interceptors allow you to observe and modify records before they are sent, and observe the acknowledgement (or error) after a send completes. Each record carries a RecordContext from one hook to the other, so an interceptor can hold a span or a timer across the send — see the Interceptors Guide for full details.

use krafka::interceptor::{InterceptorResult, ProducerInterceptor, RecordContext};
use krafka::producer::{Producer, ProducerRecord, RecordHeaders, RecordMetadata};
use krafka::error::KrafkaError;
use std::sync::Arc;

#[derive(Debug)]
struct AuditInterceptor;

impl ProducerInterceptor for AuditInterceptor {
    fn on_send(&self, record: &mut ProducerRecord, _ctx: &mut RecordContext) -> InterceptorResult {
        // Add a tracing header to every record
        record.headers.push(("x-trace-id".to_string(), Some(b"abc123".to_vec().into())));
        Ok(())
    }

    fn on_acknowledgement(
        &self,
        metadata: &RecordMetadata,
        error: Option<&KrafkaError>,
        _headers: &RecordHeaders,
        _ctx: &mut RecordContext,
    ) -> InterceptorResult {
        if let Some(err) = error {
            eprintln!("Send failed: {}", err);
        } else {
            println!("Sent to {}:{}", metadata.topic, metadata.partition);
        }
        Ok(())
    }
}

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .interceptor(Arc::new(AuditInterceptor))
    .build()
    .await?;

Next Steps

• Dead Letter Queue - Route permanently-failed records to an error topic
• Interceptors Guide - Producer and consumer interceptor hooks
• Consumer Guide - Learn about consuming messages
• Configuration Reference - All producer options
• Architecture Overview - How the producer works internally