kafka-rust

rustfs-kafka

Fork project: forked from kafka-rust.

rustfs-kafka is a Rust Kafka client workspace containing:

• rustfs-kafka: synchronous client/producer/consumer/admin APIs.
• rustfs-kafka-async: async wrapper crate based on tokio.

Current release target: 1.4.0.

Crates

[dependencies]
rustfs-kafka = "1.4.0"
rustfs-kafka-async = "1.4.0"

Core Features

• Kafka client metadata, fetch, produce, offset commit, committed-offset deletion, API version, cluster/config inspection/mutation, config-resource discovery, topic partition discovery, ACL inspection/mutation, delegation token lifecycle, client quota inspection/mutation, SCRAM credential mutation, broker log directory reassignment, KRaft quorum/feature/broker lifecycle/voter admin, replica directory assignment, partition reassignment query/mutation, partition expansion, record deletion, leader election, leader-epoch offsets, active producer, transaction offset commit, consumer group deletion/inspection, and share group inspection/mutation APIs.
• Typed raw kafka-protocol request support for advanced generated protocol messages that do not have a stable high-level client workflow.
• Runtime building blocks for telemetry subscription tracking and share-consumer request composition.
• High-level Consumer and Producer abstractions.
• TLS support via rustls:
  • security (default, aws-lc-rs provider, webpki-roots trust store)
  • security-ring (ring provider, webpki-roots trust store)
• Custom/private CAs are supported through SecurityConfig::with_ca_cert; system native root stores are not loaded by default.
• Async security authentication support includes SASL PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 over TLS.
• Optional metrics support.
• Optional producer_timestamp.
• Integration test harness with Kafka 3.9.2, 4.1.2, and 4.2.0.

Feature Flags (rustfs-kafka)

Feature	Default	Description
security	Yes	rustls + aws-lc-rs TLS backend
security-ring	No	rustls + ring TLS backend
compression	Yes	gzip, snappy, lz4, and zstd codecs
gzip	Yes	gzip record batch codec
snappy	Yes	snappy record batch codec
lz4	Yes	lz4 record batch codec
zstd	Yes	zstd record batch codec
metrics	No	metrics integration
producer_timestamp	No	producer timestamp support
nightly	No	nightly-only optimizations
integration_tests	No	integration test compilation helpers

Default builds include Kafka record batch compression support. For smaller builds, disable default features and enable only the codecs you need, for example features = ["security", "gzip"].

Runtime Behavior

• Sync request frames are written completely and flushed before waiting for a response.
• Fetch responses decode every record batch in each partition; Produce batches use contiguous relative offsets.
• Sync consumers honor pause/resume in regular and retry fetches. Committed offsets at the earliest retained position remain valid. Failed polls preserve fetch progress and outstanding retries.
• Async consumers initialize from committed offsets, then use the configured fallback for partitions without a committed position. A failed or cancelled multi-broker poll does not advance progress for unreturned messages.
• Sticky partitioning maintains separate state per topic and reselects when a partition becomes unavailable.
• Producer builders preserve TLS, client ID, and other configuration when a custom partitioner is selected; remaining options and client accessors stay available afterward.
• Async producers borrow cached partition routes. A broker leader error invalidates the affected topic route; the next send refreshes metadata and the failed send returns its original error.
• AsyncProducer::send_all batches records by broker, topic, and partition, sending one Produce request per broker. It preserves partition order and supports the same compression and acknowledgement settings as send.
• Batch producers retain unconfirmed records after failed flushes and retire only uniquely confirmed partitions. Further send calls require an explicit flush or clear while a failed batch remains pending.
• Async typed requests validate correlation and consume the complete response. Interrupted or failed connections are discarded and reconnected on the next checkout; Produce errors are returned without automatic replay.
• list_offsets preserves broker timestamps, and sync connection selection reconnects only the selected broker.
• Sync typed responses validate the pending correlation, requested version, and complete payload before reuse. Fetching an unknown topic or partition returns an error before sending any broker request.
• With producer_timestamp enabled, sync CreateTime timestamps use the current Unix milliseconds once per Produce call. The default retains zero timestamps. Configure LogAppendTime on the broker; selecting it on a sync producer returns a configuration error before network activity.
• Retry backoff caps overflow at the configured maximum and rejects non-finite or nonpositive multipliers.
• Async Fetch shares the sync decoder, consuming every batch and rejecting corrupt tails before publishing offsets. Coordinator errors during offset initialization invalidate the cached coordinator and use bounded rediscovery.
• Sync Produce prepares all broker frames before IO and maps confirmations directly. Negative offsets in successful ACKs become failed confirmations, preserving those records in buffered producers.
• Admin mutations return errors after a sending attempt without automatic replay; explicitly read-only APIs retain broker failover. Response buffer reservations fail with errors and retire incomplete connections.
• High-level consumers filter records before the requested offset, including recovery inside a record batch, and reject invalid offsets or incomplete partition responses before publishing progress.
• Sync seek rejects negative or unassigned cursors; consumed offsets must belong to an assigned partition. An OOR response with an unknown negative high watermark preserves the prior cursor and retry.
• MessageSets::iter_ref borrows topic names and message slices; the existing owned iterator remains available.
• Metadata replacement preserves the previous snapshot on failure, clears dependent coordinator caches on success, and uses newly discovered coordinator endpoints. Offset Fetch/Commit responses require complete acknowledgements.
• Ordinary producers reject unsupported idempotence/transaction options at creation. Record encoding rejects duplicate header keys before routing or IO because the current codec cannot preserve their ordered values.
• Native async startup batches unresolved fallback positions by broker and publishes initialization only after all lookups succeed. Checked routing and coordinator recovery preserve progress across failures.
• Sync ListOffsets validates complete responses while preserving raw unknown offsets. High-level consumers reject an unresolved starting position, and sync fallback resets stale consumed markers so new progress can be committed.
• Small header preflight uses borrowed comparisons for two to four keys; larger inputs retain hash-based validation.
• Fetch hides transaction control records and advances through empty or compacted batches using verified batch boundaries. Existing owned response fields and explicit sync consumption remain compatible.
• Ordinary producers require complete target acknowledgements and return errors for malformed confirmations. Async producers pre-encode every complete broker frame before starting Produce IO.

TransactionalProducer uses a separate transaction coordinator and sends transaction IDs, producer identities, epochs, and sequences in transactional batches. Commit and abort preserve sequence state. A transaction RPC failure blocks further reuse; construct a new producer instead of retrying that instance. Transactional consumer-offset workflows and read-committed high-level consumers remain follow-up work. GroupCoordinator discovers coordinators and uses each member’s subscription for assignments; callers schedule its manual heartbeat method. Automatic heartbeat and rebalance scheduling remain follow-up work. Further lifecycle work is tracked in rustfs/backlog#2713.

Documentation

• API docs: docs.rs/rustfs-kafka
• Workspace docs index: docs/README.md
• Usage guide (sync + async): docs/usage-guide.md
• Async crate readme: crates/rustfs-kafka-async/README.md

Local Development

The child crate manifests omit explicit rust-version metadata. The workspace retains its compiler reference alongside the Rust 2024 edition and dependency declarations in Cargo.toml.

cargo build
cargo test
cargo clippy --all-targets --all-features -- -D warnings

Integration tests (Docker required):

cd crates/rustfs-kafka/tests
./run-all-tests
./run-sync-secure-tests
./run-async-secure-tests

License

Apache License 2.0. See LICENSE.

This site is open source. Improve this page.