Rust Client & Streams
Asynchronous Kafka client and stream processing libraries built natively on Tokio and krabka-protocol.
krabka-client-rs and krabka-streams-rs are under active development. Core producer, consumer, and admin APIs are functional and tested in CI; streams DSL topology drivers are in developer preview.
The Rust client ecosystem is decoupled from the broker binary, depending on krabka-protocol for wire serialization. It provides memory-safe implementations of Kafka producers, consumers, admin tools, and streaming topologies without relying on librdkafka C-bindings. The client crates live in krabka-client-rs; the streams crate lives in krabka-streams-rs.
Crate Architecture
| Crate | Subsystem & Features |
|---|---|
| krabka-client-core | Broker connection pool, API version negotiation, typed request dispatch, retry backoff, SASL/TLS authentication. |
| krabka-client-producer | Idempotent producer (InitProducerId, sequence tracking), batch accumulation, and transactional APIs (KIP-447). |
| krabka-client-consumer | Consumer group protocol (rebalance, heartbeats, cooperative sticky assignor) and KIP-932 Share Groups. |
| krabka-client-admin | Administrative client for topic CRUD, partition reassignments, ACLs, configs, quotas, SCRAM credentials, and group management. |
| krabka-client-coordination | Leader election, leases, and fencing tokens as a Kafka client API. |
| krabka-client-streams | KIP-1071 streams-group membership, typed Processor API, KStream/KTable DSL with windowing, Turso-backed state stores, and interactive queries. Lives in krabka-streams-rs. |
| krabka-schema-serde | Schema Registry serialization (Avro, JSON Schema, Protobuf). Lives in krabka-schema-registry. |
Adding the Crates
These repositories do not publish to crates.io, and the krabka-* names there still belong to an older repository. Depend on them by git revision, and copy the [patch.crates-io] block from the Cargo.toml at the pinned revision of the repository you depend on, so that krabka-protocol and krabka-units resolve to one copy. The examples below need Rust 1.98.1 or newer, tokio with the macros and rt-multi-thread features, bytes for the producer, and krabka-units for the consumers.
Async Producer Example
The idempotent producer initializes producer epochs and base sequences, which the broker uses to detect duplicate batches.
send(record).await? waits for delivery and returns metadata.
For pipelining, enqueue(record).await? queues a record and returns a delivery handle;
await each handle with delivery.await? to check its result.
// Cargo.toml:
// krabka-client-producer = { git = "https://github.com/krabka-io/krabka-client-rs", rev = "c4ce581587772b3750d88c61f3389d9626a20235" }
// plus the [patch.crates-io] block from that revision's Cargo.toml
// bytes = "1"
// tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
use std::time::Duration;
use bytes::Bytes;
use krabka_client_producer::{Acks, Compression, Producer, ProducerRecord};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Build an idempotent producer with LZ4 compression and acks=all
let producer = Producer::builder()
.bootstrap("localhost:9092")
.compression(Compression::Lz4)
.acks(Acks::All)
.linger(Duration::from_millis(5))
.build()
.await?;
// Send record asynchronously with broker sequence deduplication
producer
.send(ProducerRecord {
topic: "events".into(),
value: Some(Bytes::from("hello krabka")),
..Default::default()
})
.await?;
producer.flush().await?;
producer.close().await?;
Ok(())
}Consumer Groups & Share Groups
The consumer client supports consumer group rebalancing and KIP-932 share-group queues for cooperative record delivery:
use krabka_client_consumer::{AutoOffsetReset, Consumer};
use krabka_units::millis;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Build a subscribe-style group consumer
let mut consumer = Consumer::builder()
.bootstrap("localhost:9092")
.group_id("order-processors")
.client_id("processor-node-1")
.auto_offset_reset(AutoOffsetReset::Earliest)
.subscribe(["orders".to_string()])
.build()
.await?;
loop {
let records = consumer.poll(millis(500)).await?;
for _record in records {
// Process each record
}
consumer.commit_sync().await?;
}
}use krabka_client_consumer::{ShareAckMode, ShareAckType, ShareConsumer};
use krabka_units::secs;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// KIP-932 share group: records are acquired per consumer, with no partition assignment
let mut consumer = ShareConsumer::builder()
.bootstrap("localhost:9092")
.group_id("share-workers")
.subscribe(["jobs".to_string()])
.ack_mode(ShareAckMode::Explicit)
.build()
.await?;
loop {
let records = consumer.poll(secs(1)).await?;
for record in &records {
consumer.acknowledge(record, ShareAckType::Accept)?;
}
consumer.commit().await?;
}
}Streams Topology
The Processor API wires nodes by typed handle, so a mismatched edge fails at compile time. TopologyTestDriver runs the built topology in process, with no broker:
use krabka_client_streams::{
NodeHandle, Record, StringSerde, Topology, TopologyTestDriver, impl_processor,
};
struct Upper;
impl_processor! {
impl Upper: (String, String) -> (String, String) {
async fn process(&mut self, ctx, r) {
ctx.forward(Record::new(r.key, r.value.to_uppercase(), r.timestamp));
}
}
}
fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut topo = Topology::new();
let src: NodeHandle<String, String> = topo.add_source("src", ["in"]);
let up = topo.add_processor("up", || Upper, [&src]);
topo.add_sink("out", "out", [&up]);
let built = topo.build("my-app")?;
// Broker-free test driver: runs the same topology in process
let mut driver = TopologyTestDriver::new(&built)?;
driver.pipe_input("in", (StringSerde, StringSerde), Some("k".to_string()), "hello".to_string(), 0);
let out = driver.read_output("out", (StringSerde, StringSerde));
assert_eq!(out, Some((Some("k".to_string()), "HELLO".to_string())));
Ok(())
}