kacrab

A Rust-native Apache Kafka client — producer, consumer, and admin — built from the Kafka protocol up. Not a librdkafka wrapper.

Repository · Design & Internals book · API reference

The protocol, wire, and client logic are pure Rust with unsafe_code forbidden workspace-wide. The dependency tree is not fully C-free, though: the default TLS provider (rustls + aws-lc-rs) is C/assembly, and the optional zstd, lz4-hc, and gssapi features add C. For a C-free build, use a pure-Rust rustls provider and the gzip/snappy/lz4 codecs.

Install

[dependencies]
kacrab = { version = "0.1", features = ["producer"] }
tokio = { version = "1", features = ["macros", "rt"] }

The crate compiles almost nothing by default (default = []); opt into the surfaces you use:

producer — the producer API.
consumer — the consumer API; pulls in the compression codecs (fetched batches must decompress) and regex for pattern subscription.
admin — the AdminClient API.
gzip, snappy, lz4 — pure-Rust record-batch compression codecs (no C toolchain).
zstd — compression via the C libzstd (zstd-sys); needs a C compiler at build time. The compression meta-feature enables all four (gzip + snappy + lz4 + zstd), so it needs one too — for a pure-Rust build, enable only the first three.
lz4-hc — C-FFI LZ4 backend adding high-compression levels (compression.lz4.level 3..=12); plain lz4 is fast-mode only.
gssapi — Kerberos/GSSAPI through platform Kerberos libraries.
macros — re-exports the config macro helper.

Surface

config — Java-style ClientConfig, typed producer/consumer/admin configs, official Kafka config metadata, and strict validation.
common — shared org.apache.kafka.common domain types (TopicPartition, OffsetAndMetadata, ConsumerGroupMetadata, Node), always compiled and re-exported by producer/consumer/admin.
wire — Tokio broker sessions, ApiVersions negotiation, TLS, SASL, metadata, bounded in-flight requests, and request/response dispatch.
producer — Java-style producer builder, batching, linger, bounded memory, compression, murmur2 + sticky/adaptive partitioning, multi-broker dispatch with leadership-change failover, transactions, and a Kafka-faithful idempotent path (per-partition multi-in-flight, ordered retry, deferred epoch bump, sequence wraparound). Interceptors and Kafka-named metrics included.
consumer — full Apache Kafka 4.3.0 feature parity: manual assignment, topic and pattern (regex) subscription, classic groups (range/roundrobin/sticky eager + incremental cooperative-sticky, KIP-429) and the KIP-848 server-side protocol; topic-id-keyed fetch (KIP-516, up to v18), incremental fetch sessions (KIP-227), truncation detection (KIP-320), commit_sync/commit_async/auto-commit, background heartbeat, static membership, typed deserializers, interceptors, and metrics().
admin — the full Apache Kafka 4.3.0 Admin operation surface (62 operations): topics, configs (incremental), ACLs, groups & offsets, transactions, delegation tokens, quotas, SCRAM, reassignments, KRaft quorum, and the 4.x share/streams group families.

Every client surface — producer, consumer, admin, every SASL mechanism and TLS mode, every compression codec, 3-broker failover — is verified end-to-end against real Apache Kafka 4.3.0 brokers. On the same broker and defaults, producer throughput measures +25-28% over the Java client at ~4x less memory, and consumer throughput 1.9-4x at ~16-20x less memory; methodology and caveats in the benchmarks chapter.

Producer

send is synchronous like Kafka's Producer.send: it returns a SendFuture right away, and you await that future for the broker acknowledgement. Batching happens automatically through batch.size, linger.ms, buffer memory, and flush/close boundaries.

use kacrab::producer::{Producer, ProducerRecord};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut producer = Producer::builder()
        .set("bootstrap.servers", "127.0.0.1:9092")
        .set("acks", "all")
        .set("enable.idempotence", "true")
        .set("linger.ms", "5")
        .build()
        .await?;

    let delivery = producer.send(
        ProducerRecord::new("orders", 0).key("order-42").value("created"),
    )?;

    producer.flush().await?;
    let receipt = delivery.await?;
    println!("{}-{}@{}", receipt.topic, receipt.partition, receipt.offset);

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

