Kafkaesque

A cloud-native, Rust-based Kafka-compatible broker backed by object storage (S3, GCS, Azure) with embedded Raft consensus.

Kafkaesque speaks the Kafka wire protocol, so standard Kafka clients work without modification. It uses object storage for durability and embedded Raft for coordination — no Zookeeper, no external metadata service, no local-disk reliability requirement.

Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Kafka Client   │     │  Kafka Client   │     │  Kafka Client   │
└────────┬────────┘     └────────┬────────┘     └────────┬────────┘
         │                       │                       │
         └───────────────────────┼───────────────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │   Kafkaesque Broker(s)  │
                    └────────────┬────────────┘
                                 │
                   ┌─────────────┼─────────────┐
                   │             │             │
         ┌─────────▼──────┐  ┌───▼────────┐  ┌▼──────────────┐
         │  Raft Layer    │  │ SlateDB    │  │ Object Store  │
         │ (Metadata &    │  │ (LSM, hot  │  │ (S3, GCS,     │
         │  Coordination) │  │  data)     │  │  Azure)       │
         └────────────────┘  └────────────┘  └───────────────┘

Key features:

Kafka compatible — works with librdkafka, Java, sarama, kcat, …
Object-store native — durability comes from S3/GCS/Azure; brokers are effectively stateless.
Embedded Raft — no Zookeeper. Brokers coordinate leader election and metadata over an authenticated internal Raft channel.
Single binary — zero external dependencies at runtime.
LSM-tree storage — built on SlateDB for efficient writes against object stores.

Quick Start

Building from source needs the Rust toolchain pinned in rust-toolchain.toml (Rust 1.91, edition 2024). rustup picks it up automatically. Running under Docker needs only Docker, but there is no published image yet — you build it locally (see below).

Run locally (standalone)

CLUSTER_PROFILE=development cargo run --release -p kafkaesque-bin --bin kafkaesque

The broker listens on localhost:9092 for Kafka traffic and localhost:8080 for health and metrics. Data is stored under /tmp/kafkaesque-data.

CLUSTER_PROFILE=development opts into an unauthenticated Raft control plane for local use. Outside the development profile the broker refuses to start unless RAFT_CLUSTER_SECRET is set — see Security.

Run with Docker

No image is published to a registry yet, so build one first:

docker build -f Dockerfile.minimal -t kafkaesque:latest .

docker run --rm -p 9092:9092 -p 8080:8080 \
  -e CLUSTER_PROFILE=development \
  -e OBJECT_STORE_TYPE=local \
  -e DATA_PATH=/data \
  -v /tmp/kafkaesque-data:/data \
  kafkaesque:latest

Smoke test with `kcat`

echo "hello" | kcat -P -b localhost:9092 -t demo
kcat -C -b localhost:9092 -t demo -e

Workspace layout

The repository is a Cargo workspace with three crates:

Crate	Purpose
`kafkaesque-protocol` (in `crates/kafkaesque-protocol/`)	Runtime-independent Kafka wire-protocol layer — parser, encoder, types, constants, wire-error. No tokio/slatedb/openraft, so it can be reused on its own.
`kafkaesque` (root crate)	The umbrella: cluster coordination, server, runtime, telemetry. Re-exports the protocol layer at the same module paths it used to live at, so existing `use kafkaesque::{parser, encode, types, ...}` paths keep working.
`kafkaesque-bin` (in `crates/kafkaesque-bin/`)	The production broker binary. Build via `cargo build -p kafkaesque-bin --bin kafkaesque`.

MockCoordinator and other test helpers live behind the test-utilities feature on the umbrella crate

Configuration

Configuration is handled entirely via environment variables. Common knobs:

Category	Variable	Default	Description
Core	`BROKER_ID`	`0`	Unique broker ID; must be persistent across restarts.
	`HOST`	`0.0.0.0`	Bind address for the Kafka listener.
	`PORT`	`9092`	Kafka protocol port.
	`ADVERTISED_HOST`	(value of `HOST`)	Hostname returned to clients in `Metadata` responses. Set this when the broker is reachable at a different address than it binds to (NAT, load balancer, Kubernetes).
	`CLUSTER_ID`	`kafkaesque`	Logical cluster identifier.
