# kafrust
[](https://crates.io/crates/kafrust)
[](https://docs.rs/kafrust)
[](https://github.com/TaeeunKil/kafrust/actions/workflows/ci.yml)
A pure Rust Kafka client with no librdkafka or C client binding dependency.
`kafrust` is the high-level client crate in the kafrust workspace. It provides
Tokio-based admin, producer, direct consumer, and alpha classic/KIP-848
consumer group APIs on top of the companion
[`kafrust-protocol`](https://docs.rs/kafrust-protocol) wire-format crate.
Current release: `0.2.26`.
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` or C client binding. The default build
requires no C toolchain; the optional TLS feature may require native build
tooling for its `rustls` ring crypto provider.
- 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.
## Admin
```rust,no_run
use kafrust::{AdminClient, ClientConfig, CreateTopicsOptions, NewTopic};
# async fn example() -> kafrust::Result<()> {
let admin = AdminClient::new(ClientConfig::new(["localhost:9092"]));
let cluster = admin.describe_cluster().await?;
println!("controller: {}", cluster.controller_id());
let result = admin
.create_topics(
&[NewTopic::new("orders", 6, 3).config("cleanup.policy", "compact")],
CreateTopicsOptions::new(),
)
.await?;
for topic in result.topics() {
println!("{}: Kafka error {}", topic.name(), topic.error_code());
}
# Ok(())
# }
```
Cluster and topic listing use typed Metadata v1 views, and topic configuration
inspection and alteration preserve sources, sensitivity, synonyms, operation
kinds, validation mode, and partial resource errors. CreateTopics and
DeleteTopics route through the active controller, while group descriptions
and deletion route through each group coordinator. Group listing queries all
advertised brokers. ACL describe/create/delete operations preserve typed
bindings and partial authorization results; qualify them against an
authorizer-enabled broker before production rollout. Client quota
describe/alter operations preserve typed entities, floating-point values, and
per-entity results; both paths are live-verified in the documented Kafka
3.7.2 StandardAuthorizer profile. Consumer-group offset listing and
administrative alteration use typed classic OffsetFetch v2 and OffsetCommit v2
results. See the repository's `docs/admin-api.md` for details.
## Install
```toml
[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
```rust,no_run
use kafrust::{Acks, Compression, 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(())
}
```
When no partition is specified, keyed records use Kafka-compatible Murmur2
partitioning and keyless records use the producer's batch-sticky routing. A
custom `ProducerConfig::partitioner` callback can override routing for records
without explicit partitions; explicit partitions always take precedence.
## 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.
```rust,no_run
use kafrust::{Acks, Compression, 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)
.compression(Compression::Gzip)
.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(())
}
```
`Compression::Gzip`, `Compression::Snappy`, `Compression::Lz4`, and
`Compression::Zstd` prefer topic-ID Produce API v13 when Metadata v12 supplies
a non-zero topic UUID, then flexible Produce API v12, v11, and v9, RecordBatch
encoding when available, then fall back to Produce API v7 or v3. Brokers without the required
Produce API version return an explicit `Unsupported` error when compression is
enabled.
Snappy output uses Kafka-compatible Xerial framing; LZ4 and Zstd output use
their standard frames as expected by RecordBatch v2. Brokers without the
required Produce API version return an explicit `Unsupported` error when
compression is enabled.
## Transactional Producer
Use `transactional_id` to enable the alpha transactional producer:
```rust,no_run
use kafrust::{ProducerConfig, ProducerRecord};
#[tokio::main(flavor = "current_thread")]
async fn main() -> kafrust::Result<()> {
let mut producer = ProducerConfig::new(["localhost:9092"])
.transactional_id("orders-writer")
.build()
.await?;
producer.begin_transaction()?;
producer
.send(ProducerRecord::to("orders").value("committed value"))
.await?;
producer.commit_transaction().await?;
Ok(())
}
```
The high-level commit, abort, read-committed isolation, and transactional
consumer group offset paths are verified against Kafka `3.7.2` and `4.3.1`.
`Producer::transaction_status()` and the matching buffered-producer method
expose `Ready`, `InTransaction`, and terminal `Defunct` states. If an `EndTxn`
response is lost,
`commit_transaction` or `abort_transaction` returns
`Error::TransactionOutcomeUnknown`; discard that producer because Kafka may
have applied the requested outcome, and create a new producer with the same
transactional ID for recovery. The old transaction is never reported as
successfully committed or aborted from a lost response.
## 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`.
```rust,no_run
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)
.buffer_capacity(1024)
.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.
```rust,no_run
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(())
}
```
Fetched records expose Kafka RecordBatch headers through `record.headers()`.
`ConsumerRecordHeader::value()` returns an optional byte slice because Kafka
allows null header values; legacy MessageSet records have an empty header list.
Direct and group consumers negotiate Fetch v12, then v11, when the selected
broker advertises those versions. For rack-aware reads, set
`ConsumerConfig::client_rack("rack-a")` (or the matching
`ConsumerGroupConfig` builder). Fetch v12 carries the rack ID using the flexible
schema and follows Kafka's `preferred_read_replica` response when supported;
without a rack, the same session-capable path uses an empty rack ID. Fetch v4
remains the fallback for brokers that do not advertise v11 or v12. The protocol
and injected routing tests pass. The Kafka 3.7.2 three-broker `broker.rack` plus
`RackAwareReplicaSelector` profile is live-qualified in
[`31640494509`](https://github.com/TaeeunKil/kafrust/actions/runs/31640494509).
Fetch v11/v12 requests reuse a broker-scoped fetch session across sequential
polls. Assignment changes, seek/pause/resume, reconnects, and fetch errors
discard the session; the Fetch v4 fallback does not claim session reuse.
The complete 17-job matrix for the general direct-consumer negotiation passed in
[`31673377685`](https://github.com/TaeeunKil/kafrust/actions/runs/31673377685).
The complete 17-job matrix for this path passed in
[`31671783977`](https://github.com/TaeeunKil/kafrust/actions/runs/31671783977),
including the Kafka 3.7.2 three-broker rack-aware follow-up request.
## Consumer Group
The consumer group API is an alpha classic or KIP-848 consumer group path with
dynamic or static membership, range, round-robin, or opt-in cooperative-sticky
assignment for classic groups, join, heartbeat, poll, offset fetch/commit, and
explicit leave support. The
cooperative-sticky path includes protocol, staged assignment, multi-member
ownership transfer, transient-member rollback, and member-loss recovery. These
cooperative failure paths are live-verified in the Kafka `3.7.2` three-broker
profile. The KIP-848 path, including OffsetFetch v9, OffsetCommit v9, and
background heartbeat rejoin, is live-verified against Kafka `4.3.1`; the group
API itself remains pre-`1.0`.
```rust,no_run
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")
.group_instance_id("orders-reader-1")
.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 transport. TLS transport is available only when the
non-default `tls` crate feature is enabled:
```toml
kafrust = { version = "0.2", features = ["tls"] }
```
Without that feature, `SecurityProtocol::Tls` returns `Error::Unsupported`
before connecting.
TLS server name validation defaults to the bootstrap host. Use
`tls_server_name(name)` on `ClientConfig`, `ProducerConfig`, `ConsumerConfig`, or
`ConsumerGroupConfig` when the bootstrap address differs from the broker
certificate subject alternative name.
Use `tls_root_certificate_der(bytes)` to add DER-encoded root certificates while
still keeping platform roots enabled.
SASL credentials can be stored on the shared client configuration with
`sasl_plain(username, password)`, `sasl_scram_sha_256(username, password)`, or
`sasl_scram_sha_512(username, password)`. This does not change the selected
`SecurityProtocol`; choose `SaslPlaintext` or `SaslTls` separately. Credential
debug output redacts the password. The SASL/PLAIN `SaslPlaintext` broker
roundtrip, producer, direct consumer, and consumer group smoke paths are
verified against Kafka `3.7.2`; SASL/SCRAM-SHA-256 and SCRAM-SHA-512 over
`SaslTls` are verified for the documented live smoke paths. The SHA-512 profile
covers broker roundtrip, producer, batch, buffered producer, direct consumer,
and consumer group poll paths.
SASL/OAUTHBEARER token authentication is available through
`sasl_oauthbearer(token)` or
`sasl_oauthbearer_with_username(username, token)`. Async token providers are
available through `sasl_oauthbearer_provider` and
`sasl_oauthbearer_with_username_and_provider`; kafrust calls them for each new
broker authentication. The RFC 7628 initial response and secret redaction are
covered by injected handshake tests, and Kafka 3.7.2 SASL_SSL is covered by a
dedicated smoke using the broker's built-in unsecured validator (`Live Kafka
Smoke` run `31478375106`). OAUTHBEARER initial authentication and provider
re-authentication use Kafka's flexible `SaslAuthenticate v2` wire response;
PLAIN and SCRAM continue to use `v1`. The low-level `Client` exposes the
broker-advertised `session_lifetime_ms` for an application-owned refresh
schedule. Provider-backed OAUTHBEARER connections also re-authenticate on the
existing connection before requests after half of that lifetime. A signed
JWT/JWKS policy passes the Kafka 3.7.2 OIDC fixture, Java client, static-token,
and provider-backed paths in the [`Live Kafka Smoke` OIDC job](https://github.com/TaeeunKil/kafrust/actions/runs/31584760474/job/94075906934),
while external provider-specific behavior and detached refresh workers remain
unclaimed. Provider callbacks are bounded by
`ClientConfig::request_timeout_ms`; an expired callback returns
`Error::OAuthBearerTokenTimeout` without including token material.
The default build does not include TLS dependencies. The current `tls` feature
uses the `rustls` ring crypto provider, which can require native build tooling
in some environments; this is not part of the default kafrust toolchain.
## Request Metrics
`ClientMetrics` provides lock-free operational counters shared by all broker
connections and buffered producers created from a configuration:
```rust,no_run
use kafrust::{ClientMetrics, ProducerConfig};
# async fn example() -> kafrust::Result<()> {
let metrics = ClientMetrics::new();
let producer = ProducerConfig::new(["localhost:9092"])
.metrics(metrics.clone())
.build()
.await?;
// Send records with `producer`, then export a point-in-time snapshot.
let snapshot = metrics.snapshot();
println!(
"requests={} failures={} broker_errors={} max_latency={:?}",
snapshot.requests_started,
snapshot.requests_failed,
snapshot.broker_errors,
snapshot.max_latency,
);
# Ok(())
# }
```
Snapshots include request success, failure, timeout, and cancellation counts,
non-zero Kafka error codes observed in decoded broker responses, high-level
operation retry attempts, request and response payload bytes, in-flight
requests, current and maximum outstanding buffered records, and total and
maximum latency. Successful produced records, topic-partition Produce chunks,
and records returned by consumer APIs are also counted. Retry attempts cover
producer sends, consumer fetches, metadata reconnects, and transactional
coordinator operations, plus automatic consumer-group rejoins. Individual
atomic fields are sampled independently, so a snapshot taken while requests
are changing is not a transactional view. Request metrics and `tracing` spans
contain operational metadata only; key, value, request, and response payload
contents are not recorded.
Debug-level operation spans cover immediate and buffered producer sends,
flush/close, batch sends, transaction completion and offset attachment,
direct-consumer poll/fetch, and consumer-group join, poll, heartbeat, and
offset commit. Their names use the `kafka.producer.*`, `kafka.consumer.*`, and
`kafka.consumer_group.*` prefixes. Broker `kafka.request` spans execute as
children, so subscribers can attribute wire latency to one user operation.
The buffered producer command queue defaults to 1024 records. Configure it with
`ProducerConfig::buffer_capacity`; values below one become one. When the queue
is full, `BufferedProducer::send` waits for capacity instead of allocating an
unbounded queue or dropping an accepted record.
## Decode Memory Limits
Broker response frame allocation is limited to `100 MiB` by default. Set
`max_response_bytes(bytes)` on `ClientConfig`, `ProducerConfig`,
`ConsumerConfig`, or `ConsumerGroupConfig` when a workload needs a different
limit. A broker frame length above the configured limit returns
`Error::ResponseTooLarge { size, max }` before allocating the response payload.
Kafka arrays are limited to `1,000,000` elements and an uncompressed fetched
record batch is limited to `64 MiB` by default. Configure these limits with
`max_decode_array_elements(elements)` and
`max_decompressed_record_bytes(bytes)` on the same configuration builders.
Declared arrays, record counts, and record headers are checked before reserving
their vectors. Gzip, Snappy, LZ4, and Zstd decoders enforce the batch limit and
return `protocol::Error::LimitExceeded { kind, actual, max }`.
The limit applies independently to each broker request. It must be large enough
for metadata, fetch, and other expected responses; setting it below a valid
response size fails that operation instead of partially decoding it.
## Configuration Validation
High-level builders validate configuration before opening a broker connection.
Call `ClientConfig::validate`, `ProducerConfig::validate`,
`ConsumerConfig::validate`, or `ConsumerGroupConfig::validate` to run the same
startup preflight explicitly before deciding when to connect.
Invalid timeouts, response or decode limits, negative fetch values, zero poll
limits, fetch bounds, poll limits, group IDs or subscriptions, zero commit or
heartbeat intervals, and invalid transaction settings return
`Error::InvalidConfiguration { field, reason }`. Empty bootstrap lists retain
the dedicated `Error::MissingBootstrapServer` variant, while blank bootstrap
entries are reported as invalid configuration. This makes startup failures
deterministic and avoids turning local configuration mistakes into network
retries.
## Compatibility
The `0.2.x` alpha line is verified against single-node Apache Kafka `3.7.2`,
`3.8.1`, `3.9.1`, and `4.3.1` KRaft brokers over `PLAINTEXT`. Secured and
three-broker profiles are verified against Kafka `3.7.2`.
Verified high-level paths include:
- `ApiVersions v0` and flexible `ApiVersions v3` capability roundtrips, plus
`Metadata v1` roundtrips.
- Producer single-record, batch, and buffered sends.
- Direct topic-partition fetch using Fetch v4 response decoding.
- Classic consumer group join, sync, heartbeat, poll, and offset commit.
- Cooperative-sticky multi-member transfer, transient-member rollback, and
member-loss recovery in the Kafka `3.7.2` three-broker profile.
## Current Limits
- APIs are pre-`1.0` and can change between minor versions.
- TLS is feature-gated, currently uses the `rustls` ring crypto provider, and
is verified for the broker roundtrip, producer, direct consumer, and consumer
group smoke paths against Kafka `3.7.2`;
SASL/PLAIN is verified for the `SaslPlaintext` broker roundtrip, producer,
direct consumer, and consumer group smoke paths.
- SASL/SCRAM-SHA-256 and SCRAM-SHA-512 are verified over `SaslTls` for the
documented broker roundtrip, producer, direct consumer, and consumer group
smoke paths; the SHA-512 profile also covers batch and buffered producer
paths.
- SASL/OAUTHBEARER is implemented with token-only and authorization-identity
builders plus async token-provider builders, and live-verified against Kafka
3.7.2's built-in validator and a signed local OIDC/JWKS fixture. Flexible
SASL Authenticate v2 session-lifetime metadata is exposed by the low-level
client, and provider-backed connections re-authenticate before requests as
the session expires. External provider-specific behavior and detached
refresh workers remain unclaimed.
- Broker compatibility is verified against Kafka `3.7.2`, `3.8.1`, `3.9.1`,
and `4.3.1` for the single-node plaintext profile. Secured and multi-broker
profiles are verified against `3.7.2`.
- Multi-broker clusters, coordinator and leader failover, and partition
expansion are verified in the documented `3.7.2` profiles. Rack-aware client
routing prefers flexible Fetch v12, falls back through Fetch v11 to Fetch v4,
and follows the preferred-replica response. The three-broker `3.7.2`
rack-aware profile is live-qualified in
[`31640494509`](https://github.com/TaeeunKil/kafrust/actions/runs/31640494509).
- Idempotent single-record, batch, and buffered sends are available through
`ProducerConfig::enable_idempotence(true)`. Transactional immediate and batch
sends support explicit begin, commit, and abort.
`IsolationLevel::ReadCommitted` hides aborted transaction records for direct
and group consumers, and current group assignments can be committed through
generation-fenced `Producer::send_group_offsets_to_transaction`.
Transactional buffered sends and controller-routed partition reassignment are
available. SCRAM credential administration and partition reassignment are
available through typed `AdminClient` APIs; both have Kafka 3.7.2 live smoke
paths, including three-broker reassignment completion polling.
Shared request, retry, broker-error, producer, consumer, batch, and
buffered-queue metrics are available together with high-level operation and
request spans.
- `acks=0` fire-and-forget sends are supported for immediate and batch
producer paths. The request is written and flushed without waiting for a
broker response, so returned offsets are `-1` and broker acceptance or
partition-level errors cannot be confirmed. This path is live-verified
against Kafka `3.7.2`, `3.8.1`, `3.9.1`, and `4.3.1` single-node plaintext
profiles.
## Examples
Run examples from the repository with a local Kafka broker:
```sh
KAFRUST_BOOTSTRAP_SERVERS=localhost:9092 \
KAFRUST_TOPIC=kafrust-smoke \
cargo run -p kafrust --example producer_send
```
The smoke examples accept `KAFRUST_SECURITY_PROTOCOL`,
`KAFRUST_SASL_USERNAME`, `KAFRUST_SASL_PASSWORD`, and
`KAFRUST_SASL_MECHANISM` so the same examples can be run against plaintext,
TLS, and SASL broker profiles. `KAFRUST_SASL_MECHANISM` defaults to `plain` and
also accepts `scram-sha-256`, `scram-sha-512`, or `oauthbearer`. The
OAUTHBEARER path reads its token from `KAFRUST_SASL_TOKEN` and treats
`KAFRUST_SASL_USERNAME` as an optional authorization identity. Set
`KAFRUST_SASL_TOKEN_PATH` instead to use the example's provider-backed file
callback, which reads a fresh token for each broker authentication.
`KAFRUST_BOOTSTRAP_SERVERS` accepts Kafka's comma-separated bootstrap format for
multiple brokers, for example `localhost:19092,localhost:19093`.
Available examples include:
- `broker_roundtrip`
- `producer_send`
- `producer_send_batch`
- `producer_buffered`
- `producer_transactional`
- `consumer_fetch`
- `find_group_coordinator`
- `consumer_group_poll`
- `admin_reassign_partitions`
## Project Docs
- Repository: <https://github.com/TaeeunKil/kafrust>
- Roadmap: <https://github.com/TaeeunKil/kafrust/blob/main/docs/roadmap.md>
- Compatibility: <https://github.com/TaeeunKil/kafrust/blob/main/docs/compatibility.md>
- API stability: <https://github.com/TaeeunKil/kafrust/blob/main/docs/api-stability.md>
- Public API audit: <https://github.com/TaeeunKil/kafrust/blob/main/docs/public-api-audit.md>
- Project strategy: <https://github.com/TaeeunKil/kafrust/blob/main/docs/project-strategy.md>