The expedition brief

A Rust-native Apache Kafka client, built from the wire protocol up, that is outcome-faithful to the Java client without a JVM.

Most client libraries are documented like museums: here is the API, here are the exhibits, please don’t touch the glass. This book is written as the opposite — a travelogue. kacrab was built by walking the Kafka protocol from the first ApiVersions byte to a full producer, admin client, and consumer, and nearly everything interesting about it was discovered on the way: a handshake round most clients skip, a missing metadata invalidation that wedged an entire burst, a localhost that resolved to a dead IPv6 loopback, a coverage tool that made correct code flake.

So each chapter is a leg of that journey. It tells you what we set out to build, what the terrain actually looked like, what broke, what the broker taught us — and which configuration lessons we packed for the next leg. The API docs on docs.rs tell you what each type does; this book tells you how it works, why it is built that way, and how to configure it well.

What we brought home

• A Java-faithful client that is not a Java port. The idempotent producer reproduces the Java client’s real algorithms — inflightBatchesBySequence ordering, firstInFlightSequence retry gating, maybeResolveSequences epoch handling — and is byte-exact where bytes are the contract (murmur2, CRC32C, varint/zigzag, record-batch v2). Where the runtime model differs (async tasks vs Java’s single Sender thread), kacrab keeps the outcomes identical. See Design decisions.
• A consumer at full Java feature parity. Manual assignment, topic and pattern subscription, both group protocols (classic assignors including incremental cooperative-sticky, and the KIP-848 server-side protocol), incremental fetch sessions (KIP-227), topic-id fetches (KIP-516), and truncation detection (KIP-320). See The consumer client.
• No JVM tax — measured, not asserted. At identical default configs against the same native broker, the producer is ~25–28% faster than Java’s own perf tool at ~4× less memory; the consumer reads 1.9–4× faster at ~16–20× less memory. See Benchmarks.
• A generated, oracle-checked protocol. Request/response types are generated from the upstream Kafka schemas and cross-checked against the Java client as an external oracle. See Protocol codegen.
• Verified against real brokers, not just against itself. Every SASL mechanism, every TLS mode, every compression codec, multi-broker failover, and every admin operation ran end-to-end against real Apache Kafka 4.3.0. See Verification.

The route

The parts of this book follow the order the client was actually built — each leg stands on the previous one.

flowchart LR
  A["I. Setting out<br/>wire + protocol"] --> B["II. The producer's road<br/>pipeline, idempotence"]
  B --> C["III. Hostile territory<br/>auth, failures"]
  C --> D["IV. Side expedition<br/>admin client"]
  D --> E["V. The road back<br/>consumer"]
  E --> F["VI. Proof<br/>tests, brokers, benches"]
  F --> G["VII. Field guide<br/>config best practices"]

• Part I — Setting out. Before anything can be produced or consumed, bytes must reach a broker and come back: the wire layer and the generated protocol it speaks.
• Part II — The producer’s road. One record, followed from send() to a broker acknowledgement: the pipeline, partitioning, the idempotent state machine — the client’s correctness core — and compression.
• Part III — Hostile territory. The parts of the map marked “here be dragons”: SASL & TLS handshakes (including the rounds most clients get subtly wrong) and the failure catalogue.
• Part IV — A side expedition. The admin client: 62 operations and the four routing patterns that carry them.
• Part V — The road back. Everything we wrote, read back in order: the consumer, group rebalancing, and fetching & offsets.
• Part VI — Proof. How we know any of the above is true: testing & CI, real-broker verification, and benchmarks.
• Part VII — The field guide. The distilled configuration best practices — foundations, producer tuning, and consumer tuning. If you came here to run kacrab rather than to study it, you can start there and follow the links back into the deep dives.

How to read this book

Read Parts I–VI in order if you want the journey; jump straight to Part VII if you want the destination. The two crown-jewel deep dives are Idempotency & transactions and Security — both are self-contained.

Status

kacrab 0.1.0 is published on crates.io (producer, consumer, and admin surfaces). Pre-1.0, the public API can still change between minor versions; this book tracks master. Kafka Streams and share groups are separate products and not part of this client.

The map

Every expedition starts with a map. kacrab’s territory is small on purpose: three client surfaces sharing one transport, one config system, and one generated protocol — each layer a crate or module with a sharp boundary.

flowchart TB
  subgraph app[Your application]
    P["Producer<br/>send / flush / transactions"]
    K["Consumer<br/>subscribe / poll / commit"]
    AD["AdminClient<br/>62 operations"]
  end
  subgraph kacrab[kacrab crate]
    CFG["config<br/>Java-style ClientConfig,<br/>typed configs, validation"]
    COM["common<br/>TopicPartition, Node,<br/>OffsetAndMetadata, …"]
    WIRE["wire<br/>broker sessions, request pipeline,<br/>SASL/TLS, metadata, reconnect"]
  end
  subgraph proto[kacrab-protocol crate]
    GEN["generated request/response types,<br/>record batch v2, compression, CRC32C"]
  end
  P --> WIRE
  K --> WIRE
  AD --> WIRE
  WIRE --> GEN
  P -.uses.-> CFG
  K -.uses.-> CFG
  AD -.uses.-> CFG
  P -.shares.-> COM
  K -.shares.-> COM
  AD -.shares.-> COM

The regions

• config — the Kafka property surface (bootstrap.servers, acks, group.id, compression.type, sasl.*, ssl.*, …) with typed parsing and validation, generated from official Kafka config metadata and drift-checked against it. The field guide is the practical tour of this surface.
• wire — the async transport every client rides: per-broker Tokio sessions, a bounded in-flight request pipeline, the SASL/TLS handshakes, metadata fetch + leader-change invalidation, and reconnect/backoff. See First contact and Security.
• common — shared org.apache.kafka.common domain types (TopicPartition, OffsetAndMetadata, Node, …), always compiled and re-exported by every surface, so a TopicPartition is the same type whether a producer, consumer, or admin call hands it to you.
• producer — batching accumulator, the dispatch path that groups batches per broker leader, and the idempotent/transactional state machine. See Following one record and Exactly once.
• consumer — manual assignment and group subscription over both group protocols, the fetch/position/offset state machines, and incremental fetch sessions. See The consumer client.
• admin — the full Apache Kafka 4.3.0 Admin operation surface with controller/coordinator/per-leader/broadcast routing. See The admin client.
• kacrab-protocol — the generated wire types, record-batch v2 encode/decode, compression codecs, and CRC32C. Zero hand-written byte patching. See Learning the language.

The parts left blank

Kafka Streams and share groups (queues) are deliberately outside this map — they are separate products built on clients, not client surfaces. Everything a Kafka 4.3 client does — produce, consume, administer, authenticate — is implemented.

Choosing your provisions

Every surface is a Cargo feature and default = [], so you enable exactly what you use:

[dependencies]
kacrab = { version = "0.1", features = ["producer", "consumer", "admin"] }

consumer pulls in the compression meta-feature (fetched batches must be decodable regardless of what the producer chose). zstd and lz4-hc compile C code; a pure-Rust build is possible — the trade-offs are in Traveling light and the field guide.

First contact: the wire layer

The journey starts here, because everything else has to. Before a producer can batch or a consumer can rebalance, one problem must be solved completely: turn a stream of “send this request to broker N” commands into bytes on a socket and responses back to the caller — correctly, with bounded memory, surviving reconnects and broker failovers. That is the wire module, and every higher layer rides on it. It is built on Tokio and rustls — the Kafka client logic is pure Rust, not a librdkafka wrapper. (The TLS crypto provider under rustls is aws-lc-rs, which is C/assembly; see Design decisions for which dependencies are C.)

One task per broker

Each broker connection is owned by its own Tokio task (BrokerTask), fed by an mpsc command queue. The control plane holds a cheap BrokerHandle (a Clone sender) and never touches the socket directly.

flowchart LR
  CP["control plane<br/>(producer dispatch)"] -- RequestCommand --> Q[(mpsc queue)]
  Q --> T["BrokerTask<br/>connect → negotiate → serve"]
  T -- write --> S[("TCP / TLS socket")]
  S -- frames --> R["reader task"]
  R -- completes --> CP

On connect the task runs a synchronous capability exchange — ApiVersions (v3, with client software name/version so the broker logs kacrab’s identity) and, if security.protocol requires it, the SASL/TLS handshake — before the request pipeline opens. The negotiated BrokerCapabilities are cached and shared so the control plane can gate behavior on what the broker supports (for example, the coordinator’s InitProducerId version).

The request pipeline: fixed-slot, no per-request map

Up to max.in.flight.requests.per.connection requests are on the wire at once. Correlating responses to waiters is the hot path, so kacrab does not use a per-request hashmap — it uses a fixed-slot ring (RequestPipeline holds slots: Vec<Option<InFlightRequest>>). A request takes the next slot, its correlation id encodes the slot, and the response lands back in O(1) with no allocation or hashing per request.

Back-pressure

When every slot is full, new commands wait (or are rejected with Backpressure) rather than growing an unbounded queue.

Reader / writer split

The socket is split: a writer half on the task, a reader half on a dedicated reader task.

• Writer — request frames are coalesced through a BufWriter, so a burst of small requests becomes few write syscalls instead of one per request.
• Reader — the reader task parses length-delimited response frames and completes the waiting request directly, instead of hopping the frame back to the main task first. One fewer hand-off on every response.

Metadata & leadership

The wire layer fetches cluster/topic metadata, caches it, and invalidates it on a leader change so the next request re-routes. Two triggers matter:

• A broker response carrying a NotLeader error or a current-leader update → apply the new leader.
• A connection drop (the broker went away) → invalidate the affected partitions so the retry re-fetches fresh leaders. This second case is the one that, when missing, caused the burst-wedge bug.

A configurable metadata-recovery strategy (rebootstrap) handles the case where all known brokers become unreachable.

Reconnect & backoff

When a connection fails to establish, the broker loop backs off and retries — exponential backoff with jitter, reset to the initial delay on a successful connection — while honoring each pending command’s request timeout. But not every failure should be retried:

The fatal cases mirror Java’s non-retriable SaslAuthenticationException / SslAuthenticationException: a wrong password or an untrusted certificate fails immediately with the broker’s reason, instead of looping until request.timeout.ms. See Security and Failure modes.

DNS is part of the transport

One discovery on this leg came from a hang, not a crash: a broker advertised as localhost resolved to the IPv6 loopback first — where nothing listened — and a pinned coordinator connection waited forever. DNS resolution is now centralized in the wire layer: hostnames are re-resolved on every connect, all returned addresses are tried IPv4-first, and client.dns.lookup=use_all_dns_ips is honoured. Producer, admin, and consumer all inherit the fix because they all ride this layer.

Field notes

• The backoff pair (reconnect.backoff.ms → reconnect.backoff.max.ms) is exponential and jittered out of the box — see the foundations field guide before overriding.
• An auth failure at connect is fatal by design. If startup fails fast with a SASL/TLS reason, that is the wire layer doing its job — fix the credential, don’t raise request.timeout.ms.
• In containerized or multi-homed environments, prefer client.dns.lookup=use_all_dns_ips and advertise brokers by names every client network can actually reach.

Learning the language: protocol codegen

You cannot explore a territory whose language you don’t speak — and Kafka’s language is hundreds of request/response types across dozens of versions, with flexible/compact encodings, nullable fields, tagged fields, and nested schemas. Hand-writing and hand-maintaining that is how subtle wire bugs are born. So before setting out, kacrab generates the entire protocol from the upstream schemas — and, not trusting its own pronunciation, checks the result against the Java client as an external oracle.

kacrab-codegen

A maintainer-only tool (not published to crates.io — no runtime crate depends on it) with two subcommands:

• protocol — parse the Apache Kafka 4.3.0 message schemas and emit the Rust request/response structs (and their encode/decode) into kacrab-protocol, plus the generated test fixtures.
• config — extract upstream ConfigDef declarations into the typed config metadata that backs ClientConfig and the producer/consumer/admin configs.

flowchart LR
  S["Kafka message schemas<br/>(apache/kafka@4.3.0)"] --> P["parser"]
  P --> C["codegen"]
  C --> F["rustfmt / prettyplease"]
  F --> O["kacrab-protocol::generated"]
  P --> EJ["errors_java"]
  C --> TU["test fixtures<br/>(6 families)"]

The pipeline handles the things that make Kafka’s protocol fiddly: per-version field presence, compact vs non-compact (flexible) versions, tagged fields, and nested schema traversal.

The Java oracle matrix

This is the part that makes the generated code trustworthy. Generated fixtures are encoded by Rust and decoded by the real Kafka Java client, and vice-versa, across six fixture families — 625 cases each:

Passing the matrix means: Rust encoders produce bytes Kafka Java can decode for every represented schema version, Rust decoders consume Java-produced bytes, and a decode/re-encode preserves the exact byte sequence.

Why an oracle, not just round trips

A Rust-only round trip (encode then decode in Rust) passes even if Rust consistently writes the wrong wire shape and then reads its own wrong shape back. The Java client is treated as the external source of truth for Kafka’s wire contract — the same philosophy as the real-broker verification, one layer down.

What it does not prove

The matrix is not exhaustive over every value combination, and it does not cover broker/client behavior outside message serialization (that is what the unit tests, the idempotent fixtures, and the real-broker integration tests are for). It proves cross-language wire compatibility for the generated schema surface.

The config surface is generated too

The config subcommand extracts upstream ConfigDef declarations into the catalog behind ClientConfig — which is why every key in the field guide carries Kafka’s own name, type, default, and validation. The hand-curated typed API is cross-checked against that catalog by a drift test, so a config documented in this book is a config that exists, with the semantics upstream gave it.

Following one record

The best way to understand a producer is to walk beside a single record: from the moment your code calls producer.send(record) to the moment a broker acknowledgement resolves its future. This chapter is that walk. The idempotency chapter covers the sequencing that keeps the walk honest; this one covers the path itself.

flowchart LR
  A["send(record)"] --> B["partition assignment"]
  B --> C["accumulator<br/>(per-topic-partition batches)"]
  C -- drain ready --> D["prepare<br/>(idempotent sequences)"]
  D --> E["dispatch:<br/>group by broker leader"]
  E --> F["split on max.request.size"]
  F --> G["EnqueueSequencer<br/>(ordered enqueue)"]
  G --> H["wire → broker"]
  H -- ack --> I["RecordMetadata receipt"]

send is synchronous

kacrab matches the Java Producer.send shape: send is a plain fn, not an async fn. It assigns the partition, appends to the accumulator, and returns a SendFuture you await for the broker acknowledgement.

#![allow(unused)]
fn main() {
let delivery = producer.send(ProducerRecord::new("orders", 0).value(v))?; // sync
let receipt  = delivery.await?;                                           // ack
}

This is deliberate: the append is a cheap, non-blocking, in-memory operation (until buffer.memory is exhausted, at which point it blocks up to max.block.ms), exactly like Java. Deferring the dispatch lets the accumulator coalesce many sends into a few Produce requests.

The accumulator

Records are batched per topic-partition. AccumulatorConfig controls the knobs:

• batch.size — the soft cap on bytes per partition batch.
• linger.ms — how long a partial batch waits for more records before it is eligible to drain (zero by default).
• buffer.memory + max.block.ms — the total memory the unsent records may occupy, and how long send blocks when that memory is full.

A background sender drains ready batches (full, or past their linger deadline), prepares them, and dispatches.

Drain → prepare → dispatch

1. Drain pulls ready batches out of the accumulator in a single sequential pass — the point where idempotent sequence numbers are assigned in order. The idempotent path drains only each partition’s front batch per cycle: dispatch starts at most one new request per partition per selection, so draining a deep backlog only to re-enqueue all but one batch per partition would be O(backlog) churn under the accumulator lock on every cycle. Non-idempotent dispatch drains everything ready (it coalesces all of it into one request).
2. Prepare stamps each batch with its (producer id, epoch, base sequence) and registers it in the partition’s in-flight set. Batches that cannot be sent in order yet are deferred (re-enqueued). A cycle whose every ready batch is deferred (all partitions already at the in-flight depth cap) parks the sender until a completion frees a slot — re-polling immediately would livelock in a hot drain/defer/requeue spin for the whole round trip.
3. Dispatch routes each batch to its partition leader, groups batches destined for the same broker into one Produce request, splits on max.request.size, and enqueues through the EnqueueSequencer so the broker sees ascending base sequences per partition. Responses are awaited concurrently.

Why group by broker

A partition has one leader; many partitions can share a leader. Grouping by leader turns N partition batches into one request per broker — fewer round trips, and the per-broker coalescing already saturates a low-RTT connection.

Delivery, flush, close

• Each batch yields a RecordMetadata (topic, partition, offset, timestamp, serialized sizes) delivered through the SendFuture or a send_with_callback callback.
• flush waits for all buffered + in-flight records to complete.
• close flushes then shuts the producer down (close_now for an immediate drop; close_timeout bounds the wait).

Interceptors & metrics

The producer runs the full Kafka ProducerInterceptor lifecycle (configure / on_send / on_acknowledgement / close, plus the ClusterResourceListener.onUpdate hook) and publishes metrics under their Kafka names (producer-metrics:* / producer-topic-metrics:*), including the buffer-pool gauges.

Field notes

Every stage of this walk has a tuning knob, and the producer field guide ranks them. The short version:

• linger.ms trades a bounded wait at the accumulator for fuller batches — the highest-leverage throughput lever on the whole path.
• buffer.memory + max.block.ms are the back-pressure contract: when the accumulator is full, send blocks, it doesn’t grow without bound.
• flush() waits for the entire pipeline — call it at commit points, never per record.

Choosing a partition

The first fork in every record’s road: which partition? It looks like a detail and is actually a contract. kacrab chooses exactly as the Java client does, because a different choice means a different key→partition mapping — which silently breaks ordering guarantees and consumer expectations for anyone running both clients against the same topic. This was one of the first places the journey demanded byte-exactness, not just compatibility.

Keyed records: murmur2, byte-exact

When a record has a key, the partition is

(murmur2(key_bytes) & 0x7fffffff) % num_partitions

kacrab’s murmur2 is byte-exact with Kafka’s for every key length — the same seed, the same tail handling, the same masking — verified directly against the Java implementation. Same key → same partition → same per-key order, whether the producer is kacrab or Java.

Null-key records: sticky / adaptive

With no key, partitioning by round-robin per record would scatter a batch across every partition and defeat batching. Kafka’s BuiltInPartitioner instead is sticky: it picks one partition and keeps using it until the current batch fills, then rotates. kacrab mirrors this, and the adaptive variant weights each rotation by the partitions’ live accumulator queue depths (a shorter queue gets more traffic; a slow/unavailable leader gets less), matching Kafka’s partitioner.adaptive.partitioning.enable and partitioner.availability.timeout.ms.

The queue depths are re-sampled from the accumulator at every sticky rotation — on the synchronous send path as well as the awaiting one, mirroring Java’s sender refreshing partitionLoadStats on every RecordAccumulator.ready() pass. This freshness is load-bearing, not a nicety: rotating on a frozen snapshot locks in whatever weighting the last refresh saw, the favored partition keeps absorbing most new batches, its queue serializes dispatch, and the imbalance never self-corrects (observed as a 62/31/8 split across three partitions that halved large-record throughput).

flowchart TD
  R["record"] --> K{key?}
  K -- yes --> M["murmur2(key) % partitions"]
  K -- no --> S{explicit<br/>partition?}
  S -- yes --> P["use it"]
  S -- no --> A["sticky/adaptive<br/>(stick until batch full, then rotate)"]

Explicit partition

ProducerRecord::new(topic, partition) targets a partition directly and bypasses the partitioner entirely — used by the multi-broker tests to drive specific leaders.

Custom partitioners

The ProducerPartitioner trait is the extension point:

#![allow(unused)]
fn main() {
pub trait ProducerPartitioner: Send + Sync + 'static {
    fn partition(&self, record: &ProducerRecord, metadata: &ClusterMetadata) -> Result<i32>;
}
}

RoundRobinPartitioner ships as a built-in (mirroring Kafka’s RoundRobinPartitioner), and you can supply your own for custom routing.

The point of byte-exactness

Partitioning is one of the handful of things that must be byte-for-byte identical to Java, not just “compatible”. kacrab treats murmur2, CRC32C, and the varint/zigzag encodings as that kind of contract — tested against the Java output, not just against themselves.

Field notes

• A key is an ordering contract — choose it for the order you need, and watch cardinality: a hot key is a hot partition no client can fix.
• Null-keyed traffic is well served by the default sticky/adaptive path; forcing round-robin scatters batches and costs throughput.
• The adaptive re-sampling story above is why “it worked in the old version” is not a tuning argument — measure your own partition spread (producer-topic-metrics) before overriding partitioner.adaptive.partitioning.enable.

More in the producer field guide.

Exactly once, even when the network lies

This is the hardest leg of the producer’s road, and the one the rest of the journey was planned around. The promise of an idempotent producer is simple to state and brutal to keep: every record is written to the log exactly once and in order, even across retries, reconnects, and broker failovers — even when the network swallows a response and refuses to say whether the write landed. Kafka delivers this with a (producer id, epoch, base sequence) triple stamped on every record batch, and a per-partition state machine on both the client and the broker. kacrab reproduces that state machine — the real one, from the Java client’s source, not a simplified sketch.

The invariant we defend

For each partition, the broker accepts batches whose base sequence is exactly the next expected sequence. A gap → OUT_OF_ORDER_SEQUENCE_NUMBER. A repeat → silently deduplicated (DUPLICATE_SEQUENCE_NUMBER, treated as success). So the client must guarantee that the sequence numbers it puts on the wire, per partition, are a strictly ascending, gap-free run — even when multiple requests are in flight and some of them are retried.

The pieces of state

Per partition the producer tracks, mirroring Java’s TransactionManager:

• nextSequence — the sequence to stamp on the next new batch.
• inflightBatchesBySequence — batches sent but not yet acknowledged, ordered by base sequence. This is what lets us keep several requests in flight per partition without losing the order on retry.
• firstInFlightSequence — the base sequence of the oldest unacked batch; the retry gate.
• producer (id, epoch) — bumped to fence stale state after an ambiguous failure.

Multiple in flight, still in order

With max.in.flight.requests.per.connection = 5, five ProduceRequests for the same partition can be on the wire at once. The Java client guarantees their enqueue order with a single Sender thread. kacrab has no Sender thread — it dispatches on concurrent Tokio tasks — so it reconstructs that ordering with an explicit ticket sequencer.

sequenceDiagram
    participant App
    participant Acc as Accumulator
    participant Seq as EnqueueSequencer
    participant Br as Broker task
    App->>Acc: send(r1), send(r2), send(r3)
    Note over Acc: drain → batches B0(seq0), B1(seq1), B2(seq2)
    Acc->>Seq: reserve_ticket() → t0, t1, t2 (drain order)
    par dispatch tasks (concurrent)
        Seq-->>Br: t0 wait_turn → enqueue B0 → advance_past(t0)
        Seq-->>Br: t1 wait_turn (after t0) → enqueue B1 → advance_past(t1)
        Seq-->>Br: t2 wait_turn (after t1) → enqueue B2 → advance_past(t2)
    end
    Note over Seq,Br: turn is released right after enqueue,<br/>so response waits overlap the next turn

The key subtlety: a dispatch advances the turn as soon as its request is enqueued on the socket, not after the response comes back. So the broker observes ascending base sequences per partition (the ordering guarantee), while the response waits still run concurrently (the throughput we want). A retry that reuses its ticket returns from wait_turn immediately, so retries never deadlock behind newer tickets.

Retry without reordering

When a batch fails retriably (e.g. NOT_LEADER_FOR_PARTITION), it must be re-sent before any batch with a higher sequence — otherwise the broker sees a gap. The firstInFlightSequence gate enforces this: a partition only re-sends starting from its oldest unacked sequence, and newer in-flight batches for that partition wait behind it.

stateDiagram-v2
    [*] --> Idle
    Idle --> InFlight: stamp (id,epoch,seq), enqueue
    InFlight --> Done: ack (seq == expected)
    InFlight --> InFlight: DUPLICATE_SEQUENCE (already written → success)
    InFlight --> Retry: retriable error (NotLeader, throttle, ...)
    Retry --> InFlight: re-send from firstInFlightSequence (no backoff if leader changed)
    InFlight --> Bump: ambiguous loss (no broker response)
    Bump --> InFlight: epoch++, re-stamp stale batches, restart sequences
    Done --> [*]

When to fence: the epoch bump

The dangerous case is an ambiguous loss — the connection drops with no broker response, so the client cannot know whether the batch was written. Replaying it blindly could duplicate; not replaying could drop. Kafka resolves this by bumping the producer epoch, which fences all old in-flight state, and restarting sequences for the partition.

kacrab distinguishes the two failure shapes precisely:

maybeResolveSequences defers the bump until the partition has no in-flight batches (the hasInflightBatches gate) so a bump can’t race a sibling request.

The concurrent-task adaptation

Java renumbers a partition’s in-flight batches in place because its single Sender thread owns them all. kacrab’s dispatch tasks each own their own batches and can’t reach into a sibling’s, so it can’t renumber them in place. Instead a bump performs a global epoch reset: the epoch advances, every partition’s sequences restart, and any batch stamped under the now-stale epoch is re-stamped on its next prepare (it is detected as stale and re-enqueued). The on-the-wire outcome is identical to Java’s startSequencesAtBeginning — the broker sees a fresh epoch with sequences from zero — but the mechanism fits the concurrent model. See the audit notes in Design decisions.