Storage	`OBJECT_STORE_TYPE`	`local`	`local`, `s3`, `gcs`, or `azure`.
	`DATA_PATH`	`/tmp/kafkaesque-data`	Path or object-store prefix for data.
Raft	`RAFT_PEERS`	(empty)	Peer list, e.g. `0=host:port,1=host:port`.
	`RAFT_LISTEN_ADDR`	`127.0.0.1:9093`	Internal Raft RPC address.
Security	`CLUSTER_PROFILE`	`production`	Deployment profile. Only `development` permits running without Raft auth.
	`RAFT_CLUSTER_SECRET`	(required outside dev)	Shared HMAC key authenticating all Raft RPC traffic. Identical on every node.
	`SASL_ENABLED`	`false`	Enable SASL auth on the Kafka listener (needs `--features sasl`).
	`SASL_USERS_FILE`		Path to the SASL user/credential file.
	`TLS_ENABLED`	`false`	Enable TLS on the Kafka listener (needs `--features tls`).
	`TLS_CERT_PATH` / `TLS_KEY_PATH`		PEM cert and key for TLS termination.
S3	`S3_BUCKET`		AWS S3 bucket name.
	`AWS_REGION`		AWS region (e.g. `us-east-1`).
GCS	`GCS_BUCKET`		Google Cloud Storage bucket name.
	`GOOGLE_APPLICATION_CREDENTIALS`		Path to the GCP service-account JSON.
Azure	`AZURE_STORAGE_ACCOUNT`		Azure storage account name.
	`AZURE_CONTAINER`		Azure blob container name.
Health	`HEALTH_PORT`	`8080`	Port for `/health`, `/ready`, `/live`, `/metrics`. Set to `0` to disable.
	`METRICS_AUTH_TOKEN`		When set, `/metrics` requires `Authorization: Bearer <token>`.
Telemetry	`LOG_FORMAT`	`pretty`	`pretty` or `json` (for log aggregators).
	`RUST_LOG`	`info`	Standard `tracing-subscriber` filter.
	`OTEL_EXPORTER_OTLP_ENDPOINT`	`http://localhost:4317`	OTLP collector endpoint. Requires `--features otel`.
	`OTEL_SERVICE_NAME`	`kafkaesque-kafka-broker`	Service name attached to exported spans.

This is a curated subset; see src/cluster/config.rs for the full set of tunables (thread pools, cache TTLs, retention, flush windows, ACLs, …).

Health & monitoring

Each broker runs a lightweight HTTP server on HEALTH_PORT (default 8080, bound to HOST) exposing:

Endpoint	Purpose	Notes
`GET /health` (`/healthz`)	Liveness	Always `200` while the process is running. Use as the Kubernetes liveness probe.
`GET /live` (`/livez`)	Liveness	Alias of `/health`.
`GET /ready` (`/readyz`)	Readiness	`503` when the broker is in zombie mode (fenced / not serving); `200` otherwise. Use as the readiness probe.
`GET /metrics`	Prometheus metrics	Text exposition format. Gated by `METRICS_AUTH_TOKEN` when set.

Set HEALTH_PORT=0 to disable the health server entirely.

Durability contract

Replication factor and ISR — read this first

Kafkaesque does not replicate between brokers. Each partition is stored once, in the object store, and has exactly one live owner at a time. Durability is the backing bucket's job (S3-class storage replicates underneath you), not a follower set's.

Concretely, if you are used to Kafka:

Kafka concept	Here
`replication.factor`	Only `1` (or `-1`, "broker default") is accepted. `CreateTopics` with `replication_factor > 1` is rejected with `INVALID_REPLICATION_FACTOR`.
ISR (in-sync replicas)	Metadata reports `replicas: [leader]` and `isr: [leader]` to satisfy the wire format. There is no follower set to shrink or expand.
`min.insync.replicas`	Not implemented; has no effect.
`acks=all`	Identical to `acks=1`. It waits for one durable object-store write — it does not wait for peer brokers.
Broker loss	The partition's lease expires and another broker takes ownership, reading the same object-store data. Availability, not durability, is what failover buys you.

This is a deliberate design (the same shape as other object-store-backed Kafka implementations), not a missing feature — but it means your bucket's durability is your data's durability. Use a bucket with cross-AZ or cross-region replication if you need more than one failure domain. A single local OBJECT_STORE_TYPE=local directory has exactly one.

The broker refuses replication_factor > 1 rather than accepting it and reporting a one-node ISR, so a client cannot ask for quorum durability and be told it got it.

`acks`

Kafkaesque honors the standard Kafka acks semantics. Every produce request opts into one of three durability levels by setting acks:

`acks`	Behavior	What can be lost on a hard kill
`0`	Fire-and-forget. Broker replies immediately and writes asynchronously.	Up to ~100 ms of unacked records (one SlateDB flush window).
`1`	Broker waits for the WAL to fsync before acking.	Nothing the broker has acked.
`all`	Same as `acks=1` for a single broker; identical durability across replicas because the data plane is the object store, not peer brokers.	Nothing the broker has acked.

Beyond the acks flag, two additional internal writes are always durable, regardless of the producer's choice:

Idempotent producer state. When a record batch carries a producer_id, the broker persists the producer's last sequence number and base offset in the same SlateDB write batch as the records themselves, with await_durable: true. This ensures the exactly-once dedup window survives crashes — an idle producer whose in-memory state was evicted cannot have a fresh-looking sequence-0 batch admitted as new after a restart.
Leader-epoch fencing token, log-start offset, and snapshot pointer. All written with await_durable: true.

Graceful shutdown

PartitionManager::shutdown flushes every owned partition store before releasing its lease and unregistering the broker. SIGTERM / SIGINT trigger this path; a hard kill (SIGKILL, OOM-kill) does not. On a hard kill, only the acks=0 window can lose acknowledged data.

Property test

tests/durability_contract_props.rs drives a randomized produce schedule (varying batch sizes and acks levels) interleaved with randomized "crashes" — re-opening the SlateDB instance against the same backing object store without first running graceful shutdown. The test asserts that every offset returned by append_batch_durable is still readable after the crash. Run with:

cargo test --test durability_contract_props

Security

The Raft RPC port accepts cluster-membership changes and coordination commands, so it must never be exposed unauthenticated. The broker is secure by default: startup fails unless RAFT_CLUSTER_SECRET is set to a non-empty value, or you explicitly opt out with CLUSTER_PROFILE=development for local work.

# Generate once, distribute the same value to every node:
export RAFT_CLUSTER_SECRET="$(openssl rand -base64 32)"

For the public Kafka listener, enable SASL/PLAIN and TLS at compile time:

cargo build --release --features sasl,tls

Securing the Kafka listener

The Raft port and the Kafka port are secured independently, and it is worth being explicit about what each default gives you. Outside CLUSTER_PROFILE=development the broker refuses to start without RAFT_CLUSTER_SECRET, RAFT_JOIN_TOKEN, and ACL_ENABLED — so authorization is always on. What is not required is authentication on the Kafka listener: a broker with ACLs but no SASL sees every client as the single principal User:ANONYMOUS. That is a deliberate allowance for internal-only clusters sitting behind a network boundary, but it means ACLs can only express "everyone" until SASL is turned on.

To run a listener with real per-client principals, set all of the following (TLS is not optional here — the broker rejects SASL_ENABLED=true without SASL_REQUIRE_TLS=true outside the development profile, because PLAIN puts the password on the wire):

