🦀 Krafka

A pure Rust, async-native Apache Kafka client designed for high performance, safety, and ease of use.

✨ Features

• 🦀 Pure Rust by default: No librdkafka or C dependencies; the optional zstd compression feature requires a C toolchain via zstd-sys
• ⚡ Async-native: Built on Tokio for true async I/O
• 🔒 Zero unsafe: Safe Rust by default
• 🚀 High performance: Zero-copy buffers, inline hot paths, efficient batching, concurrent batch flushing
• 📦 Full protocol support: Kafka protocol with all compression codecs
• 🔄 Incremental fetch sessions: KIP-227 fetch sessions for bandwidth-efficient multi-partition consumers
• 🔐 TLS/SSL encryption: Using rustls for secure connections
• 🔑 SASL authentication: PLAIN, SCRAM-SHA-256/512, OAUTHBEARER mechanisms
• 💯 Transactions: Exactly-once semantics with KIP-447 zombie fencing
• ☁️ Cloud-native: First-class AWS MSK support including IAM auth
• 🛡️ Security hardened: Secret zeroization, constant-time auth (subtle), decompression bomb protection, decode loop bounds (MAX_DECODE_ARRAY_LEN)
• 🔄 Built-in retry: Exponential backoff with metadata refresh on leader changes
• 📊 Metrics: Lock-free counters/gauges/latency with bounded per-topic cardinality
• 🧪 Fuzz + property tested: 6 cargo-fuzz targets and proptest round-trips across the protocol layer

Minimum Broker Version: Krafka requires Apache Kafka 3.9+. Protocol versions older than the Kafka 3.9 baseline have been removed.

🚀 Quick Start

Add Krafka to your Cargo.toml:

[dependencies]
krafka = "0.13.0"
tokio = { version = "1", features = ["full"] }

# For AWS MSK IAM authentication with full SDK support:
# krafka = { version = "0.13.0", features = ["aws-msk"] }

Producer

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

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

    // Send a message
    let metadata = producer
        .send("my-topic", Some(b"key"), b"Hello, Kafka!")
        .await?;
    
    println!("Sent to partition {} at offset {}", 
             metadata.partition, metadata.offset);

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

Consumer

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

#[tokio::main]
async fn main() -> Result<()> {
    let consumer = Consumer::builder()
        .bootstrap_servers("localhost:9092")
        .group_id("my-consumer-group")
        .auto_offset_reset(AutoOffsetReset::Earliest)
        .build()
        .await?;

    consumer.subscribe(&["my-topic"]).await?;

    loop {
        let records = consumer.poll(Duration::from_secs(1)).await?;
        for record in records {
            if let Some(ref value) = record.value {
                println!(
                    "Received: topic={}, partition={}, offset={}, value={:?}",
                    record.topic,
                    record.partition,
                    record.offset,
                    String::from_utf8_lossy(value)
                );
            }
        }
    }
}

Admin Client

use krafka::admin::{AdminClient, NewTopic};
use krafka::error::Result;
use std::time::Duration;

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

    // Create a topic
    let topic = NewTopic::new("new-topic", 6, 3)
        .with_config("retention.ms", "604800000");

    admin.create_topics(vec![topic], Duration::from_secs(30), false).await?;

    // List topics
    let topics = admin.list_topics().await?;
    println!("Topics: {:?}", topics);

    Ok(())
}

Transactional Producer

For exactly-once semantics across multiple partitions:

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

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

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

    // Atomic transaction
    producer.begin_transaction()?;
    producer.send("topic-a", Some(b"key"), b"value1").await?;
    producer.send("topic-b", Some(b"key"), b"value2").await?;
    producer.commit_transaction().await?;

    Ok(())
}

Authentication

Connect to secured Kafka clusters with SASL, SCRAM, OAUTHBEARER, or AWS MSK IAM — available on all client types:

use krafka::producer::Producer;
use krafka::consumer::Consumer;
use krafka::AdminClient;

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

// Consumer with SASL/PLAIN
let consumer = Consumer::builder()
    .bootstrap_servers("broker:9092")
    .group_id("secure-group")
    .sasl_plain("username", "password")
    .build()
    .await?;

// Producer with SASL/OAUTHBEARER
let producer = Producer::builder()
    .bootstrap_servers("broker:9093")
    .sasl_oauthbearer("your-jwt-token")
    .build()
    .await?;

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

📦 Modules

🗜️ Compression

Krafka supports all Kafka compression codecs, individually feature-gated:

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 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 select only what you need:

# Option 1: enable only the codecs you need
# `default-features = false` also drops the default `ring` TLS backend, so a
# crypto backend must be named explicitly.
krafka = { version = "0.13.0", default-features = false, features = ["lz4", "snappy", "ring"] }

# Option 2: enable all compression codecs, including zstd
# krafka = { version = "0.13.0", features = ["compression-all"] }

⚡ Performance Tuning

High Throughput Producer

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

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

Low Latency Consumer

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

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

