Apache Kafka · Rust · Tokio
Pure Rust on Tokio. No C library and no system dependency in the default build, no unsafe, no panics on a malformed response. Protocol versions tracked against Apache Kafka 4.3 and checked in CI against Kafka's own schemas.
cargo add krafkacargo add krafka needs a Rust toolchain and the C compiler ring uses — no librdkafka, no OpenSSL, no CMake, no pkg-config, no system library. ring is the only crate in the default build that compiles C; zstd encoding, the aws-lc-rs backend and MSK IAM are opt-in features that add more.
Unsafe code is denied crate-wide, as are panic, unwrap and expect. A malformed broker response surfaces as an error, never a panic; every allocation from untrusted input is bounded twice, by the declared count and by the bytes actually available.
Apache Kafka 4.3 API versions, including KIP-848 consumer groups and KIP-932 share groups. A CI job diffs krafka's version table against Kafka's own message schemas.
An idempotent producer by default that never reuses a sequence range, a timeout that says whether the record may have been written, a transaction that cannot commit after a failed send, and KIP-320 truncation detection on Fetch, ListOffsets and OffsetCommit.
cargo add krafka
cargo add tokio --features fulluse krafka::{Kafka, Record};
#[::]
async fn main() -> krafka::Result<()> {
let kafka = Kafka::builder("localhost:9092").connect().await?;
let producer = kafka.producer().build().await?;
producer.send(Record::new("orders", "hello").key("key")).await?;
producer.close().await
}use krafka::Kafka;
use krafka::consumer::AutoOffsetReset;
#[::]
async fn main() -> krafka::Result<()> {
let kafka = Kafka::builder("localhost:9092").connect().await?;
let consumer = kafka
.consumer("order-processor")
.auto_offset_reset(AutoOffsetReset::Earliest)
.build()
.await?;
consumer.subscribe(["orders"]).await?;
while let Some(record) = consumer.recv().await? {
println!("{}-{} @ {}", record.topic, record.partition, record.offset);
}
Ok(())
}Clients. A producer with batching, compression, idempotence and exactly-once transactions. A consumer supporting both group protocols — classic and KIP-848 server-side assignment. A share consumer for queue-like per-record acknowledgement (KIP-932). An admin client: topics, partitions, configs, ACLs, quotas, groups, delegation tokens, SCRAM credentials and cluster features.
Security. TLS and mTLS over rustls with hot certificate reload (KIP-1288). SASL PLAIN, SCRAM-SHA-256/512 and OAUTHBEARER, including a built-in OIDC token provider covering both the client-secret flow (KIP-768) and RFC 7523 client assertions (KIP-1258), with no JWT or RSA dependency. Secrets are zeroized, credential comparison is constant-time, and no credential-bearing type derives Debug. See Authentication and Cloud Platforms.
Operability. metrics() on every client returns one metrics snapshot with Prometheus text output; spans go through tracing on OpenTelemetry messaging conventions; KIP-714 pushes client metrics to brokers that subscribe. Interceptors for redaction and auditing. Connection settings — keepalive, the response ceiling, the in-flight cap, the connection cap — are set once on the Kafka handle and shared by every client built from it.
Testing. An in-process fake broker behind the test-broker feature speaks the wire protocol with fault injection (leader moves, coordinator failover, late responses, controller churn), needs no Docker and runs on Tokio's paused clock. See Testing. Coming from rdkafka? Read Migrating from rdkafka.