# Authentication
SASL_ENABLED=true \
SASL_REQUIRED=true \
SASL_USERS_FILE=/etc/kafkaesque/sasl/users.json \
SASL_REQUIRE_TLS=true \
# Transport
TLS_ENABLED=true \
TLS_CERT_PATH=/etc/kafkaesque/tls/tls.crt \
TLS_KEY_PATH=/etc/kafkaesque/tls/tls.key \
# Authorization — drop the ANONYMOUS bootstrap super-user once SASL is on
ACL_ENABLED=true \
ACL_DENY_BY_DEFAULT=true \
KAFKAESQUE_SUPER_USERS="User:admin" \
# Control plane
RAFT_CLUSTER_SECRET="$SHARED_SECRET" \
RAFT_JOIN_TOKEN="$JOIN_TOKEN" \
  kafkaesque

SASL_REQUIRED=true rejects unauthenticated connections outright; leave it false only during a migration window while existing clients are cut over.

On Kubernetes the Helm chart exposes this as the sasl and tls value blocks — see values-production.yaml, which turns both on. ACL bindings can be managed over the wire or seeded from a bootstrap file; see Managing ACLs.

Managing ACLs

ACL enforcement and administration are both supported over the wire. Every request path authorizes against the replicated ACL state machine (ACL_ENABLED=true, ACL_DENY_BY_DEFAULT=true), including the Describe/Read/Write/Alter operations on topics, groups, and cluster resources. Standard Kafka admin tooling can manage bindings via DescribeAcls / CreateAcls / DeleteAcls (API keys 29–31, versions 0–1) — kafka-acls.sh and the Java AdminClient speak these APIs.

What is also supported:

Seeding by file. Point KAFKAESQUE_ACL_BOOTSTRAP_FILE at a JSON array of ACL bindings; the leader writes them through Raft on startup so they replicate to every node. Re-applying the same file is idempotent (CreateAcls deduplicates), so this composes with config management.
Super-users bypass ACL checks entirely via KAFKAESQUE_SUPER_USERS (e.g. User:admin). On a cluster without SASL the only available principal is User:ANONYMOUS.

Supported resource types on the wire today are Topic, Group, and Cluster; pattern types are Literal and Prefixed. Unsupported Kafka resource types (TransactionalId, User, DelegationToken) and Match patterns are rejected with INVALID_REQUEST.

Deployment

Multi-broker cluster

For a 3-node cluster, give each broker a unique BROKER_ID and the full RAFT_PEERS list:

BROKER_ID=0 \
RAFT_LISTEN_ADDR=0.0.0.0:9093 \
RAFT_PEERS="0=node0:9093,1=node1:9093,2=node2:9093" \
RAFT_CLUSTER_SECRET="$SHARED_SECRET" \
OBJECT_STORE_TYPE=s3 \
S3_BUCKET=my-kafka-data \
cargo run --release -p kafkaesque-bin --bin kafkaesque

Kubernetes

Raw manifests are in deploy/kubernetes/:

kubectl apply -k deploy/kubernetes/

This deploys a 3-broker StatefulSet. Replace the placeholder secrets in secret.yaml before production use — see deploy/kubernetes/README.md.

Helm

A chart with dev and production value presets is in deploy/helm/kafkaesque/:

helm install kafkaesque ./deploy/helm/kafkaesque -n kafkaesque --create-namespace \
  -f deploy/helm/kafkaesque/values-production.yaml \
  --set raftAuth.clusterSecret="$(openssl rand -base64 32)"

See deploy/helm/kafkaesque/README.md for the full values reference.

Cloud infrastructure (Terraform)

Modules under deploy/terraform/ provision the object store and IAM/identity wiring for AWS, GCP, and Azure (e.g. an S3 bucket plus an IRSA role for EKS). See deploy/terraform/README.md.

Cargo features

Feature	Pulls in	Purpose
`sasl`	(no extra deps)	SASL/PLAIN and SCRAM-SHA-256 authentication.
`tls`	`tokio-rustls`, `rustls`, `rustls-pemfile`	TLS termination on the Kafka listener.
`otel`	`opentelemetry`, `opentelemetry-otlp`, `tracing-opentelemetry`	OTLP span export over gRPC.
`loom`	`loom`	Loom-based concurrency tests (dev only).
`test-utilities`	(no extra deps)	Exposes `MockCoordinator` and other helpers for integration tests.