A failure that looked fixed but wasn’t

A real bug found by verifying against a 3-broker cluster: when a broker died mid-burst, a wire/connection error retry did not invalidate the stale leader metadata. The retry kept routing the affected partitions to the dead broker until the delivery timeout — and because a dispatch retries its whole batch group as one unit, the healthy partitions batched alongside them wedged too (every delivery timed out, not just the affected ones).

The fix: a wire-failure retry now invalidates the affected partitions before retrying, so the next attempt re-fetches metadata and re-routes to the new leaders — mirroring Java’s NetworkClient requesting a metadata update on server disconnect. A 6-partition burst across a broker loss went from 0/6 (60 s timeout) to 6/6. See Failure modes.

Transactions

Transactions layer onto the idempotent core: a transactional.id gets a fenced producer id via InitProducerId, partitions are registered with AddPartitionsToTxn before their first write, consumer offsets commit through TxnOffsetCommit, and the transaction ends with EndTxn (commit or abort) routed through the transaction coordinator (found via FindCoordinator). The control flow is a state machine over those requests; the data-plane sequencing is exactly the idempotent machine above, with the transactional id and epoch added to each batch.

Tip

The whole transactional path was smoke-tested against a real broker (InitProducerId → begin → send → commit → delivery receipt). See the real_kafka_producer test in Verification.

Field notes

Everything in this chapter runs at the default config — that is the point:

• enable.idempotence=true and acks=all are the defaults; leave them. Turning idempotence off to “save overhead” forfeits both dedup and ordering under retry, for a few bytes per batch.
• max.in.flight.requests.per.connection ≤ 5 is the ceiling under which the sequencing above holds; 5 is the default and the sweet spot.
• Bound retries with delivery.timeout.ms, not retries — the state machine makes retries safe, so let time be the budget.
• transactional.id: stable per logical producer, unique per instance — sharing one across two live producers triggers fencing by design.

The producer field guide puts these in context.

Traveling light: compression

A caravan moves faster carrying less, and so does a Kafka pipeline — compressed batches multiply effective network and broker disk throughput. Compression happens at the record-batch level: a batch of records is serialized, the record block is compressed, and the compressed bytes go inside a record-batch v2 envelope with a CRC over them. Every Kafka client and broker must frame this identically — get it almost right and your own process reads the bytes back fine while the broker rejects them — which is why kacrab’s codecs are proven on a real broker, not just round-tripped in process.

Codecs

Each codec is a feature, so you only pull in the backends you use:

The compression meta-feature turns on gzip + snappy + lz4 + zstd. The first three are pure Rust; zstd uses the C libzstd (via zstd-sys), so enabling compression — or zstd on its own — needs a C compiler at build time. lz4-hc is a separate opt-in for the same reason (C-FFI to liblz4); when both lz4 and lz4-hc are enabled, the C-FFI backend wins.

For a fully pure-Rust, no-C-toolchain build (e.g. easy cross-compilation), enable only gzip + snappy + lz4 and skip zstd / lz4-hc / the compression meta-feature.

Selecting a codec

compression.type is the wire codec — gzip / snappy / lz4 / zstd / none:

#![allow(unused)]
fn main() {
Producer::builder()
    .set("compression.type", "zstd")
    .set("compression.zstd.level", "6")   // optional
    .build().await?;
}

There is no lz4-hc wire type — high-compression is an encoder choice that still emits a standard LZ4 frame, so any broker/consumer reads it back normally.

Levels

compression.{gzip,lz4,zstd}.level set the codec level, matching the Java client’s keys.

lz4 level needs lz4-hc

With the pure-Rust lz4 backend the level is ignored (fast mode only). To honor compression.lz4.level (3..=12) you must enable the lz4-hc feature, which routes those levels through liblz4’s high-compression mode.

Framing — why it must be exact

Inside record-batch v2 the layout is roughly: [batch header][CRC32C][compressed record block], and the CRC is computed over the compressed bytes. Get the codec framing subtly wrong — Snappy’s block format, LZ4’s frame descriptor and content checksum, the trailing markers — and the bytes still decompress in your own process (you wrote them) but the broker rejects them or a real consumer can’t read them.

That gap is exactly why the verification test does three things end to end: produce each codec, confirm with kafka-dump-log that the on-disk batch is stored with the right codec (a silently-uncompressed send fails here), and consume the payloads back through kafka-console-consumer.

Field notes

• Default recommendation: compression.type=zstd for ratio, lz4 when CPU is the scarce resource. Both are covered in the producer field guide.
• compression.lz4.level silently does nothing without the lz4-hc feature — the pure-Rust backend is fast-mode only.
• Consumers need no codec config: the batch header names its codec, and the consumer feature already enables all four backends.
• Building without a C toolchain (cross-compilation, minimal images): take gzip + snappy + lz4 only, and skip zstd/lz4-hc/the compression meta-feature.

Proving who we are: SASL & TLS

Past the friendly single-broker campsite, the road runs through checkpoints: production clusters demand you prove who you are before they say a word. This leg of the journey held the most traps of any — two authentication mechanisms hide an extra protocol round that a client can skip and appear to work, right up until the first real request. kacrab speaks every mechanism the Java client does, over plain TCP or TLS, and every path here is verified end-to-end against real brokers — including a real Kerberos KDC.

security.protocol selects the combination:

The SASL handshake

Every SASL mechanism rides the same two-request envelope, negotiated after an ApiVersions exchange:

sequenceDiagram
    participant C as kacrab
    participant B as Broker
    C->>B: ApiVersions
    B-->>C: supported versions
    C->>B: SaslHandshake(mechanism)
    B-->>C: enabled mechanisms (must contain ours)
    loop mechanism-specific
        C->>B: SaslAuthenticate(client token)
        B-->>C: SaslAuthenticate(server token / error)
    end
    Note over C,B: authenticated channel

What varies is the token exchange inside the loop.

• PLAIN — one round: \0username\0password.
• SCRAM-SHA-256 / -512 — two rounds: client-first → server-first (salt+iterations) → client-final (proof) → server-final (verifier). kacrab verifies the server signature, so a man-in-the-middle that can’t prove the shared secret is rejected.
• OAUTHBEARER — token-based, with an error subtlety (below).
• GSSAPI — Kerberos, with a security-layer round most home-grown clients miss (below).

Fail fast, don’t retry an auth failure

A rejected credential is not a transient error. Early on, kacrab classified a SASL/TLS failure as retryable, so a wrong password looped under reconnect backoff until request.timeout.ms and surfaced as “request timed out” 30 seconds later. Now SaslAuthentication, SaslHandshake, TlsHandshake, and a failed server-signature check are fatal setup errors — they fail fast with the broker’s real reason (“Invalid username or password”, “invalid peer certificate: UnknownIssuer”), matching Java’s non-retriable SaslAuthenticationException / SslAuthenticationException.

OAUTHBEARER: the error round you can’t skip

On a valid token the broker returns an empty server response and auth is done. On a rejected token (e.g. expired), RFC 7628 says the server returns an error challenge (a JSON blob like {"status":"invalid_token"}) and waits for the client to send a single 0x01 “kill” byte before failing.

A naive client does one round, sees no Kafka-level error code, assumes success, and sends its first application request — at which point the broker, still mid-authentication, rejects it with ILLEGAL_SASL_STATE. kacrab completes the RFC 7628 error round instead:

sequenceDiagram
    participant C as kacrab
    participant B as Broker
    C->>B: SaslAuthenticate(bearer token)
    alt token valid
        B-->>C: empty response → done
    else token rejected
        B-->>C: error JSON (non-empty)
        C->>B: SaslAuthenticate(0x01 kill byte)
        Note over C: fail fast with the broker's error
    end

GSSAPI: the security-layer round most clients miss

Kerberos is the trickiest. After gss_init_sec_context establishes the security context, the exchange is not over — RFC 4752 requires one more round: the server sends a GSS-wrapped offer (a bitmask of supported security layers + a max buffer size), and the client must unwrap it, choose a layer, and send back a wrapped reply. Skip it and the broker stays mid-authentication and rejects the next request with ILLEGAL_SASL_STATE — exactly the OAUTHBEARER trap, in a different costume.

sequenceDiagram
    participant C as kacrab (libgssapi)
    participant B as Broker (Kerberos)
    rect rgb(235,244,255)
    Note over C,B: 1. context establishment
    C->>B: AP-REQ (initial token)
    B-->>C: AP-REP
    Note over C: gss context complete
    end
    rect rgb(235,255,240)
    Note over C,B: 2. RFC 4752 security-layer round
    C->>B: empty token (yield turn)
    B-->>C: GSS-wrapped offer (layers + max buffer)
    C->>B: GSS-wrapped reply (no-security-layer, max buffer 0)
    Note over C,B: authenticated
    end

kacrab models this as an explicit Establishing → SecurityLayer → Done state machine over libgssapi, selecting “no security layer” (Kafka runs its own transport), exactly as the JDK’s GssKrb5Client does.

TLS

The transport side is rustls (with aws-lc-rs). Trust and identity material loads from PEM (inline or file), JKS, or PKCS12 — the same ssl.truststore.* / ssl.keystore.* keys as the Java client.

• Server auth verifies the broker certificate against the configured trust anchors (plus OS roots).
• ssl.endpoint.identification.algorithm — https (default) checks the certificate SAN against the broker hostname; empty skips the hostname check but still verifies the chain.
• Mutual TLS — supply a client certificate/key (ssl.keystore.*) and the broker authenticates the client too.

What “verified” means here

Not “the unit tests pass” — every mechanism above was run against real Apache Kafka 4.3.0: PLAIN/SCRAM/OAUTHBEARER over plaintext and TLS, GSSAPI against a real MIT Kerberos KDC, and one-way + mutual TLS, each authenticating and running ApiVersions, with negative cases (bad credentials, expired token, untrusted cert) confirming rejection is fast and clear. See Verification.

Field notes

• Production baseline: SASL_SSL + SCRAM-SHA-512 (or OAUTHBEARER against an identity provider). PLAIN sends the password to the broker — only ever inside TLS.
• Keep ssl.endpoint.identification.algorithm=https (the default). Blanking it disables the hostname check and invites exactly the man-in-the-middle that SCRAM’s server-signature verification exists to catch.
• Auth failures are fatal and fast by design. A wrong password surfaces as the broker’s real reason at connect — if you see a 30-second timeout instead, you are running a client with the bug this chapter opened with.
• JVM sasl.jaas.config class names cannot cross into a Rust process; custom flows use the native sasl_client_authenticator hook — see the foundations field guide.

When brokers die: failure modes

A client that works on a healthy cluster is a demo. A client you can run is one that does the right thing when a broker dies mid-flight, a credential is wrong, a leader moves under load, or a request is lost with no answer. This chapter is the storm log of the journey — the catalogue of what went wrong (in testing, on purpose, and once or twice by surprise) and what kacrab does about each one now.

Retryable vs fatal

The first decision on any error is whether retrying could possibly help. Getting this wrong in either direction is a bug: retrying a fatal error wastes the whole request.timeout.ms budget; failing fast on a transient one drops a recoverable request.

The fatal SASL/TLS rows are the ones that, before they were fixed, looped under reconnect backoff until request.timeout.ms and surfaced as “request timed out” — see Security.

Ambiguous loss → epoch bump

The subtle one. A connection drops with no broker response: the records may have been written. Blindly replaying risks a duplicate; not replaying risks a drop. Kafka resolves it by bumping the producer epoch, fencing the old in-flight state, and restarting sequences. kacrab bumps only on ambiguous losses (no response) — a definitive rejection like NotLeader re-routes and retries the same sequence, no bump. The full mechanism is in Idempotency & transactions.

Leadership change under load — the burst wedge

A real bug, found by verifying against a 3-broker cluster: a broker died while a burst of records to several partitions was in flight.

flowchart TD
  K["broker dies"] --> W["wire error on its partitions"]
  W --> X{retry handler<br/>invalidates metadata?}
  X -- "no (the bug)" --> Y["re-route to the DEAD broker again<br/>→ whole batch group wedges<br/>→ even healthy partitions time out"]
  X -- "yes (the fix)" --> Z["re-fetch metadata, re-route to new leaders<br/>→ all partitions deliver"]

Because a dispatch retries its batch group as one unit, the partitions whose leaders were alive wedged alongside the dead-broker ones — a 6-partition burst went 0/6 (60 s timeout). The fix: a wire-failure retry now invalidates the affected partitions before retrying, so the next attempt re-fetches metadata and re-routes — mirroring Java’s NetworkClient requesting a metadata update on server disconnect. After it: 6/6.

