Configuration

Every producer, consumer, admin and transport option, with defaults and the reason each default was chosen.

Producer Configuration

Acks Values

use krafka::producer::Acks;

Acks::None    // 0: Don't wait for acknowledgment
Acks::Leader  // 1: Wait for leader acknowledgment
Acks::All     // -1: Wait for all in-sync replicas

Compression Values

Each codec requires its corresponding Cargo feature (gzip, snappy, lz4, zstd). The default compression feature enables gzip, snappy, and LZ4. Zstd is explicitly opt-in through zstd or compression-all because it requires a C toolchain via zstd-sys. Use [Compression::is_available()] to check at runtime.

use krafka::protocol::Compression;

Compression::None    // No compression (always available)
Compression::Gzip    // Gzip compression (feature = "gzip")
Compression::Snappy  // Snappy compression (feature = "snappy")
Compression::Lz4     // LZ4 compression (feature = "lz4")
Compression::Zstd    // Zstandard compression (feature = "zstd")

Producer Builder Example

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

let producer = Producer::builder()
    .bootstrap_servers("kafka1:9092,kafka2:9092")
    .client_id("my-producer")
    .compression(Compression::Lz4)
    .batch_size(65536)
    .linger(Duration::from_millis(5))
    .request_timeout(Duration::from_secs(30))
    .delivery_timeout(Duration::from_secs(120))
    .retries(5)
    .retry_backoff(Duration::from_millis(200))
    .metadata_topic_cache_ttl(Duration::from_secs(300))
    .build()
    .await?;

Consumer Configuration

AutoOffsetReset Values

use krafka::consumer::AutoOffsetReset;

AutoOffsetReset::Earliest  // Start from the earliest offset
AutoOffsetReset::Latest    // Start from the latest offset
AutoOffsetReset::None      // Error if no committed offset (strictly enforced)

IsolationLevel Values

use krafka::consumer::IsolationLevel;

IsolationLevel::ReadUncommitted  // Read all messages
IsolationLevel::ReadCommitted    // Only read committed transaction messages

Consumer Builder Example

use krafka::consumer::{Consumer, AutoOffsetReset, IsolationLevel};
use std::time::Duration;

let consumer = Consumer::builder()
    .bootstrap_servers("kafka1:9092,kafka2:9092")
    .group_id("my-consumer-group")
    .client_id("my-consumer")
    .auto_offset_reset(AutoOffsetReset::Earliest)
    .enable_auto_commit(false)
    .fetch_min_bytes(1024)
    .fetch_max_bytes(10485760)
    .max_partition_fetch_bytes(1048576)
    .fetch_max_wait(Duration::from_millis(100))
    .max_poll_records(1000)
    .session_timeout(Duration::from_secs(30))
    .heartbeat_interval(Duration::from_secs(10))
    .isolation_level(IsolationLevel::ReadCommitted)
    .group_instance_id("instance-1")  // Static group membership
    .build()
    .await?;

Admin Client Configuration

Controller and coordinator retries

A NOT_CONTROLLER answer means the controller moved, and the only correct response is to re-resolve it and try again — create_topics during a rolling controller restart hits this routinely. krafka refreshes metadata, reconnects, and retries up to retries times.

Both settings used to be compile-time constants: five attempts spaced by a flat 100 ms, with the documentation claiming they were retry.backoff.ms. Two things were wrong with that. The budget is about a second of real time, which is short for a KRaft election, and the flat sleep had no jitter — so every admin client watching one election retried in lockstep and arrived at the newly elected controller as a single wave.

use krafka::admin::AdminClient;
use std::time::Duration;

let admin = AdminClient::builder()
    .bootstrap_servers("localhost:9092")
    // A cluster whose elections take a few seconds.
    .retries(15)
    .retry_backoff(Duration::from_millis(250))
    .build()
    .await?;

Admin Client Builder Example

use krafka::admin::AdminClient;
use std::time::Duration;

let admin = AdminClient::builder()
    .bootstrap_servers("kafka1:9092,kafka2:9092")
    .client_id("my-admin-client")
    .request_timeout(Duration::from_secs(60))
    .build()
    .await?;

Validating a configuration without a broker

Every client builder has two terminals over one validator:

build() runs exactly the checks build_config() runs, so a config that passes the synchronous terminal will not be rejected later for a configuration reason. That makes settings testable at startup, in a unit test, or in a config-linting tool — none of which want a live cluster.

use krafka::producer::Producer;

// Fails immediately: the `zstd` codec is not compiled in.
let err = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .compression(krafka::protocol::Compression::Zstd)
    .build_config()
    .unwrap_err();

Validation also normalises — a compression level is clamped into the selected codec's range, for example — so the returned config may differ from what was set.

There is exactly one builder per client. krafka used to ship a second, internal *ConfigBuilder alongside each public builder. They duplicated 72 methods between them and their validation had diverged — the public path, the only one anybody used, silently skipped six checks including the compression-codec availability test. The duplicates are gone, and tests/builder_surface.rs asserts at compile time that every builder keeps both terminals.

Transport Configuration

Socket- and pool-level settings live on a single TransportConfig, accepted by every builder via .transport(..) — Producer, Consumer, AdminClient, TransactionalProducer, ShareConsumer and KrafkaClient.

The defaults reproduce krafka's historical behaviour exactly, so adopting the type is never itself a behaviour change.

use krafka::network::TransportConfig;
use krafka::consumer::Consumer;
use std::time::Duration;