Supported APIs

Kafkaesque implements the core Kafka protocol used by modern clients:

Produce / Fetch — basic messaging.
Metadata — topic and partition discovery.
OffsetForLeaderEpoch — log-truncation fencing for consumer rebalances (KIP-320).
Consumer groups — JoinGroup, SyncGroup, Heartbeat, LeaveGroup, OffsetCommit, OffsetFetch, DescribeGroups, ListGroups, DeleteGroups, FindCoordinator.
Admin — CreateTopics, DeleteTopics, CreatePartitions, DescribeConfigs, AlterConfigs, IncrementalAlterConfigs, InitProducerId, DescribeAcls, CreateAcls, DeleteAcls.
API negotiation — ApiVersions.
Auth — SaslHandshake, SaslAuthenticate (PLAIN, SCRAM-SHA-256; requires --features sasl).

Exact version ranges are advertised over ApiVersions and locked to the CHANGELOG via tests/changelog_contract_tests.rs. See Client compatibility for what that means for librdkafka, Java, and kcat.

Not yet supported, called out because the wire protocol makes them look available:

Broker-to-broker replication. replication_factor > 1 is rejected; see Durability contract.
Log compaction. cleanup.policy=compact (and compact,delete) is rejected with INVALID_CONFIG — there is no cleaner, so accepting it would promise key-collapsing that never happens. Use cleanup.policy=delete with retention.ms.
Incremental fetch sessions (KIP-227). Fetch v7+ is supported for its other fields (log_start_offset, leader-epoch fencing, rack-aware replica selection). Responses always carry session_id = 0, which is the spec's "sessionless" signal telling clients to keep sending full fetches; a session_id the broker never issued is answered with FETCH_SESSION_ID_NOT_FOUND.
Transactions / EOS. transactional_id is rejected at produce and at InitProducerId. RecordBatch attribute bit 4 (transactional) and bit 5 (control) are refused on the produce path.
Static membership (KIP-345). group.instance.id requires JoinGroup v5+ / Heartbeat v3+ / SyncGroup v3+, which are not advertised.
KIP-848 consumer group protocol. ApiKeys 68/69 are absent; clients that support it fall back to the classic protocol.
Idempotent-producer deduplication beyond the per-session producer_id.

Client compatibility

Kafkaesque speaks the Kafka wire protocol. That is not the same claim as “every Kafka client feature works.” Clients that honor ApiVersions (the default for librdkafka with api.version.request=true, and for modern Java kafka-clients) negotiate min(client_max, broker_max) and do not fail hard when the broker advertises a lower ceiling. Clients that disable version negotiation, or that require APIs we do not advertise (static membership, transactions, KIP-848), will get UnsupportedVersion (35) or a typed produce/admin error — the connection stays open.

Verified in CI

Client	How	Notes
librdkafka via `rdkafka` 0.38	`tests/rdkafka_e2e.rs` (dedicated CI job)	Produce, consume with `group.id`, group offset commit, idempotent produce, admin. Uses ApiVersions negotiation against the advertised ceilings.
kcat	`scripts/run-e2e.sh`, `scripts/run-cluster-e2e.sh`	Produce/consume smoke and multi-broker failover.
Java `kafka-clients`	`java-smoke/` + `scripts/run-java-smoke.sh` (CI `java-client-matrix`)	Produce → consume → OffsetCommit with classic RangeAssignor. Do not enable `group.instance.id`, transactions, or KIP-848.

Group API ceilings (what clients will negotiate)