Delivery timeout

delivery.timeout.ms is the hard ceiling on a record’s lifetime across all retries. When it expires, the delivery fails with a DeliveryTimeout naming the batch that actually expired first (not an arbitrary one) — a small accuracy fix that fell out of the same investigation.

Back-pressure, not unbounded growth

Under load the producer applies back-pressure rather than buffering without bound: send blocks up to max.block.ms when buffer.memory is exhausted, and the wire pipeline rejects with Backpressure when every in-flight slot is full. The goal is bounded memory under sustained overload — the property the production-acceptance soak is meant to confirm over hours.

The consumer’s storms

The consumer weathers its own set, each covered in depth in Part V:

• A moved coordinator (broker restart, __consumer_offsets reassignment) answers NOT_COORDINATOR; kacrab drops the cached coordinator and re-discovers it, retrying commits and rejoins — without this, every commit fails forever after a failover (rebalancing).
• An aged-out or out-of-range offset is partition-local and routine: the one partition resets via auto.offset.reset, the rest of the poll keeps flowing (fetching).
• A truncated log after leader change is detected with OffsetForLeaderEpoch and the position steps down to the divergence point instead of reading offsets that no longer exist (KIP-320).

Field notes

• Every failure here maps to a time budget you own: delivery.timeout.ms for produced records, max.block.ms for buffer waits, default.api.timeout.ms for consumer/admin calls. Set them to your real freshness requirements — the field guide has the ladder.
• Back-pressure symptoms (“send is hanging!”) usually mean the budget is working: the buffer is full because the cluster is slow. Fix the cluster or widen the buffer knowingly; don’t reach for max.block.ms=0 first.
• The burst-wedge above is the argument for testing your own failover: it only reproduces with multiple partitions, a mid-burst broker loss, and batching — no unit test finds it.

The admin client

With the producer’s road behind us, this leg is a side expedition — mapping the whole administrative territory before doubling back for the consumer. The admin feature adds kacrab::admin::AdminClient, a native implementation of Apache Kafka’s Admin interface: the full Kafka 4.3.0 operation surface — 62 operations — on the same wire/session layer (and therefore the same TLS/SASL auth) as the producer. The real discovery of this leg wasn’t any single operation; it was that routing is the whole game — and that a handful of wire-version traps only a real broker could reveal.

Shape and conventions

The admin API follows the producer’s conventions rather than Java’s result-object style:

• methods are snake_case mirrors of Java (create_topics, describe_configs, list_consumer_group_offsets, …);
• they return plain Result<T> / Result<()> — no .values() / .all() result objects;
• each call takes a per-call options struct (CreateTopicsOptions, …);
• the facade mirrors Java: AdminClient::new / from_client_config / from_properties / from_map, plus ClientConfig::create_admin.

Domain types Kafka places in org.apache.kafka.common — TopicPartition, OffsetAndMetadata, ConsumerGroupMetadata, Node — live in the always-compiled kacrab::common module and are re-exported by both producer and admin, so a TopicPartition is the same type on either surface.

Operation coverage

All 62 Admin operations are implemented, including topics/partitions, describe/incremental-alter configs, ACLs, consumer groups & offsets, list_offsets/delete_records/elect_leaders, producer/transaction inspection (describe_producers, describe/list/abort_transaction, fence_producers), partition reassignments, delegation tokens, client quotas, user SCRAM credentials, log dirs, feature/update_features, KRaft quorum/voters, the Kafka 4.x share and streams group families, and the client_instance_id/metrics accessors. The one interface method kacrab keeps that upstream 4.3.0 removed is the legacy alter_configs (its RPC still serves older brokers).

Routing model

Different operations reach the cluster differently, and the admin client encodes each pattern once:

• Controller-routed (create/delete topics, partition & config mutations, ACLs, KRaft voters, update_features): sent to the metadata-reported controller, retrying against a freshly discovered controller on NOT_CONTROLLER (route_to_controller).
• Coordinator-routed (consumer/share/streams groups, group offsets, transactions): a FindCoordinator resolves the group/transaction coordinator, its endpoint is registered, and the request is sent there. Transient coordinator errors (NOT_COORDINATOR, COORDINATOR_NOT_AVAILABLE, COORDINATOR_LOAD_IN_PROGRESS) — from either FindCoordinator or the response — trigger a re-resolve-and-retry (route_to_coordinator), matching the Java client. This matters on a freshly started broker whose coordinator topics load lazily.
• Per-leader (list_offsets, delete_records, describe_producers): the target partitions are grouped by their current leader from metadata and one request is sent to each leader.
• Any broker / broadcast (describe_cluster, list_consumer_groups, list_transactions, config-resource listings): sent to one live broker, or fanned out to every broker and aggregated.

Metrics

AdminClient::metrics() returns an AdminMetricsSnapshot — kacrab’s native analogue of Java’s Admin.metrics(), a typed struct rather than a Map<MetricName, KafkaMetric>. Every broker request funnels through one send_metered helper, so the snapshot reflects all routing paths (controller-routed, coordinator-routed, per-leader, and broadcast) uniformly:

• request_total / request_error_total — completed requests and how many failed at the wire/transport layer;
• request_latency_avg_nanos / request_latency_max_nanos — mean and peak request latency;
• buffer_pool — the shared wire BufferPoolStats (read/write buffer acquire / reuse / release counters) at snapshot time.

The counters live behind an Arc, so AdminClient clones report the same aggregate.

Wire-version pitfalls (found by real-broker verification)

Verifying against a real Apache Kafka 4.3.0 broker surfaced a few version- sensitivity traps that the pure-unit tests could not:

• ApiVersions client identity. describe_features (which reuses ApiVersions) must send a non-empty client_software_name; v3+ brokers reject an empty one with INVALID_REQUEST.
• Topic name vs. topic id. OffsetCommit/OffsetFetch v10 dropped the topic name in favour of the topic id, and the strict codec refuses to serialize a stale name at v10. The admin client resolves topic ids from metadata and sends the name only when the id is unknown.
• Optional-API version negotiation. Client telemetry is optional, so a broker may advertise a lower max version for GetTelemetrySubscriptions than the client supports; client_instance_id uses the broker-negotiated version rather than the client maximum.

Verification

Every operation’s wire round-trip is exercised against a real broker (kacrab/tests/real_kafka_admin*.rs, run with --ignored):

• real_kafka_admin_smoke — core ops over docker-compose.kafka.yml.
• real_kafka_admin_extended — ACLs, quotas, SCRAM, transactions, and the share/streams group families over docker-compose.kafka-admin.yml (a broker with StandardAuthorizer and the share/streams features enabled).
• real_kafka_admin_token — delegation tokens over SASL against docker-compose.auth.yml.

Operations that need cluster state the test cannot easily create (a live group member, a hanging transaction) are asserted at the wire layer: a well-formed broker error code proves the request encoded and the response decoded correctly, even when the operation cannot semantically succeed.

The consumer client

The road back: everything the producer wrote, read again — in order, from the right place, shared fairly across a group, surviving the log moving underneath. Consuming turned out to be the stateful half of the journey; where the producer defends one invariant (sequences ascend), the consumer juggles three state machines at once. The consumer feature adds kacrab::consumer::Consumer, a native implementation of Apache Kafka’s Consumer, built on the same wire/session layer (and therefore the same TLS/SASL auth) as the producer and admin clients. Like the rest of kacrab, “Java-compatible” means Kafka-protocol- and behaviour-compatible, not a literal port.

It supports manual partition assignment, topic and pattern subscription, and both consumer-group protocols — the classic client-side-assignment protocol (eager range/roundrobin/sticky and incremental cooperative-sticky assignors) and the KIP-848 server-side protocol (group.protocol=consumer) — plus record fetching with incremental fetch sessions, offset commit/fetch, and interceptors.

This chapter is the architectural tour. The algorithms behind it get their own deep dives: Group membership & rebalancing (the two protocols, the assignors, cooperative-sticky and KIP-848 reconciliation) and Fetching, positions & offsets (the position state machine, incremental fetch sessions, per-partition error recovery, truncation detection, and in-order commits).

Shape and conventions

The consumer follows the producer/admin conventions rather than Java’s:

• methods are snake_case mirrors of Java (subscribe, assign, poll, commit_sync, seek, position, pause, …);
• construction mirrors the other clients: Consumer::new / from_client_config / from_properties / from_map;
• keys and values are raw bytes (ConsumerRecord.key/value: Option<Bytes>), matching the producer’s bytes-first ProducerRecord; a typed deserializer layer rides on top rather than threading generics through the fetch pipeline;
• domain types (TopicPartition, OffsetAndMetadata, ConsumerGroupMetadata) come from the shared kacrab::common module, so they are the same types the producer and admin use.

The consumer is single-owner and not Sync: poll(Duration) drives fetch and rebalance I/O on the caller’s task (the classic protocol adds a dedicated background heartbeat task; the KIP-848 heartbeat runs from poll), mirroring the Java consumer’s user-thread model.

Manual assignment and fetching

assign(partitions) takes direct control of a partition set (no group). poll then, each round:

1. refreshes metadata for the assigned topics;
2. resolves an initial position for any partition without one via auto.offset.reset (earliest/latest through ListOffsets, or a NoOffsetForPartition error under none);
3. issues one Fetch per partition leader and decodes the returned record batches into ConsumerRecords, advancing each partition’s position and respecting max.poll.records.

seek/seek_to_beginning/seek_to_end, position, and pause/resume/ paused give the usual position control; wakeup interrupts a blocking poll.

Fetch negotiates up to the broker’s version: v13+ keys partitions by topic id (KIP-516), resolved from the routing metadata and mapped back to names on the response; a topic without an id (or a pre-v13 broker) downgrades that fetch to v12, the last name-keyed version, matching Java. Fetches use incremental fetch sessions (KIP-227): the first fetch to a leader is a full fetch that opens a session, and later fetches send only the partitions whose position changed (plus a forgotten list for ones no longer fetchable), so the broker returns only partitions with new data. Behaviour is identical — it is a smaller request. Positions are also validated with OffsetForLeaderEpoch after a leader change and reset on truncation (KIP-320).

Group subscription and rebalancing

subscribe(topics) (requires a group.id) joins the consumer group on the next poll; subscribe_pattern(regex) subscribes to every matching topic, re-matched against the live topic list (excluding internal topics per exclude.internal.topics). Which protocol runs depends on group.protocol.

Classic (group.protocol=classic, the default) uses the client-side protocol:

• FindCoordinator locates the group coordinator (retried with backoff while a freshly started broker loads __consumer_offsets);
• JoinGroup registers the member (transparently retrying the MEMBER_ID_REQUIRED round), and the elected leader runs the chosen assignor — range, roundrobin, sticky, or the incremental cooperative-sticky — over every member’s subscription using partition counts from metadata;
• SyncGroup distributes the assignment; the consumer resumes each assigned partition from its committed offset before fetching;
• a dedicated background task heartbeats the group (poll-independent) and flags a rejoin on REBALANCE_IN_PROGRESS / ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID;
• close sends a best-effort LeaveGroup.

Under cooperative-sticky the subscription reports each member’s currently owned partitions and the assignor withholds any partition still owned by another member until it is revoked, so a partition is never owned by two members at once (KIP-429). partition.assignment.strategy selects the assignor and defaults to Java’s range,cooperative-sticky. Static membership (group.instance.id) and enforce_rebalance are supported.

KIP-848 (group.protocol=consumer) uses the server-side protocol: the consumer generates its own member id and drives membership with a single ConsumerGroupHeartbeat RPC, reporting its subscribed topics and owned partitions each heartbeat and reconciling toward the coordinator-computed (topic-id-keyed) target assignment incrementally. Fencing rejoins from epoch 0; close sends a leaving heartbeat.

Offsets

For a consumer carrying a group.id (manual-assignment or subscribed):

• commit_sync commits the current position of every assigned partition; commit_sync_offsets(map) commits explicit offsets; commit_async commits without blocking, invoking a callback with the result. With enable.auto.commit, positions are committed on a background interval and before each rebalance. Offsets carry the leader epoch, and a group member’s commits identify themselves with the member id + generation/epoch (required by the coordinator, and by KIP-848 in particular).
• committed(partitions) reads back committed offsets, omitting partitions with none.
• beginning_offsets/end_offsets/offsets_for_times query the log bounds and timestamp offsets; current_lag reports a partition’s lag.
• group_metadata() returns the ConsumerGroupMetadata.

OffsetCommit is capped at v9 and OffsetFetch at v7 so both stay topic-name keyed, sidestepping the v10/v8 topic-id strict-codec forms (the same trap the admin offset paths handle).

Deserializers, interceptors, and metrics

