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.

Backoff is exponential with jitter. Without it, every admin client watching one controller election retries in lockstep and arrives at the newly elected controller as a single wave. Raise retries if your elections routinely take longer than the default budget.

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, with two terminals: build_config() validates without a broker, build() validates and connects. Both run the same validator, so a configuration that passes one passes the other. tests/builder_surface.rs asserts this at compile time.

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?;