API	Advertised	Why it stops there
OffsetCommit	v0–v6	v5 drops retention; v6 leader_epoch ignored. `groupInstanceId` is v7+ (refused).
OffsetFetch	v0–v5	v5 response `committed_leader_epoch` (-1). Request body same as v2 (no `require_stable`; that is v7+). Flexible / multi-group v6+ refused.
JoinGroup	v0–v4	v4 MemberIdRequired (KIP-394). v5 static membership refused.
SyncGroup	v0–v3	v3 `group_instance_id` parsed and ignored.
Heartbeat	v0–v3	v3 `group_instance_id` parsed and ignored.
LeaveGroup	v0–v3	v3 batch leave supported; instance ids ignored.
FindCoordinator	v0–v2	v2 error_message. Flexible / multi-key v3+ refused.

A client that forces JoinGroup v5 (or any version above the advertised max) receives UnsupportedVersion without the broker closing the socket. The contract is locked by tests/client_compatibility_matrix_tests.rs.

Produce batch contract

Compressed RecordBatches (gzip / snappy / lz4 / zstd) are accepted when the payload decompresses and the inner record count matches the header records_count. Undefined codec ids, undecompressible payloads, and header/body count mismatches are rejected. Transactional and control attribute bits are rejected even when the CRC is valid.

Testing

# Unit + integration tests (mirrors CI's split):
cargo test --lib
cargo test --tests

# Doc tests:
cargo test --doc

# End-to-end (requires kcat):
./scripts/run-e2e.sh             # single-node
./scripts/run-cluster-e2e.sh     # 3-node, simulates failover
./scripts/run-k8s-e2e.sh         # in-cluster

# All CI checks (fmt, clippy, tests, audit, deny):
make ci

Fuzzing

The fuzz/ directory holds cargo-fuzz targets covering every attacker-reachable parser, decoder, and validator — Kafka request parsers, the connection frame reader, RecordBatch header validation, SCRAM, postcard decoders for cluster RPC, and the identifier validators.

cargo install cargo-fuzz
cargo +nightly fuzz list
cargo +nightly fuzz run api_produce -- -max_total_time=60

See fuzz/README.md for the full target catalogue, reproduction workflow, and corpus / coverage management.

Development

Format: cargo fmt --all
Lint: cargo clippy --all-targets -- -D warnings
Benchmarks: cargo bench (Criterion; results in target/criterion/)
Audit: make audit (RustSec advisories)
Deny: make deny (license / banned-crate policy in deny.toml)

License

Apache-2.0

Name		Name	Last commit message	Last commit date
.cargo		.cargo
.github/workflows		.github/workflows
benches		benches
crates		crates
deploy		deploy
examples		examples
fuzz		fuzz
java-smoke		java-smoke
scripts		scripts
src		src
tests		tests
.dockerignore		.dockerignore
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
Dockerfile.ci		Dockerfile.ci
Dockerfile.minimal		Dockerfile.minimal
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
deny.toml		deny.toml
rust-toolchain.toml		rust-toolchain.toml

Navigation Menu

Repository files navigation

Kafkaesque

Contents

Architecture

Quick Start

Run locally (standalone)

Run with Docker

Smoke test with `kcat`

Workspace layout

Configuration

Health & monitoring

Durability contract

Replication factor and ISR — read this first

`acks`

Graceful shutdown

Property test

Security

Securing the Kafka listener

Managing ACLs

Deployment

Multi-broker cluster

Kubernetes

Helm

Cloud infrastructure (Terraform)

Cargo features

Supported APIs

Client compatibility

Verified in CI

Group API ceilings (what clients will negotiate)

Produce batch contract

Testing

Fuzzing

Development

License

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

Kafkaesque

Contents

Architecture

Quick Start

Run locally (standalone)

Run with Docker

Smoke test with kcat

Workspace layout

Configuration

Health & monitoring

Durability contract

Replication factor and ISR — read this first

acks

Graceful shutdown

Property test

Security

Securing the Kafka listener

Managing ACLs

Deployment

Multi-broker cluster

Kubernetes

Helm

Cloud infrastructure (Terraform)

Cargo features

Supported APIs

Client compatibility

Verified in CI

Group API ceilings (what clients will negotiate)

Produce batch contract

Testing

Fuzzing

Development

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Smoke test with `kcat`

`acks`