Records are bytes-first; record.deserialized(key, value) maps them through typed ConsumerDeserializers (BytesDeserializer, ByteArrayDeserializer, StringDeserializer, or your own). add_interceptor registers a ConsumerInterceptor whose on_consume can rewrite or filter each poll’s records before they reach the caller and whose on_commit observes committed offsets; the chain is panic-isolated. metrics() returns a typed snapshot (poll/records/fetch/commit/heartbeat/rebalance totals plus the wire buffer pool), and client_instance_id() returns the broker-assigned telemetry id.

Verification

The consumer is exercised end-to-end against a real Apache Kafka 4.3.0 broker (kacrab/tests/real_kafka_consumer.rs, run with --ignored):

• manual assignment: consume records a kacrab producer wrote, then commit_sync and read the committed offset back;
• subscription: one subscriber owns both partitions of a two-partition topic and consumes every record;
• rebalance: two consumers in one group split to one partition each (driven from independent tasks, like real consumers) and together consume everything;
• cooperative-sticky: two consumers negotiate the incremental protocol and converge to a clean split with no double-owned partition;
• assignors and offsets: the roundrobin assignor, the offset-query APIs, and auto/async commit;
• pattern subscription: a regex joins exactly the matching topics;
• interceptors: on_consume/on_commit observe every record and commit;
• KIP-848: a group.protocol=consumer subscriber joins via ConsumerGroupHeartbeat, is assigned both partitions, consumes, and commits.

Truncation detection (KIP-320) and the fetch-session state machine are covered by unit tests, since a real truncation / leader change cannot be staged on the single-broker compose.

A latent connection bug surfaced during verification: a coordinator advertised as localhost resolving to a dead IPv6 loopback made a pinned connection hang. The fix centralized DNS in the wire layer — brokers are re-resolved on connect, IPv4-first, honouring client.dns.lookup=use_all_dns_ips — so the producer, admin, and consumer all benefit (see First contact).

Field notes

The consumer’s configuration is where semantics live — the consumer field guide is the full treatment. The three decisions to make before deployment:

• Commit discipline: auto-commit (periodic, at-least-once with idempotent handlers) vs manual commit_sync after processing.
• auto.offset.reset: latest, earliest, or none — chosen, not inherited, because it also decides what happens when retention outruns a stopped consumer.
• Rebalance posture: cooperative-sticky alone (or group.protocol=consumer on 4.x clusters) plus group.instance.id per instance for calm rolling deploys.

Sharing the load: groups & rebalancing

This is the consumer’s correctness core, and the deepest cave on the road back. A consumer group spreads the partitions of the subscribed topics across its members and keeps that division valid as members come and go — through two different protocols, four assignors, and a handoff dance whose entire purpose is to defend one invariant.

The invariant we defend

At any instant, every partition of a subscribed topic is owned by at most one member of the group, and in steady state by exactly one. A partition owned by two members at once means two consumers process the same records — double-processing. A partition owned by none means records pile up unconsumed. Every rebalance must move ownership from the old division to the new one without ever transiently double-owning a partition.

kacrab implements both Kafka group protocols behind the same subscribe/poll surface, selected by group.protocol:

Finding the coordinator

Both protocols first locate the group’s coordinator — the broker that owns the group’s slice of __consumer_offsets — with FindCoordinator. A freshly started broker loads that log lazily, so the first lookups can answer COORDINATOR_NOT_AVAILABLE / COORDINATOR_LOAD_IN_PROGRESS; kacrab retries with backoff, like the Java client.

The coordinator is cached (coordinator_id). It can move — a broker restart or a __consumer_offsets partition reassignment hands the group to a new broker. When any operation then answers NOT_COORDINATOR / COORDINATOR_NOT_AVAILABLE / COORDINATOR_LOAD_IN_PROGRESS, kacrab drops the cached id so the next call re-discovers it; commits and the join/sync loop retry after re-finding. Without this, a moved coordinator would fail every commit and rejoin permanently — a real bug the review caught (see Failure modes).

The classic protocol

sequenceDiagram
    participant M as Member
    participant C as Coordinator
    participant L as Leader (a member)
    M->>C: FindCoordinator(group)
    M->>C: JoinGroup(subscription, owned)
    C-->>M: generation, member_id, protocol, [members if leader]
    Note over L: the leader runs the assignor over<br/>every member's subscription
    L->>C: SyncGroup(assignments per member)
    M->>C: SyncGroup()  %% followers send empty
    C-->>M: this member's assignment
    loop every heartbeat.interval.ms (background task)
        M->>C: Heartbeat(generation, member_id)
        C-->>M: OK | REBALANCE_IN_PROGRESS → rejoin
    end
    M->>C: LeaveGroup()  %% on close (dynamic members)

JoinGroup collects every member’s subscription at the coordinator and elects one member the leader. The leader — not the broker — runs the assignor and returns each member’s partitions in SyncGroup; the coordinator relays them. The subscription and assignment travel as the version-prefixed ConsumerProtocol blobs; kacrab caps the surrounding RPCs at name-keyed versions to sidestep the topic-id strict codec.

A dedicated background task heartbeats on heartbeat.interval.ms independent of poll cadence (Java’s HeartbeatThread); a REBALANCE_IN_PROGRESS / ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID reply flags a rejoin that the next poll performs.

The assignors

The leader picks the assignor the whole group supports (from partition.assignment.strategy) and runs it over (member → subscribed topics) plus partition counts from metadata.

range/roundrobin/sticky rebalance eagerly: on any change every member revokes everything and the new assignment is applied wholesale. Simple, but it stops the world.

Cooperative rebalancing (KIP-429)

cooperative-sticky keeps consuming through a rebalance and only moves the partitions that actually change hands — while still never double-owning one. It does this by making the assignor and the members cooperate over two rounds.

Two ingredients:

1. Each member reports the partitions it currently owns in its subscription blob (ConsumerProtocol v1).
2. The assignor computes a balanced sticky target, then withholds any target partition that another member still owns — it is only granted once its current owner has revoked it.

sequenceDiagram
    participant A as Member A (owns t0..t3)
    participant Co as Coordinator
    participant B as Member B (new)
    Note over A,B: Round 1 — B joins
    A->>Co: JoinGroup(owned=t0..t3)
    B->>Co: JoinGroup(owned=[])
    Note over A: leader targets A=t0,t1 · B=t2,t3<br/>but t2,t3 are still owned by A → withheld
    Co-->>A: assigned t0,t1  (revoke t2,t3)
    Co-->>B: assigned []      (its share is withheld)
    A->>Co: JoinGroup(owned=t0,t1)   %% A revoked, rejoins
    Note over A,B: Round 2 — t2,t3 now unowned
    Co-->>B: assigned t2,t3
    Note over A,B: converged — t2,t3 never owned by both

The client side of the handoff (apply_assignment with cooperative semantics):

• Retained partitions keep their live fetch position — they are never interrupted.
• Revoked partitions (owned but not in the new assignment) are dropped and a follow-up rejoin is triggered so the coordinator can hand them on.
• Added partitions resume from their committed offset.

A member leaving is cheaper: its partitions become unowned immediately, so the survivors pick them up in a single round.

Why kacrab’s sticky assignor withholds client-side

In the classic protocol there is no server arbiter of “who revoked what”, so the assignor enforces the invariant: cooperative_sticky_assign computes the balanced target and then removes from each member’s list any partition a different member still reports owning. The withheld partition reappears — unowned — only after its previous owner drops it and rejoins. Verified with two real consumers: they converge to a clean split with no partition ever in both assignments.

KIP-848: server-side reconciliation

The next-generation protocol (group.protocol=consumer) moves assignment into the coordinator and collapses membership into a single RPC.

sequenceDiagram
    participant M as Member (self-assigned UUID id)
    participant C as Coordinator
    M->>C: ConsumerGroupHeartbeat(epoch=0, subscribed, owned=[])
    C-->>M: member_epoch=1, assignment (topic-id keyed)
    Note over M: reconcile: resolve ids→names,<br/>revoke gone, add granted
    loop every heartbeat_interval_ms (from the response)
        M->>C: ConsumerGroupHeartbeat(epoch, owned)
        C-->>M: [new assignment when it changes]
    end
    M->>C: ConsumerGroupHeartbeat(epoch=-1)  %% leave, on close

The client generates its own member id (a UUID kept for the process lifetime), joins at member_epoch=0, and reports its subscribed topics and currently owned partitions on every heartbeat. The coordinator computes the target and sends the member the partitions it may hold now.

The double-own invariant, now server-enforced

Reconciliation is server-driven: the coordinator withholds a partition from a member’s target until the previous owner has revoked it (reported a reduced owned set in a heartbeat). So applying the received assignment directly — revoke what’s gone, add what’s granted — never double-owns a partition; the withholding that the client-side cooperative assignor does explicitly is done by the coordinator here. Revocation is reflected in the next heartbeat’s owned set.

Assignments arrive keyed by topic id, resolved to names against cluster metadata. Fencing (FENCED_MEMBER_EPOCH / UNKNOWN_MEMBER_ID) drops the membership and rejoins from epoch 0. Because the heartbeat is the protocol, it runs from poll at the interval the coordinator dictates.

Static membership & clean shutdown

A group.instance.id makes the member static: it keeps its identity across a restart, so a quick bounce does not trigger a rebalance (the coordinator waits out the session). On close, a dynamic member leaves the group (LeaveGroup, or a leaving heartbeat under KIP-848); a static member stays. close is bounded by request.timeout.ms so a hung coordinator cannot hang it.

What’s verified

Against a real Apache Kafka 4.3.0 broker (Verification): a single subscriber owning both partitions; two consumers splitting a topic and rebalancing to one each; the roundrobin assignor; cooperative-sticky converging two consumers to a clean split; and a group.protocol=consumer member joining via ConsumerGroupHeartbeat, being assigned both partitions, consuming, and committing. The multi-member handoff is exercised end to end for the classic cooperative path; the KIP-848 handoff relies on the coordinator’s documented server-side withholding.

Field notes

• The default partition.assignment.strategy (range,cooperative-sticky) negotiates to eager range in steady state — to actually get the incremental handoff above, set cooperative-sticky alone.
• On Kafka 4.x clusters, group.protocol=consumer gets you server-side reconciliation and one-RPC membership — prefer it for new groups.
• group.instance.id (static membership) is the cheapest rebalance-storm fix in Kafka: a bounce within session.timeout.ms triggers no rebalance at all.
• Heartbeats are background; poll cadence is the liveness you own — the max.poll.interval.ms arithmetic is in the consumer field guide.

Reading in order: fetching, positions & offsets

Once a consumer owns partitions, the rest of the road is reading: in order, from the right place, surviving the log moving underneath, remembering how far it got. Three state machines cooperate — the position of each partition, the per-broker fetch session, and the committed offset at the coordinator — and this leg is also where the consumer’s performance was won: the three mechanisms that carry the benchmark numbers all live in the fetch loop below, each added after a measurement exposed its absence.

The invariant we defend

Each assigned partition has a single monotonic fetch position. Records are delivered in offset order starting at that position; the position only ever advances past records actually handed to the caller. A partition whose position becomes invalid (aged out, truncated) is re-resolved, never skipped and never stuck — and one bad partition never stalls the others.

The position state machine

A newly assigned partition has no position. Before it can be fetched, one is resolved — from the committed offset if there is one, otherwise from auto.offset.reset (earliest/latest via ListOffsets, or a NoOffsetForPartition error under none).

stateDiagram-v2
    [*] --> Unpositioned: assigned
    Unpositioned --> Positioned: committed offset, or auto.offset.reset (ListOffsets)
    Unpositioned --> Error: reset=none & no committed offset
    Positioned --> Positioned: fetch → advance past delivered records
    Positioned --> Paused: pause()
    Paused --> Positioned: resume()
    Positioned --> Positioned: seek(offset)
    Positioned --> Unpositioned: OFFSET_OUT_OF_RANGE → reset

seek/seek_to_beginning/seek_to_end overwrite the position; pause/resume toggle whether a positioned partition is fetchable. Only assigned, unpaused, positioned partitions are eligible for a fetch.

The fetch loop

One fetch typically returns far more than one poll may yield (up to max.partition.fetch.bytes per partition against a max.poll.records cap of 500), so responses are buffered across polls — Java’s completedFetches + nextInLineFetch, kacrab’s FetchBuffer. Each poll first drains the buffer: up to max.poll.records records are returned straight from memory, advancing each partition’s position past what was yielded. Buffered blobs decode lazily, one record batch at a time (Java’s CompletedFetch iterator), so memory holds the raw blobs plus roughly one batch of decoded records. A partition is only re-fetched once its buffered data runs dry — for a 5M-record scenario that is ~13 Fetch RPCs instead of 10,000. Buffered data is invalidated lazily at drain time (a seek/reset moved the position, or a rebalance revoked the partition) and retained across pause/resume.