🎯 Delivery Semantics

Pick the mode that matches what your data is worth.

Under acks=0, RecordMetadata::confirmation reports Unacknowledged — do not mistake the returned metadata for a durability guarantee.

Exactly-once

send_offsets_to_transaction takes a [ConsumerGroupMetadata], not a bare group ID. That metadata is what lets the group coordinator fence a zombie: an instance that was partitioned away, lost its partitions to a rebalance, and came back still holding a transaction. Without it the coordinator accepts the zombie's commit and it overwrites the position of the member that now owns the partition.

// Re-read for every transaction. The generation changes on every rebalance,
// so a cached value stops fencing at exactly the moment it matters.
let group_metadata = consumer
    .group_metadata()
    .await
    .ok_or("consumer has not joined the group yet")?;

producer.begin_transaction()?;
producer.send("out-topic", Some(b"key"), b"value").await?;
producer.send_offsets_to_transaction(&offsets, &group_metadata).await?;
producer.commit_transaction().await?;

See examples/exactly_once.rs for the full read-process-write loop.

🧩 Consumer Groups

Both rebalance protocols are supported: the classic JoinGroup/SyncGroup protocol and KIP-848 (group.protocol = consumer).

Four assignors ship: Range, RoundRobin, Sticky (eager), and CooperativeSticky. The default is the preference list [Range, CooperativeSticky], matching the Java client — every member advertises both, so a group moves from eager to cooperative rebalancing in a single rolling bounce rather than a full stop-the-world restart.

Rebalances do not wait on your poll loop. The background group task keeps heartbeating through a rebalance and sends JoinGroup/SyncGroup itself, so a consumer that is idle or busy between poll() calls does not hold the rest of the group up. The new assignment is applied — and your rebalance listener called — on the next poll(), so callbacks and record delivery stay on one thread and an offset commit cannot race a revocation.

max.poll.interval.ms is still enforced: an application that genuinely stops polling leaves the group so its partitions are reassigned promptly, rather than holding them while a background task vouches for it. Static members (group.instance.id) instead keep their assignment until the session expires, so a restart can reclaim it.

📐 Scope

Krafka speaks the client side of the Kafka protocol: 60+ API keys covering produce, fetch, group coordination, transactions, and administration. Broker-internal APIs (LeaderAndIsr, UpdateMetadata, Vote, FetchSnapshot, BrokerHeartbeat, …) are deliberately absent — a client does not speak them.

The schema-registry module implements the Confluent and AWS Glue wire formats (magic byte, schema ID framing, caching). Bring your own Avro/Protobuf/JSON-Schema codec; krafka does not impose one.

Tokio is the async runtime.

Authentication

SASL/PLAIN, SASL/SCRAM-SHA-256/512 (with RFC 5929 channel binding), SASL/OAUTHBEARER (with proactive token refresh), AWS MSK IAM, and mTLS.

GSSAPI/Kerberos is outside the scope of this client.

🧪 Testing Against a Fake Broker

Enable the test-broker feature to get an in-process Kafka broker your tests can drive directly. Real Producer/Consumer/AdminClient instances connect to it over a real TCP socket, so you exercise the actual client — no Docker, no containers, and failure modes you cannot reproduce against a healthy cluster.

use krafka::testing::{Control, FakeBroker};
use krafka::protocol::ApiKey;
use krafka::error::ErrorCode;

let broker = FakeBroker::start().await?;

// Make the next CreateTopics land on a non-controller and assert the client
// refreshes metadata and retries instead of surfacing the error.
broker.on(ApiKey::CreateTopics, |_| Control::Error(ErrorCode::NotController));

let admin = AdminClient::builder()
    .bootstrap_servers(broker.bootstrap_servers())
    .build()
    .await?;

Control covers Error, Delay, Disconnect, Silence and pass-through, and the cluster can be manipulated mid-test: set_leader, bump_leader_epoch, set_group_coordinator, set_txn_coordinator, set_controller, set_broker_online. That makes leader moves, coordinator failover, late responses and controller churn ordinary unit tests.

📚 Documentation

Full documentation is available at hupe1980.github.io/krafka

• Getting Started
• Producer Guide
• Consumer Guide
• Admin Client
• Configuration Reference
• Performance Tuning
• Architecture Overview
• Metrics & Observability
• Error Handling
• Interceptors
• Authentication

🎮 Examples

Run the examples with:

# Producer example
cargo run --example producer

# Consumer example
cargo run --example consumer

# Advanced consumer example (pause/resume, seek, manual commits)
cargo run --example consumer_advanced

# Admin client example
cargo run --example admin

# Transactional producer example
cargo run --example transactional_producer

# Exactly-once read-process-write (KIP-447 zombie fencing)
cargo run --example exactly_once

# Authentication examples (SASL, SCRAM, MSK IAM)
cargo run --example authentication

🤝 Contributing

Contributions are welcome!

📄 License

Licensed under either the MIT License or the Apache License 2.0, at your option.