kafrust 0.2.1

A pure Rust Kafka client with no librdkafka or C toolchain dependency.
Documentation

kafrust

Crates.io Docs.rs CI

A pure Rust Kafka client with no librdkafka or C toolchain dependency.

kafrust is the high-level client crate in the kafrust workspace. It provides Tokio-based producer, direct consumer, and alpha classic consumer group APIs on top of the companion kafrust-protocol wire-format crate.

Current release: 0.2.1.

This crate is alpha. Use it for experiments, local broker checks, simple internal tools, and API evaluation. For broad production Kafka workloads that need mature features immediately, rust-rdkafka remains the practical Rust default today.

Design Goals

  • Keep Kafka concepts visible in public APIs.
  • Stay pure Rust with no librdkafka, C client binding, or required C toolchain.
  • Make protocol and runtime behavior auditable through small, tested slices.
  • Claim compatibility only when a real broker profile has been verified.

The public model intentionally exposes Kafka terms such as bootstrap servers, client IDs, topics, partitions, offsets, acknowledgements, metadata refresh, consumer groups, generations, members, heartbeats, and commits.

Install

[dependencies]

kafrust = "0.2"

tokio = { version = "1", features = ["macros", "rt"] }

For a multi-threaded application runtime, enable Tokio's rt-multi-thread feature in the application.

Producer

use kafrust::{Acks, ProducerConfig, ProducerRecord};

#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
    let mut producer = ProducerConfig::new(["localhost:9092"])
        .client_id("example-producer")
        .acks(Acks::Leader)
        .build()
        .await?;

    let metadata = producer
        .send(
            ProducerRecord::to("orders")
                .key("order-123")
                .value("created")
                .header("source", "checkout"),
        )
        .await?;

    println!(
        "produced {}-{}@{}",
        metadata.topic(),
        metadata.partition(),
        metadata.offset()
    );

    Ok(())
}

Batch Producer

Producer::send_batch returns metadata in input order. Use Producer::send_batch_report when partial per-record failures need to be inspected without losing successful records.

use kafrust::{Acks, ProducerConfig, ProducerRecord};

#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
    let mut producer = ProducerConfig::new(["localhost:9092"])
        .client_id("example-batch-producer")
        .acks(Acks::Leader)
        .max_records_per_batch(500)
        .max_batch_bytes(64 * 1024)
        .build()
        .await?;

    let report = producer
        .send_batch_report([
            ProducerRecord::to("orders").key("order-124").value("created"),
            ProducerRecord::to("orders").key("order-125").value("created"),
        ])
        .await?;

    for outcome in report.records() {
        if let Some(metadata) = outcome.metadata() {
            println!("{}-{}@{}", metadata.topic(), metadata.partition(), metadata.offset());
        }
        if let Some(failure) = outcome.failure() {
            eprintln!(
                "record {} failed on {}-{}: {}",
                failure.record_index(),
                failure.topic(),
                failure.partition(),
                failure.error()
            );
        }
    }

    Ok(())
}

Buffered Producer

ProducerConfig::build_buffered creates an opt-in buffered producer. Records are flushed by linger time, record count, byte count, explicit flush, or close.

use kafrust::{Acks, ProducerConfig, ProducerRecord};

#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
    let mut producer = ProducerConfig::new(["localhost:9092"])
        .client_id("example-buffered-producer")
        .acks(Acks::Leader)
        .linger_ms(10)
        .max_records_per_batch(100)
        .build_buffered()
        .await?;

    let delivery = producer
        .send(ProducerRecord::to("orders").value("buffered value"))
        .await?;

    let metadata = delivery.await?;
    producer.close().await?;

    println!("buffered record offset {}", metadata.offset());

    Ok(())
}

Direct Consumer

The direct consumer path fetches from explicit topic partitions and offsets. This is useful when consumer group behavior is not needed.

use kafrust::ConsumerConfig;

#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
    let mut consumer = ConsumerConfig::new(["localhost:9092"])
        .client_id("example-consumer")
        .max_poll_records(500)
        .build()
        .await?;

    consumer.assign("orders", 0, 0);

    for record in consumer.poll().await? {
        println!(
            "fetched {}-{}@{} value={:?}",
            record.topic(),
            record.partition(),
            record.offset(),
            record.value().map(String::from_utf8_lossy)
        );
    }

    Ok(())
}

Consumer Group

The consumer group API is an alpha classic consumer group path with join, sync, heartbeat, poll, and offset commit support.

use kafrust::ConsumerGroupConfig;

#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
    let mut group = ConsumerGroupConfig::new(["localhost:9092"], "orders-reader")
        .client_id("example-consumer-group")
        .subscribe("orders")
        .join()
        .await?;

    let records = group.poll().await?;
    group.commit_offsets().await?;

    println!("processed {} records", records.len());

    Ok(())
}

Security Protocols

SecurityProtocol models Kafka connection modes:

  • Plaintext
  • Tls
  • SaslPlaintext
  • SaslTls

Plaintext is the default and the only implemented transport in 0.2.1. TLS and SASL variants are configuration targets and currently return Error::Unsupported before connecting.

Compatibility

The 0.2.x alpha line is verified against a single-node Apache Kafka 3.7.2 KRaft broker over PLAINTEXT.

Verified high-level paths include:

  • ApiVersions v0 and Metadata v1 roundtrips.
  • Producer single-record, batch, and buffered sends.
  • Direct topic-partition fetch using Fetch v2 response decoding.
  • Classic consumer group join, sync, heartbeat, poll, and offset commit.

Current Limits

  • APIs are pre-1.0 and can change between minor versions.
  • TLS and SASL are not implemented yet.
  • Broker compatibility is verified against Kafka 3.7.2 only.
  • Multi-broker clusters, leader failover, rack awareness, and partition expansion are not yet claimed.
  • Idempotent producers, transactions, compression, admin APIs, and broad observability are not implemented yet.
  • acks=0 remains unsupported because the current request loop expects a broker response.

Examples

Run examples from the repository with a local Kafka broker:

KAFRUST_BOOTSTRAP_SERVERS=localhost:9092 \

KAFRUST_TOPIC=kafrust-smoke \

cargo run -p kafrust --example producer_send

Available examples include:

  • broker_roundtrip
  • producer_send
  • producer_send_batch
  • producer_buffered
  • consumer_fetch
  • find_group_coordinator
  • consumer_group_poll

Project Docs