The next fetch is pipelined (Java’s network thread): as soon as fetchable partitions run dry, their Fetch is spawned as a background task — grouped by leader, one in flight — while the caller keeps draining buffered records; a poll that finds the buffer empty awaits the in-flight fetch only up to its own timeout. One guard matters (Java’s buffered-node gate): a node that still hosts buffered partitions is not fetched from at all. Omitting its buffered partitions would drop them from the broker’s fetch-session cache, and a request listing only caught-up partitions would long-poll fetch.max.wait.ms while the buffer drains dry behind it. Fetches negotiate up to the broker’s version: v13+ keys partitions by topic id (KIP-516), with ids resolved from the routing metadata and mapped back to names on the response (Java’s sessionTopicNames); a topic with no id — or a pre-v13 broker — downgrades that fetch to v12, the last name-keyed version, exactly like Java’s AbstractFetch.

Because the broker holds a fetch for up to fetch.max.wait.ms waiting for fetch.min.bytes, kacrab clamps that wait to the caller’s remaining poll budget — a poll(200ms) is not blocked for a 500 ms (or 2 s) broker wait. The long-poll means an idle poll doesn’t busy-spin, while the clamp keeps short polls responsive.

Incremental fetch sessions (KIP-227)

Re-sending every partition’s offset on every fetch is wasteful when the assignment is stable. A fetch session lets the broker remember the set: the first fetch to a leader is full and opens a session; later fetches send only the partitions whose position changed, plus a forgotten list for ones no longer fetchable. The broker replies with only the partitions that have new data. (The buffered-node gate above keeps buffered partitions from churning through the forgotten list: a node is simply not fetched from until its buffered data drains.)

stateDiagram-v2
    [*] --> Full: epoch 0, session_id 0
    Full --> Incremental: response session_id ≠ 0 → epoch 1
    Incremental --> Incremental: epoch++, send only changed + forgotten
    Full --> Full: response session_id = 0 (broker declined) → stay full
    Incremental --> Full: INVALID_FETCH_SESSION_EPOCH / ID_NOT_FOUND → reset

The behaviour is identical to full fetches — it is purely a smaller request. A stale-session error drops the session and re-establishes it with a full fetch on the next poll. kacrab keeps this state per broker, across polls.

One bad partition doesn’t sink the poll

A Fetch response carries a status per partition. Treating any of them as fatal is wrong — most are routine and partition-local. In particular OFFSET_OUT_OF_RANGE (retention aged the committed offset out, or a seek landed past the log end) is a normal condition that must reset the position; failing the whole poll would loop forever, because the out-of-range partition keeps its stale position and so is never re-resolved.

kacrab classifies each partition’s error, mirroring Java’s AbstractFetch, and keeps the records decoded from the healthy partitions in the same response:

The response handling is a pure function returning FetchProgress { partitions, resets, stale }, unit-tested with synthesized responses and verified end to end: a fetch positioned past the log end resets to earliest and consumes every record instead of erroring.

Truncation detection (KIP-320)

A leader change can expose that the log was truncated below the consumer’s position — the offsets it was about to read no longer exist. After a leader epoch bump is visible in metadata, kacrab validates the position with OffsetForLeaderEpoch: the leader reports the end offset of the largest epoch at or below the position’s epoch. If that end offset is below the position, the log diverged and the position resets down to it; otherwise the position is confirmed and its recorded epoch advances so it isn’t re-validated. (The decision and request encoding are unit-tested; a real truncation cannot be staged on the single-broker rig.)

Committing offsets

Committed offsets live at the coordinator (OffsetCommit/OffsetFetch, capped at name-keyed versions). A commit carries the member identity — generation/epoch + member id — which the coordinator requires from a group member under either protocol. commit_sync blocks; committed reads back; both retry once after re-finding a moved coordinator.

Asynchronous commits stay in order

commit_async must not let a later commit lose to an earlier, slower one — a stale commit landing last would move the committed offset backwards and cause reprocessing after the next rebalance.

The concurrent-task adaptation

A naïve commit_async spawns a task per call, and two tasks race to the coordinator with no ordering guarantee. kacrab instead routes async commits through a single unbounded queue drained by one worker that applies each commit to completion before the next — the same guarantee Java gets from its single network thread. Because commit_async takes &mut self, calls are serialized at the source, so queue order is call order, and the worker preserves it. Verified: ten back-to-back commits with increasing offsets fire their callbacks in call order and leave the last offset committed.

sequenceDiagram
    participant App
    participant Q as commit queue (mpsc)
    participant W as commit worker
    participant Co as Coordinator
    App->>Q: commit_async(100)
    App->>Q: commit_async(200)
    W->>Co: OffsetCommit(100)
    Co-->>W: ok → callback(100)
    W->>Co: OffsetCommit(200)
    Co-->>W: ok → callback(200)
    Note over W,Co: FIFO — 200 always lands after 100, never before

Auto-commit

With enable.auto.commit, the current positions are committed on a background interval and once more before each rebalance (so a partition about to be revoked commits its progress). Auto-commit failures are swallowed and retried on the next interval, matching Java’s best-effort async auto-commit.

What’s verified

Against a real Apache Kafka 4.3.0 broker (Verification): manual assign → consume → commit_sync → read committed; the offset-query APIs and lag; auto and async commit; async commits applying in call order with no regression; a fetch past the log end resetting and recovering; and a short poll returning without waiting out a long fetch.max.wait.ms. The fetch-session state machine and truncation classifier are unit-tested (a real leader change / truncation can’t be staged on one broker).

Field notes

• max.poll.records shapes your poll-loop cadence, not network traffic — the buffer and prefetch above mean small values don’t add RPCs. Size it to fit max.poll.interval.ms.
• Raise fetch.min.bytes to make the broker batch for you on high-throughput topics; kacrab’s poll-budget clamp keeps short polls responsive anyway.
• OFFSET_OUT_OF_RANGE is weather, not an outage — but what it resets to is your auto.offset.reset choice. Under latest, an aged-out offset is a silent skip; choose deliberately.
• Prefer commit_async on the hot path (kacrab guarantees FIFO application) with one commit_sync at shutdown.

The consumer field guide develops all four.

Testing, coverage & CI

An expedition’s claims are worth what its instruments can prove. Everything asserted in Parts I–V rests on three layers of evidence, each catching what the layer below cannot:

The first two are the bulk of the suite and gate every change. The third — SASL/TLS, compression on disk, multi-broker failover — is covered in Verification against real brokers.

The Java oracle

A Rust-only round trip is a closed loop: kacrab encodes and kacrab decodes, so a subtly wrong CRC, varint, or murmur2 seed passes anyway. The oracle breaks the loop by pinning kacrab’s output to values produced by Kafka’s own Java code: murmur2 hashes for every key length, CRC32C frames, zig-zag varints, the sticky partitioner’s distribution. These are committed fixtures, so the parity check runs with no Java toolchain in CI. See Design decisions & Java parity.

Coverage — measured with cargo-llvm-cov

CI gates line coverage with cargo-llvm-cov. Maintained-source coverage is ~87.5% (generated protocol/config artifacts excluded via --ignore-filename-regex), with the producer module around 92%. Generated code is held to a different standard — it is validated by the generator’s own tests and the Java oracle, not by line coverage — so counting it would only dilute the signal (the raw all-files figure is ~63%, dominated by message structs for APIs not yet wired).

cargo llvm-cov --workspace --all-features \
  --ignore-filename-regex '(benches/|kacrab-codegen/src/main\.rs|kacrab-macros/src/lib\.rs|kacrab/src/config/catalog\.rs|kacrab-protocol/src/generated)'

Why not tarpaulin

The coverage tool itself turned out to matter. kacrab’s test suite leans heavily on real timeouts and blocking windows — max.block.ms, delivery.timeout.ms, leadership-retry deadlines — often set to a few milliseconds so a test can assert the timeout fires. cargo-tarpaulin’s instrumentation slows execution by ~10–50×, enough to blow through those windows: a FindCoordinator + InitProducerId round trip that takes microseconds bare would exceed a 30 ms budget under instrumentation, so the producer correctly — but spuriously — timed out. The result was a shifting set of flaky failures, different tests from one run to the next.

cargo-llvm-cov uses LLVM source-based coverage and runs the tests at near-native speed. The same timeout tests that flaked under tarpaulin finish in 0.07 s instrumented versus 0.06 s bare — the timing windows hold, so coverage is both reliable and a real gate, not a best-effort report.

The lesson generalises: for an async, timeout-driven codebase, prefer a coverage tool that doesn’t perturb timing. A coverage job that flakes is worse than none — it trains everyone to ignore red.

The CI pipeline

Three jobs run on every push to master and every pull request:

Two deliberate refinements keep the pipeline honest and cheap:

• The test job is the source of truth, not coverage. It runs the whole suite at native speed; the coverage job measures the same suite but is judged only on the coverage floor. Correctness and measurement are separated on purpose.
• Docs-only changes skip the code CI. A paths-ignore filter means editing this book, a README, or a license file does not trigger the ~20-minute fmt/clippy/test/coverage run. The book has its own deploy (docs.yml → GitHub Pages); a change that touches both code and docs still runs the full pipeline.

Verification against real brokers

The map is not the territory. Unit tests prove kacrab is self-consistent — they do not prove it is Kafka-consistent: a codec framed the wrong way, or a SASL round skipped, passes a Rust-only round trip because kacrab is on both ends. Most of the discoveries in this book — the OAUTHBEARER error round, the GSSAPI security layer, the burst wedge, the IPv6 hang, the topic-id codec traps — were found here, walking the actual territory: every path exercised end-to-end against real Apache Kafka 4.3.0, with the environments captured as docker-compose.*.yml files.

Quote

“A few minutes of docker compose proves a path works; proving it holds up needs hours of load.” — see Benchmarks and the production acceptance plan.

SASL — every mechanism

docker-compose.auth.yml (a single broker with PLAIN + SCRAM-256/512 + OAUTHBEARER) and docker-compose.gssapi.yml (a real MIT Kerberos KDC + a GSSAPI broker). Each mechanism authenticates and runs ApiVersions; negative cases confirm rejection is fast and clear.

TLS — SSL, SASL_SSL, mutual TLS

docker-compose.tls.yml publishes three listeners; certs are generated by scripts/gen-tls-certs.sh. One-way SSL, SASL/SCRAM over TLS, and mutual TLS all authenticate; an untrusted server certificate is rejected fast with the rustls reason.

Compression — every codec, proven on disk

The decisive test: produce gzip/snappy/lz4/zstd batches, then verify three things end to end — the broker accepts the produce, kafka-dump-log shows the on-disk batch stored with the correct codec (so a silently-uncompressed send would fail), and kafka-console-consumer decompresses the exact payloads back.

on-disk batch codecs: [gzip×3, snappy×3, lz4×3, zstd×3]
all 12 compressed records round-tripped through the broker and CLI consumer

Consumer — groups, offsets, and both protocols

kacrab/tests/real_kafka_consumer.rs (12 tests, --test-threads=1) walks the consumer surface against a real broker: manual assignment and commit round-trips; one subscriber owning a whole topic; two consumers splitting it via a real rebalance; cooperative-sticky converging with no double-owned partition; the roundrobin assignor; pattern subscription; interceptors; offset queries and lag; async commits landing in call order; seek-past-end recovering; and a KIP-848 (group.protocol=consumer) member joining, consuming, and committing through ConsumerGroupHeartbeat. See the consumer chapters for what each proves.

Multi-broker — dispatch & leadership-change

docker-compose.cluster.yml runs a 3-broker KRaft cluster. The tests confirm records route to every broker’s leaders, and that a broker loss re-routes affected partitions to their new leaders without wedging the healthy partitions batched alongside them — the bug described in Idempotency & transactions and Failure modes. A 6-partition burst across a broker loss delivers 6/6.

Admin — the full operation surface

docker-compose.kafka.yml (core ops), docker-compose.kafka-admin.yml (a broker with StandardAuthorizer and the share/streams features, for ACLs and the Kafka 4.x group families), and docker-compose.auth.yml (SASL, for delegation tokens) back kacrab/tests/real_kafka_admin*.rs. Every one of the 62 admin operations is exercised across all four routing paths (controller, coordinator, per-leader, broadcast). Operations that need cluster state the test cannot create are asserted at the wire layer — a well-formed broker error code proves correct encode/decode. This pass caught real wire bugs the unit tests could not: an empty ApiVersions client name, the OffsetCommit/OffsetFetch v10 topic-name-vs-id switch, and the missing transient-coordinator-error retry (see The admin client).

Note

