Configuration

Every connection, producer, consumer, admin and transport option, with their defaults.

krafka has one place for each setting. Connection settings — how the process reaches the cluster — live on the Kafka builder and apply to every client built from that handle. Role settings live on the role builders: kafka.producer(), kafka.consumer(group), kafka.consumer_without_group(), kafka.share_consumer(group), and on the AdminClient that kafka.admin() returns. Where a setting has a Java client equivalent, its description names it.

use krafka::Kafka;
use std::time::Duration;

let kafka = Kafka::builder("kafka1:9092,kafka2:9092") // connection settings
    .client_id("orders-service")
    .request_timeout(Duration::from_secs(30))
    .connect()
    .await?;

let producer = kafka.producer().linger(Duration::from_millis(5)).build().await?; // role settings
let consumer = kafka.consumer("orders").build().await?;

No setter returns a Result: connect() and build() validate, and an invalid value is a KrafkaError::Config naming the setting. A second identity, other credentials or a separate connection pool is a second Kafka handle.

Cargo Features

Generated from Cargo.toml and xtask/kips.toml; Default follows default = [...], and the gated API versions come from the api_versions! table.

Connection Configuration (Kafka::builder)

The transport settings below are connection settings too.

Transport

use krafka::Kafka;
use std::time::Duration;

let kafka = Kafka::builder("localhost:9092")
    .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)))
    .connect()
    .await?;

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

Every client of the handle shares the pool, and with it the proxy, the descriptor cap and the reload interval.

Producer Configuration (kafka.producer())

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

Every codec decodes in every build (zstd through the pure-Rust ruzstd). Gzip, Snappy and LZ4 also encode in every build; zstd encoding is opt-in through the zstd feature, which compiles the C zstd library through zstd-sys. Use Compression::is_available() to check whether a codec can encode; build() rejects a codec that cannot.

use krafka::Compression;

Compression::None    // No compression
Compression::Gzip    // Gzip compression
Compression::Snappy  // Snappy compression
Compression::Lz4     // LZ4 compression
Compression::Zstd    // Zstandard compression (encoding needs feature = "zstd")

Producer Builder Example

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

let kafka = Kafka::builder("kafka1:9092,kafka2:9092")
    .client_id("my-producer")
    .connect()
    .await?;
let producer = kafka
    .producer()
    .acks(Acks::All)
    .compression(Compression::Lz4)
    .batch_size(64 * 1024)
    .linger(Duration::from_millis(5))
    .delivery_timeout(Duration::from_secs(120))
    .retry_backoff(Duration::from_millis(200))
    .build()
    .await?;

Consumer Configuration (kafka.consumer(group))

kafka.consumer(group) joins consumer group group; kafka.consumer_without_group() builds a consumer that assigns its own partitions and commits nothing.

AutoOffsetReset Values

use krafka::consumer::AutoOffsetReset;

AutoOffsetReset::Earliest  // Start from the earliest offset
AutoOffsetReset::Latest    // Start from the latest offset
AutoOffsetReset::ByDuration(Duration::from_secs(86_400))
                           // First record of the last 24 h, else the log end (KIP-1106)
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::Kafka;
use krafka::consumer::{AutoOffsetReset, IsolationLevel};
use std::time::Duration;

let kafka = Kafka::builder("kafka1:9092,kafka2:9092")
    .client_id("my-consumer")
    .connect()
    .await?;
let consumer = kafka
    .consumer("my-consumer-group")
    .auto_offset_reset(AutoOffsetReset::Earliest)
    .enable_auto_commit(false)
    .fetch_min_bytes(1024)
    .fetch_max_bytes(10 * 1024 * 1024)
    .max_partition_fetch_bytes(1024 * 1024)
    .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")
    .build()
    .await?;

Share Consumer Configuration (kafka.share_consumer(group))

kafka.share_consumer(group) builds a KIP-932 share-group member; see Share Consumer.

Admin Client Configuration (kafka.admin())

kafka.admin() returns an AdminClient directly; it contacts no broker until its first call. Its settings are set on the client:

Retries and the call deadline

On NOT_CONTROLLER the client fetches metadata, finds the new controller and sends again. Retries continue until the call's deadline: its options' timeout, else default_api_timeout. See Admin Client for what is retried.

use std::time::Duration;

// A cluster whose elections take a while.
let admin = kafka
    .admin()
    .default_api_timeout(Duration::from_secs(120))
    .retry_backoff(Duration::from_millis(250));

Validation

Every setting is checked where it takes effect: connection settings by Kafka::builder(..).connect(), role settings by the role builder's build() (or build_transactional). The error is a KrafkaError::Config that names the setting:

// Without the `zstd` feature this fails at build time, not on the first send.
let result = kafka
    .producer()
    .compression(krafka::Compression::Zstd)
    .build()
    .await;

Two checks keep the surface complete:

• tests/builder_surface.rs — do these methods exist, with these shapes? Each line fails to compile if the method it names disappears.
• just config-reachability — is any field unreachable? It walks every config struct's fields and requires each to have a same-named setter on its builder, or an entry in an exception list carrying a reason.

SOCKS5 Proxy

proxy routes every broker connection through a SOCKS5 proxy, for brokers reachable only through a bastion. The proxy resolves broker hostnames; the client sends them unresolved.

Proxy Without Authentication

use krafka::{Kafka, ProxyConfig};

let kafka = Kafka::builder("kafka.internal:9092")
    .proxy(ProxyConfig::new("socks5-proxy.corp:1080"))
    .connect()
    .await?;

Proxy With Authentication

use krafka::{Kafka, ProxyConfig};

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

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

Proxy With TLS/SASL

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

use krafka::auth::{AuthConfig, TlsConfig};
use krafka::{Kafka, ProxyConfig};

let kafka = Kafka::builder("kafka.secure.internal:9093")
    .security(AuthConfig::ssl(TlsConfig::new()))
    .proxy(ProxyConfig::new("bastion:1080"))
    .connect()
    .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

AuthConfig::from_env() builds the security settings from the standard KAFKA_SECURITY_PROTOCOL, KAFKA_SASL_* and KAFKA_SSL_* variables; see Authentication. Everything else is read by your application and passed to the builders.

Performance Tuning Profiles

High Throughput Producer

use krafka::producer::Acks;
use krafka::Compression;
use std::time::Duration;

let producer = kafka
    .producer()
    .acks(Acks::All)
    .compression(Compression::Lz4)
    .batch_size(1024 * 1024)
    .linger(Duration::from_millis(50))
    .build()
    .await?;

Low Latency Producer

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

let producer = kafka
    .producer()
    .acks(Acks::Leader)
    .idempotent(false) // idempotence needs acks = All
    .linger(Duration::ZERO)
    .build()
    .await?;

High Throughput Consumer

let consumer = kafka
    .consumer("high-throughput")
    .fetch_max_bytes(100 * 1024 * 1024)
    .max_partition_fetch_bytes(10 * 1024 * 1024)
    .max_poll_records(10_000)
    .build()
    .await?;

Low Latency Consumer

use std::time::Duration;

let consumer = kafka
    .consumer("low-latency")
    .fetch_min_bytes(1)
    .fetch_max_wait(Duration::from_millis(10))
    .max_poll_records(10)
    .build()
    .await?;