Transactions use the same producer (transactional.id + init_transactions/begin_transaction/commit_transaction). Serializers are a compile-time Rust trait (ProducerSerializer<T> via build_with_serializers), not key.serializer class names.

Consumer

use std::time::Duration;
use kacrab::consumer::{Consumer, StringDeserializer};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut consumer = Consumer::from_map([
        ("bootstrap.servers", "localhost:9092"),
        ("group.id", "orders-workers"),
        ("auto.offset.reset", "earliest"),
        // Incremental rebalancing; use ("group.protocol", "consumer") for KIP-848.
        ("partition.assignment.strategy", "cooperative-sticky"),
    ])
    .await?;
    consumer.subscribe(["orders"])?;

    let (keys, values) = (StringDeserializer, StringDeserializer);
    loop {
        let records = consumer.poll(Duration::from_secs(1)).await?;
        for record in &records {
            let (key, value) = record.deserialized(&keys, &values)?;
            println!(
                "{}-{}@{}: {key:?} = {value:?}",
                record.topic, record.partition, record.offset
            );
        }
        consumer.commit_sync().await?;
    }
}

Records are bytes-first (ConsumerRecord.key/value: Option<Bytes>), with a typed ConsumerDeserializer layer on top. Offsets commit sync, async, or automatically, with leader-epoch awareness.

Admin

Admin mirrors Java's Admin with snake_case methods and per-call options structs:

use kacrab::admin::{AdminClient, CreateTopicsOptions, NewTopic};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let admin = AdminClient::from_map([("bootstrap.servers", "localhost:9092")]).await?;

    admin
        .create_topics(vec![NewTopic::new("orders", 6, 3)], CreateTopicsOptions::default())
        .await?;

    for topic in admin.list_topics(Default::default()).await? {
        println!("{}", topic.name);
    }
    Ok(())
}

Java Compatibility

Auth, producer, consumer, and admin are outcome-faithful to the Java client for the implemented surface — not a literal class-for-class port:

Familiar Java client keys work as-is: bootstrap.servers, security.protocol, ssl.truststore.location, sasl.mechanism, sasl.jaas.config, acks, enable.idempotence, transactional.id, batch.size, linger.ms, max.in.flight.requests.per.connection, ...
PLAINTEXT, SSL, SASL_PLAINTEXT, and SASL_SSL; TLS trust/identity material in PEM, JKS, and PKCS12; SASL PLAIN, SCRAM-SHA-256/512, OAUTHBEARER (JAAS options, files, HTTP(S) token endpoints, or locally signed JWT assertions), and feature-gated GSSAPI. Handshake and auth failures fail fast with the broker's reason, matching Java.
JVM login module and callback handler classes are the intentional boundary: Rust cannot load Java classes. sasl.jaas.config strings are parsed for their credential options only; custom SASL flows plug in through the native sasl_client_authenticator(...) hook.
Protocol request/response structs are generated from the Apache Kafka message schemas and checked byte-for-byte against the Kafka Java client as an external oracle.

Status

kacrab is pre-release: the public API and runtime behavior are not stable release guarantees yet. Protocol, wire, auth, producer, consumer, and admin all have a verified usable baseline; the remaining work before production-ready is measurement under load (sustained multi-broker stress, cross-DC/high-RTT coverage, memory soak, latency-percentile gates), not correctness. See the roadmap.

Kafka Streams is out of scope. kacrab is a Kafka client library — the equivalent of KafkaProducer/KafkaConsumer/Admin, not a stream-processing framework. It deliberately provides the primitives a streams runtime would build on (transactions, consumer groups, offsets) and stops there.

Author

kacrab is authored and maintained by pirumu.

License

This crate is licensed under either MIT or Apache-2.0, matching the workspace.

kacrab 0.1.0