These are #[ignore] integration tests — they need the compose environments and (for the cluster/compression cases) the docker CLI. They are not part of the default cargo test run; CI runs the in-process suite.

Benchmarks & methodology

The journey’s final weighing: what did going native actually buy? Numbers are only as good as the method behind them — two of this book’s performance war stories (the SSH-tunnel trap, the getenv lock) are about the harness lying, not the client — so this chapter gives the headline figures and, more importantly, how they were measured and what they do not claim. Full reproduction lives in benches/README.md.

The headline

Measured against native Apache Kafka 4.3.0 single-node KRaft on the same machine, through the public producer API at the default Kafka-compatible config (acks=all, enable.idempotence=true, no compression):

Where the +25–28% comes from

Throughput here is broker-bound: both clients spend most of the run waiting on acks=all round trips, so cheaper per-record CPU barely moves the needle. kacrab’s records/sec edge comes from keeping the broker’s write path busier — a deeper per-partition pipeline plus coalescing one ready batch from every partition into each produce request (on 10 KiB records, where each batch holds a single record, that coalescing is the entire difference between ~540 MB/s and one-record-per-round-trip collapse). The native-vs-JVM win also shows up in efficiency: ~4× less resident memory (no JVM heap/metaspace) and ~1.5× less CPU per record. The Java CPU figure also includes one-time JVM startup + JIT warmup that amortizes over a long-lived producer; the peak-RSS gap is steady-state.

Bench against a native broker. A broker behind a Colima/OrbStack published port is reached through an SSH tunnel that roughly triples request RTT — it silently caps every number (10 KiB throughput measured ~3× lower through the tunnel). And never read env vars on a per-record path in the harness itself: macOS getenv takes a global libc lock, and one env::var call inside the record factory cost ~28% of small-record throughput until it was hoisted.

The latency tradeoff

Java keeps a lower typical latency; kacrab trades it for pipeline depth.

• At max.in.flight=5 kacrab fills the per-partition pipeline (higher p99 on a single low-RTT broker, where the extra depth only adds queue latency). At max.in.flight=1 its p99 drops to ~2 ms at the same throughput.
• Depth pays off the other way under a broker pause: at depth 5 a GC/fsync pause on one in-flight request lets the others drain (p99.9 ~10 ms); at depth 1 the single slot blocks (p99.9 ~100 ms). The gap shrinks in production — broker off the client machine, real network RTT.

The consumer head-to-head

The consumer benchmark mirrors Java’s kafka-consumer-perf-test.sh exactly (fresh group per run, the tool’s own props, 100 ms poll slices, the same CSV columns) against prefilled topics on the same native broker (2026-07-02):

kacrab consumes small records ~1.9× faster and large records ~4× faster than Java at identical defaults, at ~16–20× less memory and ~9–17× less CPU, with a poll() tail 14–25× lower (Java keeps a slightly tighter p99.9 on 10 B records: ~1.9 ms vs ~2.5 ms). Three Java-parity mechanisms carry it, each added after the benchmark exposed its absence: cross-poll fetch buffering (completedFetches — before it, every poll re-fetched the response surplus and 10 B throughput sat at ~132K rec/s), background prefetch with the buffered-node gate (the network thread; without the gate, a fetch listing only caught-up partitions long-polled fetch.max.wait.ms and collapsed throughput 13×), and lazy per-batch decode (CompletedFetch’s iterator — decoding whole blobs up front cost ~536 MiB of allocator churn; per-batch it is ~18 MiB and the p99.9 decode spike halved).

Micro-benchmarks

Criterion benchmarks against local mock brokers cover the hot paths in isolation: the accumulator append/drain, the wire request pipeline (send_to_broker req/s), and multi-broker produce dispatch. They catch hot-path regressions without real broker storage/replication noise.

Honesty about units and scope

kacrab prints payload MiB/s; Java’s perf tool prints decimal MB/s — don’t compare them as the same unit. And these are five-run smoke measurements on a shared host, not release gates. What is not measured here — sustained soak, cross-DC RTT, memory growth over hours, latency-percentile gates — is deliberately scoped in the README’s Production acceptance plan, not claimed.

Foundations every client shares

Every leg of the journey ended with configuration lessons. This part of the book collects them: not a reference dump of every key (the config surface is generated from Kafka’s own metadata and documented on docs.rs), but the settings that decide outcomes — and the mistakes we made or watched brokers punish. This chapter covers what producer, consumer, and admin all share; the next two tune each surface.

One config surface, Java names

kacrab deliberately uses the exact Kafka property names and defaults the Java client uses. Anything you learned from Kafka’s documentation, an ops runbook, or an existing producer.properties transfers verbatim:

#![allow(unused)]
fn main() {
let producer = Producer::builder()
    .set("bootstrap.servers", "kafka-1:9092,kafka-2:9092")
    .set("compression.type", "zstd")
    .build()?;

// or load the same .properties file your Java services use
let consumer = Consumer::from_properties("consumer.properties")?;
}

Values are parsed and validated into typed configs at build time — an invalid value fails construction with the key and the accepted values, not at first use. The typed surface is drift-checked against the upstream Kafka catalog in CI, so a key documented here is a key that actually exists.

Best practice: keep configs boring

The single most reliable Kafka tuning advice is to change few defaults and know why for each one. Kafka’s defaults (which kacrab inherits) already encode a decade of production learning — acks=all, idempotence on, exponential jittered backoff. Every override below states the reason to make; if you can’t state one, don’t override.

Start with the features

kacrab ships default = [] — a bare kacrab = "0.1" compiles nothing usable. Enable every surface you call:

[dependencies]
kacrab = { version = "0.1", features = ["producer", "consumer", "admin"] }

• consumer implies the compression meta-feature: a consumer must decode whatever codec the producer chose, so all four codecs come along.
• compression (and zstd alone) needs a C compiler at build time (libzstd). For a pure-Rust build take only gzip + snappy + lz4 — see Traveling light.
• gssapi (Kerberos) is opt-in and links libgssapi.

bootstrap.servers: plural, on purpose

Bootstrap servers are only used to discover the cluster — after the first metadata response, the client talks to the brokers the cluster advertises. But that first contact is a single point of failure you control:

• List at least two brokers (three across racks/AZs if you have them). A one-entry list means one rebooted broker blocks every fresh client start.
• The entries don’t need to be the whole cluster — they need to be alive at startup.
• If all known brokers become unreachable later, metadata.recovery.strategy=rebootstrap (the default) goes back to the bootstrap list instead of spinning on dead cached endpoints.

What the broker taught us: DNS is part of your config

A coordinator advertised as localhost resolved to the IPv6 loopback first — where nothing listened — and a pinned connection hung forever. kacrab now re-resolves broker hostnames on every connect, tries all returned addresses IPv4-first, and honours client.dns.lookup=use_all_dns_ips. The portable lesson: advertise brokers by names that resolve to addresses clients can actually reach, and prefer use_all_dns_ips so a multi-A-record entry fails over instead of failing.

Name yourself: client.id

Set client.id to something meaningful (orders-api-producer, not the default empty string). It appears in broker request logs, quota enforcement, and kacrab’s own ApiVersions identification — when an SRE asks “who is hammering partition 12?”, this is the answer. One name per logical application, shared across instances; use group.instance.id (consumer) for per-instance identity.

The timeout ladder

Kafka timeouts nest, and misordering them produces “impossible” symptoms. From inner to outer:

• Keep delivery.timeout.ms ≥ linger.ms + request.timeout.ms — kacrab (like Java) validates this, because a delivery window smaller than a single attempt can never succeed.
• Bound retries with time, not counts. Leave retries at its effectively infinite default and shrink delivery.timeout.ms to your real freshness requirement. A time budget degrades gracefully under a slow broker; a retry count is either too small during an incident or meaninglessly large.
• Don’t raise request.timeout.ms to “fix” timeouts. A request that needs more than 30 s is telling you about broker load, an unreachable leader, or (historically, in this client’s own past) an auth failure being silently retried — see below.

Backoff: already jittered, leave it

retry.backoff.ms → retry.backoff.max.ms and reconnect.backoff.ms → reconnect.backoff.max.ms implement exponential backoff with jitter, resetting on success — the same curve as Java. The defaults are right for almost everyone; the one case for raising reconnect.backoff.max.ms is a very large fleet whose simultaneous reconnects can thundering-herd a recovering broker.

Security: fail fast, and mind the one hard boundary

security.protocol picks the transport/auth pairing (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL); sasl.mechanism picks PLAIN, SCRAM-SHA-256/512, OAUTHBEARER, or GSSAPI; ssl.truststore.*/ssl.keystore.* load PEM, JKS, or PKCS12 material with the Java key names. Every combination is verified against real brokers.

Two field lessons:

• A rejected credential is not a transient error. kacrab fails authentication fast with the broker’s real reason. If you see SaslAuthentication at startup, fix the credential — no timeout tuning will help. (Early kacrab retried these under backoff and surfaced “request timed out” 30 seconds later; the security chapter tells that story.)
• sasl.jaas.config class names are the one thing that cannot cross. JVM login modules and callback handlers can’t load in a Rust process. Static PLAIN/SCRAM credentials configure directly; custom token flows use the native authenticator hook (sasl_client_authenticator) instead of a class name. See Design decisions.

Metadata freshness

metadata.max.age.ms (5 min default) caps how stale routing metadata may get; leader changes are additionally learned reactively — from error responses and connection drops — so you rarely need to lower it. If your cluster reassigns partitions frequently and you observe brief retry bursts after moves, lower it toward 60 s rather than disabling anything.

Field notes

• Set two things on day one: a plural bootstrap.servers and a meaningful client.id. Half of production debugging is knowing who is talking to whom.
• Express reliability budgets in time (delivery.timeout.ms, default.api.timeout.ms), not attempt counts.
• Treat auth failures as configuration bugs, not weather. They fail fast by design.
• Prefer client.dns.lookup=use_all_dns_ips in any environment with multi-homed or containerized brokers.

Tuning the producer

Producer tuning is a triangle: durability, throughput, latency — pick the corner that matters and move deliberately toward it. The good news from the benchmarks is that the safe corner is no longer the slow corner: at the default acks=all + idempotence config, kacrab outruns the Java client’s own perf tool. You do not need to trade safety for speed; you tune within safety.

The defaults are the durability recipe

acks=all                    # leader + in-sync replicas must confirm
enable.idempotence=true     # exactly-once, in-order per partition
retries=<effectively ∞>     # bounded by delivery.timeout.ms instead

These are Kafka’s defaults and kacrab keeps them. Resist the classic anti-patterns:

• Don’t set retries=0 to avoid duplicates. That trades duplicates for data loss on any transient error. Idempotence already removes retry duplicates — the broker deduplicates by sequence number (how) — so retries are safe.
• Don’t turn off idempotence “for performance”. Its wire cost is a few bytes per batch; every benchmark in this book ran with it on. Turning it off also forfeits ordering under retry (below).
• Complete the recipe broker-side with min.insync.replicas=2 on replication-factor-3 topics; acks=all with min.insync.replicas=1 only promises one disk.

Ordering depends on idempotence, not just max.in.flight

With max.in.flight.requests.per.connection > 1 and idempotence off, a retried batch can land behind a newer one — silent reordering. With idempotence on, sequence numbers pin the order and up to 5 in-flight requests stay strictly ordered per partition. Keep idempotence on and you keep both pipeline depth and order; the mechanism is the idempotency chapter’s whole story.

Chasing throughput

Throughput comes from batching — fewer, fuller requests. In descending order of effect:

1. linger.ms (default 0): the number-one lever. 5–20 ms lets the accumulator fill batches under moderate traffic; at very high send rates batches fill before the linger expires and the added latency is ~zero. The pipeline chapter shows where the wait sits.
2. compression.type: zstd for the best ratio (needs the zstd feature and a C toolchain), lz4 for the cheapest CPU. Compression multiplies effective network and broker disk throughput — it is often the biggest win on 1 KiB+ payloads. Codec levels via compression.{zstd,gzip,lz4}.level; note compression.lz4.level needs the lz4-hc feature to have any effect (why).
3. batch.size (default 16 KiB): raise to 64–256 KiB when records are large or throughput is high. A batch is up to this size, never a wait for it — oversizing wastes only memory, undersizing caps batching. With 10 KiB records a 16 KiB batch holds one record; per-broker request coalescing is what saved that workload (benchmarks), but a right-sized batch is still cheaper.
4. buffer.memory (default 32 MiB): the total unsent-record budget. Size it ≈ target throughput × worst-case broker stall you want to absorb. When it fills, send() blocks up to max.block.ms — that back-pressure is a feature (failure modes); raising the buffer only delays it.

Two field lessons that outrank any key:

• Never flush() per record. flush waits for everything in flight — per-record it serializes the pipeline down to one round-trip at a time. Flush at commit points and shutdown (close flushes for you).
• Keep send() cheap. The send path is an in-memory append; anything slow around it (per-record env::var on macOS took a global libc lock and cost 28% of throughput in our own harness) dominates long before the client does.

Chasing latency

• linger.ms=0 (default) dispatches as soon as the pipeline has a slot.
• max.in.flight.requests.per.connection: depth 5 (default) maximizes pipeline utilization; on a single low-RTT broker it adds queue latency (p99 ~13 ms in our bench vs ~2 ms at depth 1 — same throughput). But depth also absorbs broker pauses: at depth 5 a GC/fsync stall on one request lets four others drain; at depth 1 everything waits (p99.9 ~10 ms vs ~100 ms under an injected pause). Rule of thumb: keep 5 unless you’ve measured your own tail on your own RTT.
• acks=1 buys a little latency at a real durability cost (leader-only confirmation). Prefer keeping acks=all and spending the latency budget on linger.ms=0 + right-sized batches.

Large records

max.request.size (1 MiB default) caps a single request; the broker’s message.max.bytes caps what it accepts — raise both together or the producer’s larger request just gets rejected. Oversized batches are split and requeued automatically on MESSAGE_TOO_LARGE; a single unsplittable record fails its delivery. If a payload doesn’t have to be a Kafka record (files, images), a blob store + reference usually beats raising the caps.

Keys and partitioning

• A record key is an ordering contract: same key → same partition → same order, byte-identically between kacrab and Java (partitioning). Choose keys for the ordering you need, and check cardinality — a hot key is a hot partition no client setting can fix.
• Null keys are fine — the sticky/adaptive partitioner batches them efficiently and steers around slow leaders (partitioner.adaptive.partitioning.enable, on by default). Don’t force RoundRobinPartitioner for “fairness”; per-record scatter defeats batching.
• partitioner.ignore.keys=true gives keyed records sticky treatment too — only when you don’t rely on key ordering.

Transactions

For atomic multi-partition writes (or consume-transform-produce):

• Set transactional.id — stable per logical producer instance, unique per instance. Two producers sharing an id fence each other by design (that’s the failover mechanism, not a bug).
• Keep transactions short; every open transaction holds coordinator state and delays read_committed consumers.
• The consumer side needs isolation.level=read_committed (consumer tuning).

Field notes

And the two habits: flush at boundaries, not per record; state a reason for every non-default line in the config.

Tuning the consumer

Producer tuning is mostly about efficiency; consumer tuning is mostly about semantics. Before touching a fetch knob, decide what a crash mid-batch should mean — everything else follows from that answer.

Start from delivery semantics

The committed offset is the group’s durable “read up to here” mark. When you commit relative to when you process decides your semantics:

• enable.auto.commit=true (default) commits positions on auto.commit.interval.ms and once more before each rebalance. It is periodic, not transactional: a crash between commits reprocesses up to one interval. Fine when handlers are idempotent; switch it off and commit manually when they aren’t.
• commit_async is safe to chain in kacrab: commits apply strictly in call order through a single worker queue (how, and why that was worth a bug fix), so a slow early commit can never overwrite a later one. Use it on the hot path, with one commit_sync at shutdown.

auto.offset.reset: choose, don’t inherit

This key answers two questions — where a new group starts, and what to do when a committed offset ages out of retention — so pick it deliberately:

• latest (default): new groups skip history; an aged-out position jumps to the live edge, silently skipping whatever retention deleted.
• earliest: new groups replay history; aged-out positions resume from the oldest surviving record — at-least-once friendly.
• none: turn the situation into an explicit NoOffsetForPartition error — right for pipelines where a silent skip or replay is worse than a page.

An out-of-range position is routine, not fatal: retention deletion and seeks race the log every day. kacrab resets exactly the affected partition and keeps the rest of the poll flowing (one bad partition doesn’t sink the poll). The practical guard is broker-side: keep retention comfortably longer than your worst-case consumer downtime, or an outage quietly becomes data loss under latest.

Group membership: the liveness contract

Two independent timers decide whether you’re alive; know which one fired:

• session.timeout.ms (45 s) vs heartbeat.interval.ms (3 s) — process liveness. Heartbeats run from a background task, so a busy poll loop does not miss them. Keep the ratio ≥3:1; the defaults are right.
• max.poll.interval.ms (5 min) — progress liveness: exceed it between polls and the group evicts you even though heartbeats flowed. If your handler can be slow, either raise this honestly or shrink the batch with max.poll.records (500 default) so each poll’s work fits the budget. Size it as max.poll.interval.ms > max.poll.records × worst-case per-record time, with margin.

The rebalance eviction→reassign→reprocess loop (“rebalance storm”) is almost always this arithmetic being false, not a broker problem.

Choosing the rebalance protocol

• partition.assignment.strategy defaults to Java’s range, cooperative-sticky — which negotiates to eager range in steady state. To get incremental rebalancing (nothing stops the world; only moved partitions pause), set it to cooperative-sticky alone. The two-round handoff and its never-double-owned invariant are the rebalancing chapter.
• group.protocol=consumer (KIP-848, Kafka 4.0+ brokers) moves assignment server-side: one heartbeat RPC, incremental reconciliation, faster convergence. kacrab supports it fully; prefer it on 4.x clusters for new groups.
• group.instance.id (static membership): set one stable id per pod/ instance and a restart within session.timeout.ms triggers no rebalance at all — the single cheapest fix for rolling-deploy churn.
• One deliberate non-feature: the consumer is single-owner (&mut self, not Sync) like Java’s. Scale with more consumers in the group, not by sharing one across threads.

Fetch tuning: batching in reverse

The fetch knobs mirror the producer’s linger/batch pair, in reverse:

• fetch.min.bytes (1) + fetch.max.wait.ms (500): raise fetch.min.bytes (e.g. 64 KiB) to make the broker batch for you — fewer, fuller responses, less client CPU — at up to fetch.max.wait.ms of added latency on quiet topics. kacrab clamps the broker wait to your poll timeout, so a short poll stays responsive regardless (fetching).
• max.partition.fetch.bytes (1 MiB) and fetch.max.bytes (50 MiB) are ceilings, not targets — and both yield to make progress: a first record larger than the limit is still returned. Raise the per-partition cap for large-record topics rather than raising max.poll.records.
• max.poll.records shapes poll-loop cadence, not network traffic — fetched surplus is buffered across polls and decoded lazily, batch by batch, with the next fetch prefetched in the background (the mechanics). Small values keep rebalance and shutdown latency low; they do not add RPCs.

Defaults here are genuinely good: the 4× consumer throughput edge in the benchmarks was measured at stock fetch settings.

Reading transactional data

Set isolation.level=read_committed to see only committed transactions (aborted records are filtered; reads stop at the last stable offset). The cost is added end-to-end latency equal to the producers’ commit cadence — another reason producers should keep transactions short.

Watch the right numbers

• Lag (current_lag, or kafka-consumer-groups --describe) is the consumer SLO. Alert on sustained growth, not absolute values.
• metrics() exposes poll/fetch/commit/rebalance counters; a rising rebalance count with stable membership is the max.poll.interval.ms arithmetic failing (above).
• pause/resume give per-partition back-pressure without leaving the group — buffered data survives a pause; heartbeats keep flowing.

Field notes

And the habit that prevents the classic outage: whenever you change handler logic, re-check that max.poll.records × worst-case handler time still fits inside max.poll.interval.ms.

Design decisions & Java parity

Every expedition carries a compass. A handful of principles were fixed before the first line of code and shape every file in kacrab — they explain why the code looks the way it does, why some things that could be simpler aren’t, and how each fork in the road in Parts I–V was decided.

“Java-compatible” means Kafka-protocol-compatible

The target is the behavior and wire output of the Java client, not a class-for-class port. Concretely:

• The config surface uses the same property names and defaults (acks, enable.idempotence, compression.type, sasl.*, ssl.*, …).
• The bytes on the wire are the Java client’s bytes — guaranteed for the things that must be byte-exact (murmur2, CRC32C, varint/zigzag, record-batch v2) by the oracle matrix.
• The algorithms are the real Java algorithms (the idempotent inflightBatchesBySequence / firstInFlightSequence / maybeResolveSequences machinery), not a simplified approximation.

What it is not: a translation of Java’s class hierarchy, threading model, or internal APIs. kacrab is idiomatic Rust underneath.

Outcome over mechanism

Where the runtime models genuinely differ, kacrab keeps the observable outcome identical and adapts the mechanism:

• Java orders enqueues with a single Sender thread; kacrab dispatches on concurrent Tokio tasks and reconstructs that order with the EnqueueSequencer.
• Java renumbers in-flight batches in place on an epoch bump because one thread owns them all; kacrab’s tasks can’t reach into a sibling’s batches, so a bump does a global epoch reset + re-stamp — different mechanism, identical bytes on the wire (a fresh epoch, sequences from zero).

The test is always: would the broker, or a Java consumer, be able to tell? If not, the Rust-idiomatic mechanism wins.

Generate and verify, don’t hand-write and hope

The wire types are generated from the upstream schemas and checked against the Java client; the security, compression, and multi-broker paths are verified against real brokers, not just self-consistent unit tests. The recurring theme — from the byte-level oracle to the docker-compose integration tests — is an external source of truth, because a system that only checks itself can be consistently wrong.

Safety and strictness, by default

• unsafe_code is forbidden workspace-wide.
• The lint set is strict: clippy pedantic + nursery + cargo denied, plus a curated list of restriction lints (expect_used, unwrap_used, indexing_slicing, arithmetic_side_effects, …) that must be justified with a reason when allowed.

“Pure Rust”, precisely

kacrab is a native-Rust implementation of the Kafka client — the protocol, wire framing, producer, idempotency, partitioning, and the pure-Rust codecs are Rust, with unsafe_code forbidden in kacrab’s own crates, and it does not wrap librdkafka. That is the claim worth making.

It is not a fully C-free dependency tree, and it is honest to say so:

The always-present C piece is the TLS crypto provider: rustls defaults to aws-lc-rs (AWS-LC, C + assembly). A genuinely C-free build would swap in a pure-Rust rustls crypto provider (e.g. rustls-rustcrypto, less battle-tested) and enable only the gzip/snappy/lz4 codecs — no zstd, lz4-hc, or gssapi.

The boundary kacrab won’t cross

JVM-only callback-handler and login-module classes cannot be loaded in a Rust process — that is a hard boundary, not a missing feature. Custom authentication uses the native Rust SASL authenticator hook (ProducerBuilder::sasl_client_authenticator) instead of a sasl.jaas.config class name.

Glossary

Terms coined or leaned on along the journey.

Producer side

• Base sequence — the sequence number stamped on the first record of a batch; per-partition it must ascend without gaps.
• Epoch — the producer epoch, bumped to fence stale idempotent state after an ambiguous failure.
• Ambiguous loss — a failure with no broker response, where the client cannot know if the write landed; resolved by an epoch bump.
• EnqueueSequencer — kacrab’s ticket-based ordering that reconstructs Java’s single-Sender-thread enqueue order across concurrent dispatch tasks.
• inflightBatchesBySequence — the per-partition map of sent-but-unacked batches, ordered by base sequence.
• firstInFlightSequence — the oldest unacked sequence on a partition; the retry gate.
• Sticky partitioning — batching-friendly null-key partitioning: stick to one partition until the batch fills, then rotate (adaptively weighted by queue depth).

Consumer side

• Position — a partition’s monotonic fetch cursor; advances only past records actually delivered to the caller.
• Committed offset — the group’s durable “read up to here” mark at the coordinator; where a member resumes after restart or rebalance.
• Coordinator — the broker owning a group’s slice of __consumer_offsets; target of joins, heartbeats, and commits. It can move; clients must re-discover it.
• Assignor — the algorithm dividing partitions among group members (range, roundrobin, sticky, cooperative-sticky).
• Cooperative rebalance (KIP-429) — the two-round incremental handoff that moves only the partitions that change hands, never double-owning one.
• Member epoch (KIP-848) — the server-side protocol’s generation counter, advanced by the coordinator on each assignment change.
• Fetch session (KIP-227) — per-broker state letting incremental fetches send only changed partitions instead of the full set.
• Static membership — a group.instance.id identity that survives restarts, so a quick bounce triggers no rebalance.

Wire & protocol

• Record batch v2 — the Kafka record-batch format (magic byte 2) that wraps compressed records with a CRC32C over the compressed payload.
• Topic id (KIP-516) — the UUID that keys newer Fetch versions in place of the topic name.
• Oracle matrix — the Java-client-checked test fixtures that prove Rust’s generated wire encoding matches Kafka byte-for-byte.
• KRaft — Kafka’s ZooKeeper-less consensus mode, used by every test cluster here.