let transport = TransportConfig::builder()
    .tcp_keepalive(Some(Duration::from_secs(30)))
    .max_response_size(200 * 1024 * 1024)
    .socket_receive_buffer(Some(4 * 1024 * 1024))   // long-haul fetch
    .max_connections(Some(64))
    .tls_reload_interval(Some(Duration::from_secs(3600)))
    .build()?;

let consumer = Consumer::builder()
    .bootstrap_servers("localhost:9092")
    .group_id("my-group")
    .transport(transport)
    .build()
    .await?;

build() rejects values that cannot be honoured — a zero max_in_flight_requests (nothing could ever be sent), a max_response_size below 1 KiB, or a zero interval where None is the way to switch a period off.

One configuration surface across every client

Producer, Consumer, AdminClient, TransactionalProducer, ShareConsumer and KrafkaClient accept the same core settings — client_id, request_timeout, connect_timeout, metadata_max_age, metadata_recovery_strategy, transport, auth and the SASL convenience helpers.

They did not always. ShareConsumerBuilder exposed request_timeout but not connect_timeout, and because build() rejects request_timeout < connect_timeout, any share consumer wanting a sub-10-second timeout failed at construction with an error naming a value the builder could not change. TransactionalProducer had no KIP-899 recovery configuration at all, and AdminClient hard-coded its metadata age. tests/builder_surface.rs now asserts the matrix at compile time.

Two checks keep it that way, and they answer different questions:

• tests/builder_surface.rs — do these specific methods exist on every client? Each line fails to compile if the method it names disappears. This is the right tool for a cross-client judgement ("both consumers should accept a deserializer"), which no script can infer.
• just config-reachability — is any field unreachable? It walks every config struct's fields and requires each to have a same-named builder setter and public accessor, or an entry in an exception list carrying a reason.

The second exists because the first is blind by construction to the defect it was written for: a setting nobody remembered to expose is also a setting nobody remembered to add a line for. ShareConsumerConfig carried five such fields — including all four of KIP-932's fetch knobs, read when the ShareFetch request was built and settable by no one — while passing builder_surface.rs.

Per-client timeouts

These stay on the client builders, because they are request semantics rather than transport tuning:

SOCKS5 Proxy

Route all broker connections through a SOCKS5 proxy. This is useful for VPN/bastion setups where brokers are not directly reachable. The proxy handles DNS resolution, so broker hostnames are sent as-is (not pre-resolved).

Enable the socks5 feature:

cargo add krafka --features socks5

Proxy Without Authentication

use krafka::network::ProxyConfig;

let consumer = Consumer::builder()
    .bootstrap_servers("kafka.internal:9092")
    .group_id("my-group")
    .proxy(ProxyConfig::new("socks5-proxy.corp:1080"))
    .build()
    .await?;

Proxy With Authentication

use krafka::network::ProxyConfig;

let producer = Producer::builder()
    .bootstrap_servers("kafka.internal:9092")
    .proxy(ProxyConfig::with_credentials(
        "socks5-proxy.corp:1080",
        "proxy-user",
        "proxy-password",
    ))
    .build()
    .await?;

Proxy credentials are zeroized from memory on drop and redacted in Debug output.

Proxy With TLS/SASL

Proxy and authentication can be combined — the SOCKS5 tunnel is established first, then TLS and/or SASL negotiation proceeds over the tunneled connection:

use krafka::auth::AuthConfig;
use krafka::network::ProxyConfig;

let consumer = Consumer::builder()
    .bootstrap_servers("kafka.secure.internal:9093")
    .group_id("secure-group")
    .auth(AuthConfig::tls())
    .proxy(ProxyConfig::new("bastion:1080"))
    .build()
    .await?;

Topic Configuration

For NewTopic when creating topics:

use krafka::admin::NewTopic;

let topic = NewTopic::new("my-topic", 12, 3)
    .with_config("cleanup.policy", "compact")
    .with_config("retention.ms", "604800000")      // 7 days
    .with_config("segment.bytes", "1073741824")    // 1GB
    .with_config("min.insync.replicas", "2");

Common Topic Configs

Environment Variables

krafka can be configured via environment variables:

export KAFKA_BOOTSTRAP_SERVERS=kafka1:9092,kafka2:9092
export KAFKA_CLIENT_ID=my-app
export KAFKA_GROUP_ID=my-group

Note: Environment variable support requires explicit configuration in your application.

Performance Tuning Profiles

High Throughput Producer

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::Leader)
    .compression(Compression::Lz4)
    .batch_size(1048576)                  // 1MB batches
    .linger(Duration::from_millis(50))    // Allow batching
    .build()
    .await?;

Low Latency Producer

let producer = Producer::builder()
    .bootstrap_servers("localhost:9092")
    .acks(Acks::None)
    .batch_size(1)
    .linger(Duration::ZERO)    // The default: never wait, but still coalesce under load
    .build()
    .await?;

High Throughput Consumer

let consumer = Consumer::builder()
    .bootstrap_servers("localhost:9092")
    .group_id("high-throughput")
    .fetch_max_bytes(104857600)           // 100MB
    .max_partition_fetch_bytes(10485760)  // 10MB
    .max_poll_records(10000)
    .build()
    .await?;

Low Latency Consumer

let consumer = Consumer::builder()
    .bootstrap_servers("localhost:9092")
    .group_id("low-latency")
    .fetch_min_bytes(1)
    .fetch_max_wait(Duration::from_millis(10))
    .max_poll_records(10)
    .build()
    .await?;