Introduction

kaas-lib is a general-purpose Kafka 4.x client library for Rust, built directly on the kafka-protocol crate. Admin, produce and consume are all first-class, configuration is built by chaining with_* builders, and there is no librdkafka underneath.

It was written to support the kaas initiative — the kaas broker, kaas-ui, and the tooling around them. That is its origin rather than its boundary: the crates are published for any Rust program that talks to a Kafka 4.x cluster, and no public API assumes a kaas component on either end. The library began admin-first, with a read path shaped for browsing a topic; phase 2 added a real producer and both consumer-group protocols, which is what "general-purpose" was shorthand for. See the roadmap.

If you know Apache Kafka, you know what every RPC in here does. What this book is actually about is the layer around those RPCs: which broker each one has to reach, what its errors mean, which of them are allowed to run at all, and what happens when a broker returns something no released version of the schema describes. That last question is not hypothetical here — it is the normal case, and an entire chapter.

This is a client. kaas is a broker.

The two projects speak the same protocol from opposite ends and share no code. kaas decodes requests and encodes responses; kaas-lib does the mirror. Nothing in kaas's architecture carries over, and its codec is deliberately not a dependency here.

That separation is load-bearing rather than incidental. kaas-lib is the natural conformance harness for kaas — point testkit at a kaas broker instead of apache/kafka:4.3.1 and the acceptance suite becomes a typed parity check with real diffs. A shared codec would defeat that: two ends agreeing on a mutual misreading of the spec encode and decode consistently, pass green, and hide precisely the class of wire bug the harness exists to catch.

Three invariants and a constraint

Almost every design decision in this book follows from four statements. They are worth reading once here, because the rest of the book keeps referring back to them.

1. No upstream type reaches a public signature. kafka-protocol's generated types are #[non_exhaustive] and regenerate on every Kafka release. If one appears in our public API, every upstream bump becomes a breaking change for everyone downstream. Each crate defines owned domain types and converts at the boundary — including the quiet case, StrBytes, which is why domain types hold String and Bytes. See The domain boundary.

2. Nothing panics. A single malformed record on one topic must not take down a server hosting other clusters for other users. There is no unwrap, expect or panic! in library code, denied at the workspace root rather than repeated per crate, and the tolerant decoder turns a batch that will not parse into a value you can render rather than an error that ends a scan.

3. Partial failure is a result, not an error. Any call naming several resources returns one answer per resource — Vec<(ResourceId, Result<T, _>)>, never Result<Vec<T>, _>. Describing five hundred topics while three are mid-deletion returns 497 descriptions and three errors, because the alternative makes a UI unusable on exactly the clusters that need one.

And the constraint: the codec is a Kafka release behind the broker. kafka-protocol 0.17 ships Kafka 4.0 schemas; the acceptance suite runs against a 4.3.1 broker. So our ceiling binds more often than the broker's, brokers routinely advertise APIs and versions we cannot encode, and they return error codes the crate cannot name. This is the normal operating condition, not an edge case — which is why version negotiation, an Unknown(i16) arm on both the api-key and error-code enums, and honest gap documentation are structural rather than defensive.

The goal: which Kafka version a cluster runs is not your problem

That constraint has a payoff, and it is the clearest statement of what this library is for.

A caller should never have to ask what version a cluster runs. Not in a config file, not in a feature flag, not in a match on a version number. You ask for topics; you get topics. The library works out what the cluster can actually do and does the most it can with it.

Concretely, that means a caller never writes any of this:

struct Client; struct Config { kafka_version: String }
// None of this exists in kaas-lib's API, deliberately.
let client = Client::connect(cfg, KafkaVersion::V3_7)?;   // no
if cluster.version() >= (4, 0) { /* use the new API */ }   // no
config.set("api.version.request", "false");                // no

Five mechanisms carry that promise, and most of Part I is one of them seen up close:

The library also switches request shapes on the negotiated version without telling you: Fetch v13+ identifies topics by UUID and older versions by name; OffsetFetch moved its group field at v8. Both paths exist, and which one runs is not a decision a caller makes.

Where the abstraction stops, it says so

"As much as possible" is doing real work in that sentence, and pretending otherwise would be the more damaging choice. Some differences between Kafka versions are not absorbable, and for those the library's job is to be legible rather than silent:

• A sentinel that needs a schema this build cannot encode is a documented Unsupported, not a silently wrong answer — see ListOffsets -6.
• A group kind with no schema renders as Unrecognized carrying its type, not as an error and not as a fabricated description.
• An api the cluster genuinely lacks is Error::UnsupportedApi, carrying both version ranges so the reader can tell whether the cluster is old or this build is.

A version difference you can see and act on is a feature. One that is papered over into a wrong number is the bug this design exists to prevent.

What is here, and what is not

Six library crates, layered strictly: kafka-conn (the wire), kafka-meta (routing and cluster state), kafka-admin (31 admin RPCs), kafka-read (forward and backward scans), kafka-produce (the write path) and kafka-consume (fetch sessions and group membership), plus testkit for container fixtures.

They publish in lockstep at one version: this is one library split along a layering boundary, not six independently useful things.

What is deliberately not here is on Non-goals, and it is a shorter list than it used to be.

Rust, not a binding

There is no librdkafka in this dependency tree, and no cmake. That is the main reason to reach for this library over rdkafka, which wraps a mature C client and is the right answer when maturity matters more than toolchain.

One honest exception, because it would be easy to imply otherwise: two compression codecs reach C. kafka-protocol's lz4 and zstd features pull lz4-sys and zstd-sys, which cc builds from source, so a downstream build wants a C compiler for those. gzip (miniz_oxide) and snappy (snap) are pure Rust. Closing the gap would mean dropping codecs that real clusters use — zstd especially — or hand-rolling compression, and hand-rolling wire formats is exactly what this codebase does not do.

Everything is built by chaining

Optional settings are consuming with_* builders on owned types, so a configuration is one expression rather than a let mut and a sequence of assignments, and a half-built value never has to be passed anywhere. It is a single convention across every crate: if a type has an optional setting, it has a with_ method for it.

Contributions are very welcome

Especially on the above. The project is Apache-2.0 on GitHub, and a few things make it a friendlier codebase to contribute to than the size suggests:

• Every milestone has an acceptance command that must pass, so "is this done?" has an answer that is not a matter of opinion.
• No mocked brokers. Everything runs against apache/kafka:4.3.1 in a container via testkit, so a green test means it works against Kafka.
• The traps are written down. Part I explains why each decision is what it is, usually by naming the specific way of getting it wrong — which is the part that is hard to reconstruct from the source alone.

If you are picking something up, M12 (a single produced record, round-tripped) and M16 (fetch sessions) are the two that unblock the most downstream work. Open an issue first for anything milestone-sized, so the design conversation happens before the code does.

How to read this book

• Evaluating it? This page, then Non-goals for the fastest honest answer to "can I use this", then the API support matrix.
• Using it? Getting started, then Part IV — connecting, producing, consuming, admin, reading.
• Working on it? Part I in order from the system overview; Part III is the crate-by-crate tour of where it all lives.

Getting started

Adding the crates

Take only the layers you need — each one re-exports the types from the layers below it, so a producer never names kafka-conn directly — and pull them all at the same version. The crates publish to crates.io in lockstep because they are one library split along a layering boundary, so kafka-admin 0.4 against kafka-conn 0.3 is not a combination anyone tests.

[dependencies]
kafka-produce = "0.4"   # writing records
kafka-consume = "0.4"   # reading them back, with or without a group
kafka-admin   = "0.4"   # topics, configs, acls, groups
kafka-read    = "0.4"   # browse-shaped scans and tails, for a UI
tokio         = { version = "1", features = ["full"] }
futures       = "0.3"   # only for the scan stream

The workspace targets Rust 1.97 and edition 2024.

Connecting

Everything starts from a Cluster — the metadata cache, connection pool and retry policy behind one cheap-to-clone handle. Admin wraps one; the read path borrows one.

use kafka_admin::{Admin, ClusterConfig};

async fn example() -> kafka_admin::Result<()> {
let admin = Admin::connect(["localhost:9092"], ClusterConfig::default()).await?;
Ok(())
}

Bootstrap addresses are a list because they are re-resolved when every known broker goes unreachable — a rolling restart onto new addresses is a normal Kubernetes event, and a pool that only remembers the addresses from its last successful metadata fetch never recovers from one. See Metadata, routing and the pool.

Listing and describing topics

Note the shape of the return value: one result per topic, not one result for the batch. This is the third invariant from the introduction, and it is the single most visible thing about this API.

use kafka_admin::Admin;
async fn example(admin: &Admin) -> kafka_admin::Result<()> {
for (name, result) in admin.describe_topics(["orders", "shipments"]).await? {
    match result {
        Ok(topic) => println!("{name}: {} partitions", topic.partitions.len()),
        Err(error) => println!("{name}: {error}"),
    }
}
Ok(())
}

Creating a topic and changing a config

use kafka_admin::{Admin, ConfigChange, ConfigResource, NewTopic};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
for (name, result) in admin.create_topics([NewTopic::new("orders", 6, 3)]).await? {
    match result {
        Ok(created) => println!("{name}: {} partitions", created.partitions),
        Err(error) => println!("{name}: {error}"),
    }
}

admin
    .alter_configs([(
        ConfigResource::topic("orders"),
        vec![ConfigChange::set("retention.ms", "604800000")],
    )])
    .await?;
Ok(())
}

Producing a record

Producer::connect takes bootstrap addresses of its own; Producer::new wraps a Cluster you already have — from admin.cluster(), say. A producer is cheap to clone, and every clone shares one accumulator, so two clones writing to the same partition fill the same batch rather than two.

use kafka_produce::{ClusterConfig, Producer, ProducerConfig, ProducerRecord};

async fn example() -> kafka_produce::Result<()> {
let producer = Producer::connect(
    ["localhost:9092"],
    ClusterConfig::default(),
    ProducerConfig::new(),
)
.await?;

let meta = producer
    .send(
        ProducerRecord::new("orders")
            .with_key("customer-7")
            .with_value(r#"{"total":42}"#)
            .with_header("content-type", "application/json"),
    )
    .await?;

println!("landed at {}-{} offset {}", meta.topic, meta.partition, meta.offset);
Ok(())
}

That is one record, acknowledged by the full ISR, with the partition chosen by murmur2 over the key — the same hash a Java or C client uses, so a co-partitioned join still lines up. Records without a key go to a sticky partition instead of round-robin (KIP-480). Give ProducerRecord::with_partition an index to choose yourself.

Writing many records? Use enqueue, not a loop of send. send waits for the broker, so awaiting it in a loop keeps exactly one record in flight and batches nothing. enqueue returns as soon as the record is buffered and hands back a Delivery to await later:

use kafka_produce::{Producer, ProducerRecord};
async fn example(producer: &Producer) -> kafka_produce::Result<()> {
let mut pending = Vec::new();
for i in 0..10_000 {
    pending.push(
        producer
            .enqueue(ProducerRecord::new("orders").with_value(format!("{i}")))
            .await?,
    );
}

for delivery in pending {
    let meta = delivery.await?;   // where that record landed, or why it did not
}
Ok(())
}

Idempotence is on by default, which is what makes a re-send after a timeout safe rather than a duplicate. There is no acks=0, deliberately. Both are in Producing records, along with compression, transactions and the batching knobs.

Consuming records

A Consumer reads an explicit set of partitions: nothing rebalances it, nothing heartbeats, and no broker knows it exists as a member of anything. That is the mode to pin a reader to a partition, and it is the engine the group protocols sit on.

use kafka_consume::{Consumer, ConsumerConfig, Position};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer = Consumer::new(cluster, ConsumerConfig::new().group_id("reporting"));

consumer
    .assign(
        [("orders".to_owned(), 0), ("orders".to_owned(), 1)],
        Position::Earliest,
    )
    .await?;

loop {
    // Empty is a normal answer: a consumer at the log end is caught up, not
    // broken.
    for record in consumer.poll().await? {
        println!(
            "{}-{} @{}: {:?}",
            record.topic, record.partition, record.offset, record.value
        );
    }

    for ((topic, partition), result) in consumer.commit().await? {
        if let Err(error) = result {
            eprintln!("{topic}-{partition}: commit failed: {error}");
        }
    }
}
}

Setting group_id on a manually-assigned consumer borrows that group's offset storage — commit and committed read and write under it using the non-member sentinel. Borrowing the storage is not joining the group. Note the commit result: one entry per partition, the same shape as every other multi-resource call here.

Joining a group

GroupConsumer joins a KIP-848 group and lets the broker compute the assignment:

use kafka_consume::{ConsumerConfig, GroupConsumer};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer =
    GroupConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders"]).await?;

for _ in 0..100 {
    for record in consumer.poll().await? {
        println!("{}-{} @{}", record.topic, record.partition, record.offset);
    }
}

consumer.leave().await?;
Ok(())
}

poll is what heartbeats. It beats when one is due, reconciles any new assignment and then reads, so a member that stops polling — because it is doing slow work between batches — is a member the coordinator evicts. Nothing is owned until the first heartbeat comes back with an assignment, so the first poll or two returning empty is normal.

ClassicConsumer speaks the older JoinGroup/SyncGroup/Heartbeat protocol for pre-4.0 brokers and mixed groups. Which to use, offsets, the rebalance listener and the one hard constraint the classic path carries are all in Consuming records.

Browsing a topic

kafka-read answers a different question from a consumer: not "keep delivering this topic to me" but "show me a page of it". Two shapes, because a UI asks two things. tail answers "what just happened" and is the most-used view in any Kafka UI; scan answers "show me this topic from the beginning" and streams without ever materialising a Vec.

use futures::StreamExt;
use kafka_read::{ScanEvent, ScanSpec, StartPosition, TailSpec};

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
// The last 500 records per partition.
let tails = kafka_read::tail(cluster, &TailSpec::new("orders", 500)).await?;

// Or browse forwards from the start.
let mut stream = Box::pin(
    kafka_read::scan(cluster, ScanSpec::new("orders").from(StartPosition::Earliest)).await?,
);
while let Some(event) = stream.next().await {
    match event? {
        ScanEvent::Record(record) => println!("{}: {:?}", record.offset, record.value),
        ScanEvent::Progress(progress) => println!("{:?}", progress.fraction()),
        ScanEvent::Malformed { offset, reason, .. } => {
            println!("offset {offset} did not decode: {reason}")
        }
        _ => {}
    }
}
Ok(())
}

ScanEvent::Malformed is not an error path you can ignore into a _ arm and forget — it is how a batch that will not decode reaches your UI instead of ending the scan. Tolerant decoding explains why the granularity is a batch rather than a record.

Get a &Cluster from an Admin with admin.cluster(), or construct one directly with Cluster::connect.

Connecting safely to production

A client constructed read-only refuses every mutating API before opening a socket, enforced on the api key rather than on the method surface:

use kafka_admin::{Admin, ClusterConfig};

async fn example() -> kafka_admin::Result<()> {
let admin = Admin::connect_read_only(["localhost:9092"], ClusterConfig::default()).await?;
// admin.create_topics(..) now returns Error::ReadOnly without touching the network.
Ok(())
}

This is worth reaching for whenever a UI points at a cluster its operator would rather nobody mutated. The read-only gate covers what it does and does not protect.

TLS and SASL

use kafka_conn::{ConnectionConfig, SaslConfig, SaslMechanism, TlsConfig};
use kafka_meta::ClusterConfig;

fn example() -> kafka_conn::Result<()> {
let connection = ConnectionConfig::new()
    .with_client_id("cluster-ui")
    .with_tls(TlsConfig::default())
    .with_sasl(SaslConfig::new(
        SaslMechanism::ScramSha512,
        "ui-service",
        "hunter2",
    ));

let config = ClusterConfig {
    connection,
    ..ClusterConfig::default()
};
let _ = config;
Ok(())
}

Long-lived connections re-authenticate themselves; on any cluster that sets connections.max.reauth.ms — Confluent Cloud does — this is the difference between a session that survives the afternoon and periodic unexplained disconnects. See TLS, SASL and re-authentication.

Running the tests

cargo xtask ci                     # fmt + clippy + unit tests, no Docker
cargo xtask integration            # acceptance tests, boots real brokers
cargo xtask docs --serve           # this book, with live reload

The integration tests wear #[testkit::integration_test], which expands to #[tokio::test] + #[ignore = "needs Docker"] — so cargo test stays fast without a Docker daemon — and caps each test at two minutes of wall clock, container boot included. Each one boots apache/kafka:4.3.1 through testkit; nothing in this repository is verified against a mock broker.

System overview

Five crates in a strict stack. Each depends only on the ones below it, and each converts the layer below's vocabulary into its own before exposing anything. There are no cycles and no sideways dependencies — kafka-admin and kafka-read do not know about each other.

graph TD
    admin["kafka-admin<br/><i>37 admin RPCs, per-item results</i>"]
    read["kafka-read<br/><i>forward scan, backward tail</i>"]
    meta["kafka-meta<br/><i>metadata cache, routing, pool, retry</i>"]
    conn["kafka-conn<br/><i>framing, correlation, versions, TLS, SASL</i>"]
    codec["kafka-protocol<br/><i>the codec — schemas only</i>"]

    admin --> meta
    read --> meta
    meta --> conn
    conn --> codec

What each layer decides

kafka-conn owns one socket. Length-prefixed framing, a correlation map so many requests can be in flight at once, version negotiation per api key, TLS and SASL, and the read-only gate. It knows nothing about clusters — give it an address and it gives you request/response against that one broker. It also owns the two protocol vocabularies, ApiKey and ErrorCode, because every layer above needs to name them and a workspace with two of each would push conversions into every call site.

kafka-meta owns the cluster. Which brokers exist, which one leads each partition, which one coordinates each group, and which of them a given request is even allowed to go to. It holds the connection pool and the retry policy, so a caller above sends through Cluster and gets routing, reconnection and stale-metadata retries for free. Its two tables — routing and errors — are first-class artifacts in their own files.

kafka-admin and kafka-read own the domain. Both are pure translation: build a request from owned types, send it through Cluster, convert the response back into owned types. Neither opens a socket or picks a broker.

The one deliberate exception

Connection::send is generic over kafka_protocol::protocol::Request. This is the only place an upstream type appears in a public signature anywhere in the workspace, and it is deliberate: kafka-conn is the wire boundary, and a parallel request trait defined here would convert protocol types into protocol types for no gain.

Everything above it is held to the rule without exception. See The domain boundary.

How a request actually travels

Take admin.describe_topics(["orders"]):

sequenceDiagram
    participant C as caller
    participant A as kafka-admin
    participant M as kafka-meta
    participant N as kafka-conn
    participant B as broker

    C->>A: describe_topics(["orders"])
    A->>A: pick DescribeTopicPartitions or Metadata
    A->>M: send(request)
    M->>M: routing(api_key) → Any
    M->>M: pool: connection to some live broker
    M->>N: send(request)
    N->>N: negotiate version for this api key
    N->>B: framed request + correlation id
    B-->>N: framed response
    N->>N: match correlation id, decode
    N-->>M: typed response
    M->>M: retriable error? refresh metadata, retry
    M-->>A: typed response
    A->>A: convert into owned domain types
    A-->>C: Vec<(String, Result<TopicInfo, _>)>

Three decisions on that path are worth naming because each has its own chapter:

1. Which api version — never hardcoded, always the overlap of what the broker advertises and what this build can encode (Version negotiation).
2. Which broker — four routing classes, and sending to the wrong one produces a retry loop that looks like a flaky cluster rather than an error (Metadata, routing and the pool).
3. Whether the answer is retriable — classified along three independent axes, in one table (The error taxonomy).

Where the read path differs

kafka-read sends Fetch to a partition leader and then does something the admin path never does: it decodes record batches. That is the only place in the workspace where bytes from an untrusted producer are parsed, and it is why two chapters exist that have no admin equivalent — the read path for the scan shapes, and tolerant decoding for what happens when those bytes are wrong.

What is not here

The producer, group membership and fetch sessions that this section once listed as absent have all landed — they live in kafka-produce and kafka-consume. What remains true is narrower: kafka-read's scan API is one-shot by design and maintains no incremental fetch state, because a browse is not a subscription. kafka-consume is the crate that keeps a session.

See Non-goals for the decisions that are still decisions, and Roadmap for what shipped.

The domain boundary

No kafka_protocol::* type may appear in a public API signature.

This is the first invariant from the introduction, and the one with the least visible payoff and the highest cost of violating. It is worth being precise about what it buys.

Why

kafka-protocol's message types are generated from Apache Kafka's schema JSON, and they are all #[non_exhaustive]. Two consequences:

1. They regenerate on every Kafka release. Fields are added; enums gain variants. If TopicInfo were kafka_protocol::messages::MetadataResponseTopic, then every upstream bump would be a semver-breaking change for every consumer of kaas-lib, whether or not the field they use moved.
2. Consumers cannot construct or exhaustively match them. #[non_exhaustive] means downstream code cannot build one with a struct literal or write a match without a wildcard arm. Handing those types out makes callers inherit a constraint that exists for the codec's benefit, not theirs.

So each crate defines owned types and converts at its boundary. TopicInfo, GroupDescription, Record, ErrorCode, ApiKey — all ours.

The quiet violation

The obvious cases are easy to spot. The one that slips through is StrBytes, which reads like a bytes-crate type and is not: it is kafka_protocol::protocol::StrBytes. A domain struct holding one has violated the rule as thoroughly as one holding a MetadataResponseTopic, and it looks entirely reasonable in review.

The rule of thumb the workspace follows:

The one exception

Connection::send is generic over kafka_protocol::protocol::Request:

use kafka_conn::{Connection, Result};
async fn example(conn: &Connection) -> Result<()> {
use kafka_conn::protocol::Request;
// conn.send(request) — generic over the codec's own Request trait.
Ok(())
}

kafka-conn is the wire boundary. Defining a parallel Rpc trait here and requiring every request type to implement it would mean converting protocol types into protocol types for no gain — the crate's whole job at that point is to encode a kafka-protocol struct.

The exception stops there. kafka-meta, kafka-admin and kafka-read are held to the rule without exception, which is why kafka-admin has a 524-line types.rs doing nothing but owning the vocabulary.

kafka-conn also re-exports the codec as kafka_conn::protocol, so crates above it pin the dependency in exactly one manifest. Re-exporting is not licence to expose: those types may be used above, never returned.

What this costs, honestly

A lot of conversion code that does nothing clever. kafka-admin is 4,636 lines and a large fraction of it is field-by-field translation from a response struct into an owned one.

The alternative is worse in a specific way: a UI backend hosting many clusters cannot afford a client whose types change shape when the library tracks a new Kafka release. Absorbing the churn at one boundary inside this workspace is exactly the point.

Where the boundary is enforced

By review, not by a lint — there is no clippy rule for "this type came from that crate". The practical guards are:

• Owned types live in obvious places: crates/kafka-admin/src/types.rs, crates/kafka-meta/src/snapshot.rs, crates/kafka-read/src/record.rs.
• The conversion is always at the edge of a public function, never half-done: a public function returns owned types or it does not return.

Note what the rule does not say. Every crate above kafka-conn still depends on kafka-protocol directly and uses its types freely in private code — crates/kafka-read/src/batch.rs is built around RecordBatchDecoder, and every admin module constructs request structs. The constraint is on signatures, not on imports. A pub(crate) helper passing a MetadataResponseTopic around is fine; a pub fn returning one is not.

The connection actor

One socket, two tasks, one correlation map. Connection is the whole of kafka-conn's runtime surface: hand it a request, get a response, with many requests in flight at once.

graph LR
    caller1["caller A"] -->|encoded frame| tx
    caller2["caller B"] -->|encoded frame| tx
    tx["writer task"] -->|"framed bytes"| sock[("TCP / TLS")]
    sock -->|"framed bytes"| rx["reader task"]
    rx -->|"by correlation id"| pending{{"pending map<br/>i32 → oneshot"}}
    pending -->|response bytes| caller1
    pending -->|response bytes| caller2

Decoding happens on the calling task

The reader task does exactly one thing with a response frame: look up its correlation id and hand the raw Bytes to whoever is waiting. It does not decode.

That placement is deliberate. If the reader decoded, a single response that failed to parse — a version negotiated wrongly, a schema that drifted — would be an error in the reader task, and the reader task is shared by every request on that connection. Decoding on the calling task means a malformed DescribeConfigs response fails describe_configs and nothing else.

Framing

A 4-byte big-endian length prefix, then header and body. That is LengthDelimitedCodec's default configuration, and crates/kafka-conn/src/codec.rs states it explicitly anyway so that a future edit cannot quietly change endianness.

Frames are capped at DEFAULT_MAX_FRAME_BYTES — 100 MiB, matching Kafka's own socket.request.max.bytes default. A frame larger than that is a protocol desync rather than a big fetch, and reading it would be an unbounded allocation driven by the peer.

Two header traps

Both produce an off-by-a-few-bytes failure rather than a clear error, which is what makes them expensive:

The response header version is not the request's api version. It is a per-api, per-version mapping. The code asks ApiKey::response_header_version rather than deriving it, because deriving it is how you end up two bytes into the body wondering why a string length is nonsense.

ApiVersions responses always use response header v0, even once the connection is flexible. This is a real special case in the protocol — a chicken-and-egg escape hatch, because the client does not yet know what the broker speaks when it sends the first request. Get it wrong and your very first round trip fails. kafka-protocol encodes this in ApiVersionsResponse::header_version, which is another reason to go through the helper rather than compute it.

Pipelining

max_in_flight defaults to 5, matching Kafka's own default. The broker processes one connection's requests in order regardless, so this is about pipelining rather than parallelism — raising it trades head-of-line blocking for memory.

A permit is acquired before writing and released by a guard, so a dropped future cannot leak one. with_max_in_flight(0) is clamped to 1: zero permits is a deadlock, not a configuration, and there is a unit test asserting it.

This default becomes load-bearing the moment a producer exists. Five requests in flight is only safe with idempotence enabled; without it, a retried produce batch can land after a later one and silently reorder the log. Nothing in the workspace retries a write today, so it is currently harmless — see Roadmap, where wiring this to the idempotence setting is called out as a milestone requirement.

Cancel safety

Rule 5: dropping a send future must never leave the socket half-read. It cannot here, and the reason is structural rather than careful — the caller never touches the socket.

Drop the future and the oneshot receiver goes away. The request is still written by the writer task, the response is still read by the reader task, and the reader discards it on finding no waiter. The in-flight permit is released by its guard. The connection stays perfectly consistent; the only cost is one wasted round trip.

Cancel safety covers what this means for the layers above.

Death

When the socket dies, every pending caller resolves to Error::ConnectionClosed and every subsequent send fails immediately.

The alternative — futures that hang — is much worse than it sounds for this library's use case. A UI backend that leaks one hung future per dead broker degrades into a process that appears to be working while doing nothing, and the symptom shows up far from the cause.

Bootstrapping ApiVersions

The first request on a connection is a bootstrapping problem: you cannot know the broker's supported range until you have asked, and asking requires picking a version.

The connection sends at our max and treats error code 35 UNSUPPORTED_VERSION as data rather than as a handshake failure — the broker still returns its version table in that error response, so the correct reaction is to read it and retry at v0. Treating it as fatal is a client that cannot talk to any broker older than itself.

See Version negotiation for what happens with that table afterwards.

Per-connection counters

Every connection tracks bytes and requests sent and received (crates/kafka-conn/src/stats.rs). These exist from the beginning rather than being added when something needed them, because two acceptance criteria depend on them: the backward-scan test asserts that reading the last 500 records of a 100k-record partition fetches less than 5% of the partition, and that assertion is unverifiable without a byte counter.

Version negotiation

Never hardcode an API version. A hardcoded version works on your laptop and fails on the customer's cluster.

Every Kafka API is versioned independently, and brokers advertise a (min, max) range per api key in their ApiVersions response. The rule is to intersect that with what this build can encode and take the highest version in the overlap.

negotiated = min(broker.max, ours.max)   — provided the ranges overlap at all

Which side binds

In most Kafka clients the broker's ceiling is the interesting one: the client is current and some clusters are old. Here it is the other way round, and that inversion drives a lot of the design.

kafka-protocol 0.17 ships Kafka 4.0 schemas. The acceptance suite runs against a 4.3.1 broker. So our max is the binding side more often than not, and schemas older than the broker is the normal case, not an edge case.

Three consequences, each of which had to be designed for rather than discovered:

1. A broker will advertise versions we cannot encode. Clamp to ours.
2. A broker will advertise api keys we cannot name at all — StreamsGroupDescribe on any 4.1+ broker running Kafka Streams. The version table keeps those rows.
3. A broker will return error codes the crate does not name. Both our ApiKey and ErrorCode enums carry an Unknown(i16) arm.

The upstream schema gap lists exactly what is unreachable today because of this.

The table

ApiVersions is keyed by wire code rather than by ApiKey, precisely so that a key we have no name for still survives into the table:

use kafka_conn::{ApiKey, Connection};
async fn example(conn: &Connection) {
for entry in conn.versions().entries() {
    println!(
        "{} broker={:?} ours={:?} negotiated={:?}",
        entry.api_key,
        entry.broker,
        entry.ours,        // None when this build has no schema for the key
        entry.negotiated(),
    );
}
assert!(conn.versions().supports(ApiKey::Metadata));
}

ours is Option<VersionRange>, and None means kafka-protocol has no schema for that key. Returning None rather than guessing is the whole point — the gap stays visible instead of being papered over.

broker_ahead() reports the normal case against a newer broker, and is what the acceptance test asserts on to prove the clamp is happening on our side.

What "ahead" looks like in practice

Measured against a Kafka 4.2.0 broker, 8 of the 75 advertised keys are ahead of kafka-protocol 0.17 — every one of them by exactly one version. The numbers move with each broker release; the shape does not.

This is the steady state, not a defect to chase. Each of these negotiates to our max and works; five are share-group APIs still settling upstream. Only one costs a feature — ListOffsets v11 is what puts the -6 sentinel out of reach, in The upstream schema gap.

ours comes from ApiKey::valid_versions() and nothing else — this workspace declares no version support of its own, so every row above is a fact about the codec crate rather than a decision made here. Bumping the crate is the only thing that moves any of them.

Why negotiate_with exists

There are two ways to ask "what version can we send", and the difference between them is a real bug that was found live.

ApiVersions::negotiate uses our_range, which reads ApiKey::valid_versions(). That is derived per api key, and where a request and its response have different schema ranges it reports the wider of the two.

OffsetFetch is the live example: the response reaches v10, the request stops at v9. Negotiating from the api key alone picks v10, and the encoder then refuses to encode a v10 request that does not exist.

So Connection::send calls negotiate_with, passing the request and response types' own VERSIONS constants instead of the api key's:

Failure is typed, never a guess

When there is no overlap — or the broker never advertised the key at all — the result is Error::UnsupportedApi, carrying both ranges:

no usable version of StreamsGroupDescribe: broker offers Some((0, 1)), we speak None

Both halves matter for diagnosis. ours: None means we have no schema (bump the codec); a narrower ours than broker means the broker is ahead (also bump the codec); a narrower broker than ours means the cluster is old (nothing to do but degrade).

Falling back to "send it at v0 and hope" would turn a clear, actionable error into a decode failure several layers away.

Version-dependent request shapes

Negotiating the number is only half the job. Where a request's shape changes with the version, the code has to build a different request — and the codec rejects a field set outside its own version range rather than ignoring it, so "set both the old field and the new one" is an encode failure, not a compatibility trick.

The live examples in this workspace:

• Fetch v13+ identifies topics by Uuid instead of by name. Below v13 it is the name. Both paths exist in crates/kafka-read/src/fetch.rs.
• DescribeTopicPartitions exists at all only on newer brokers, so kafka-admin checks supports() and falls back to Metadata. That fallback has to handle two causes — "the broker is too old" and "our schemas are too old" — arriving at the same place.

Per connection, not per cluster

The table is negotiated on each connection during the handshake and stored on it. Brokers in one cluster can be mid-rolling-upgrade and genuinely disagree about what they support, so a cluster-wide table would be wrong during exactly the window when being right matters.

TLS, SASL and re-authentication

TLS

tokio-rustls with the ring provider rather than the default aws_lc_rs. That is a build-time decision, not a cryptographic one: aws-lc-sys needs cmake and a C toolchain, ring builds with cc alone, and CI on a minimal runner image has already lost time to exactly that.

TlsConfig covers system roots, a custom CA, client certificates for mTLS, and an SNI override for the case where the address you dial is not the name on the certificate — which is routine behind a Kubernetes service or a load balancer.

SASL mechanisms

PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512, negotiated via SaslHandshake followed by SaslAuthenticate.

PLAIN sends a recoverable password over the wire and the code knows it — SaslMechanism::sends_cleartext_password exists so the connection layer can reason about the combination of mechanism and transport rather than leaving it to the caller to notice.

SASLprep is not trim()

A password containing a non-ASCII space authenticates against a Java client and fails against ours if this is skipped.

SCRAM hashes the password, so both ends must normalise it identically before hashing. RFC 4013 (SASLprep, a stringprep profile) maps non-ASCII spaces — U+00A0, the U+2000 block — to U+0020, strips soft hyphens, and applies a set of prohibited-character and bidirectional rules.

Java clients normalise. A hand-rolled trim().to_lowercase() does not, and the failure mode is a password that works in every test you wrote and fails for one user whose password manager inserted a non-breaking space. The error says authentication failed and nothing about why.

This is why the stringprep crate is a dependency rather than a few lines of inline character handling.

The server signature is verified in constant time

SCRAM is mutual authentication: the server's final message carries a signature that proves it knows the stored key. Verifying it with == leaks a timing oracle on a value an attacker is actively trying to forge, so the comparison goes through subtle::ConstantTimeEq.

Kafka does not do channel binding, so the gs2 header is always n,,.

KIP-368 re-authentication

This is the one people skip, and the symptom looks like a network fault.

SaslAuthenticate's response carries a session lifetime. On any cluster where connections.max.reauth.ms is set — Confluent Cloud sets it — the broker kills the connection when that expires unless the client re-issues SaslAuthenticate on the live socket first.

A UI backend holds connections for hours. Without re-authentication you get periodic unexplained disconnects, spread across brokers, that read as flakiness in the network or the cluster and not as an auth problem at all.

The mechanism this needs is slightly awkward, and it is why the SASL exchange is written against a SaslTransport trait rather than directly against a socket. The same exchange has to run in two very different places:

1. On a bare framed stream, during connect, before the connection actor has started.
2. On a live, fully multiplexed connection, hours later, interleaved with other in-flight requests.

Both must behave identically. Abstracting the transport is what makes that true by construction rather than by keeping two code paths in agreement.

sequenceDiagram
    participant C as Connection
    participant B as broker

    Note over C,B: connect — bare framed stream
    C->>B: ApiVersions
    B-->>C: version table
    C->>B: SaslHandshake(SCRAM-SHA-512)
    B-->>C: ok
    C->>B: SaslAuthenticate(client-first)
    B-->>C: server-first
    C->>B: SaslAuthenticate(client-final)
    B-->>C: server-final + session_lifetime_ms
    Note over C: actor starts; normal traffic flows

    Note over C,B: hours later — live multiplexed connection
    C->>B: SaslAuthenticate(client-first)
    B-->>C: server-first
    C->>B: SaslAuthenticate(client-final)
    B-->>C: server-final + new lifetime
    Note over C: connection survives; callers never notice

The gate lets auth through

SaslHandshake and SaslAuthenticate are classified non-mutating by the read-only gate, which looks wrong at first glance since they plainly change state. They change connection state, not cluster state, and gating them would leave a read-only client unable to authenticate at all — which is to say, unable to read.

Verification

The acceptance test boots brokers configured for SASL_PLAINTEXT/PLAIN and SASL_SSL/SCRAM-SHA-512, asserts both authenticate, and asserts a wrong password yields Error::Authentication rather than a timeout. A third case runs a broker with connections.max.reauth.ms set to roughly 10 seconds and asserts a connection survives past twice that window while still serving requests — the only way to prove KIP-368 works is to let a session expire.

Metadata, routing and the pool

kafka-meta is the layer that knows what a cluster looks like. Everything above it sends through Cluster, which resolves the right broker, retries on the errors that mean "your view is stale", and keeps an immutable snapshot that readers can take without blocking.

The metadata snapshot

An ArcSwap over an immutable MetadataSnapshot, refreshed in the background and invalidated on demand. Reads never block and never wait on a refresh.

The snapshot carries its own fetch timestamp (snapshot.age()), which exists for the UI: rendering "as of 4 seconds ago" is honest, and rendering stale data as though it were live is not.

allow_auto_topic_creation is always false

MetadataRequest::default() sets allow_auto_topic_creation: true.

The schema default really is true, and kafka-protocol honours it. So following "construct with Default plus builders" literally — which is the right rule everywhere else — produces a UI that creates a topic every time someone typos a name into the search box, on any cluster with auto.create.topics.enable=true.

The metadata layer calls .with_allow_auto_topic_creation(false) unconditionally. There is no legitimate case for true in this codebase, and a unit test asserts it, because it is a one-word regression with a destructive blast radius.

The acceptance test goes further: it requests metadata for a nonexistent topic against a broker with auto-creation enabled, then uses a second client to assert the topic was not created.

The routing table

Not every RPC goes to any broker, and getting it wrong does not produce an error — it produces a NOT_CONTROLLER or NOT_COORDINATOR retry loop that presents as a flaky cluster. So this is a first-class table in its own file (crates/kafka-meta/src/routing.rs), next to the error table, rather than a decision scattered across call sites.

Two things this table encodes that the four-class summary glosses over:

Specific splits in two. A broker the caller names (DescribeLogDirs against a particular node) and a broker the snapshot names (a partition leader, for Fetch) are the same routing class but need completely different resolution. BrokerSelector carries the distinction.

Controller-only is stricter than KRaft requires. A KRaft broker will forward most of these. "Most" is doing a lot of work in that sentence, and the forwarding path has failure modes of its own, so the table routes them directly.

The wildcard arm is Any

And unlike the read-only gate, that is the safe default here. Mis-routing costs at worst a redirect and a retry; mis-classifying a mutating API costs the security property. Two wildcard arms, opposite defaults, for reasons specific to each.

The connection pool

One connection per broker, opened lazily, reconnected with capped jittered backoff.

Bootstrap re-resolution matters more than it looks. When every known broker is unreachable, the pool falls back to the bootstrap addresses. A cluster that rolls every broker onto new addresses is a normal Kubernetes event, and a pool that only knows the addresses from its last successful metadata fetch never recovers from one — it retries a set of dead endpoints forever.

Endpoint is therefore an enum, not a string: Node(i32) for a broker known by id whose address comes from metadata, Bootstrap(String) for the addresses we were given.

Connecting happens under a per-endpoint async mutex, not a global one. Two consequences, both deliberate: a slow handshake to a dead broker does not stall connections to healthy ones, and twenty concurrent callers wanting the same broker open one socket rather than twenty.

Retry and the two refresh axes

Cluster::send retries on the codes that mean the caller's view is stale, with capped jittered backoff. Which cache to invalidate is decided by the error taxonomy's two ownership axes:

• needs_metadata_refresh → drop the snapshot, refetch, retry. NOT_LEADER_OR_FOLLOWER, UNKNOWN_TOPIC_OR_PARTITION mid-reassignment.
• needs_coordinator_refresh → drop the cached coordinator for that group or transactional id, re-run FindCoordinator, retry. NOT_COORDINATOR, COORDINATOR_NOT_AVAILABLE.

These are independent. A code can need both, either, or neither, which is why they are two booleans rather than one enum.

Verification

The acceptance test runs against a 3-broker cluster with a 6-partition, RF=3 topic and asserts every partition resolves to a leader that is genuinely in its own replica set, and that leadership is spread across brokers rather than all resolving to whichever broker answered first.

The error taxonomy

Two types, in two files, answering two different questions.

Both are defined in kafka-conn and re-exported upward. That placement is forced: every crate in the workspace, including the connection layer itself, has to classify a broker's answer, and a workspace with two error types would push a From conversion into every call site in it.

Error — distinguishable at the type level

A UI renders these differently, and collapsing them into one string throws that away. A transport error means "the cluster is unreachable"; an authorization error means "ask your admin"; a decode failure means "this is our bug".

Decode is worth calling out separately. Every other variant describes something about the cluster or the caller; Decode describes a bug in this library or a schema that has drifted, and it should be reported rather than retried.

ErrorCode — derived, not transcribed

The table is derived from kafka_protocol::ResponseError, and that is the load-bearing part rather than an implementation detail.

• retriable() delegates to the crate's own is_retriable(), which encodes what the protocol says rather than what we remember it saying.
• from_response_error matches ResponseError exhaustively. ResponseError is a plain enum, so when an upstream bump adds a code, that match stops compiling. A new error code becomes a build failure to triage rather than a silent hole in the classification.

Hand-transcribing a 100+ entry table would be correct exactly once.

Three independent axes

kafka-protocol models one of these. The other two are ours, and they are exhaustive matches over our own enum for the same compile-failure reason.

They are genuinely independent — a code can be retriable without needing any refresh (REQUEST_TIMED_OUT), need a metadata refresh (NOT_LEADER_OR_FOLLOWER), or need a coordinator refresh (NOT_COORDINATOR). Modelling them as one enum would force a false choice on the codes that need two. Metadata, routing and the pool is what acts on them.

Unknown(i16) is not optional

kafka-protocol 0.17 knows error codes through Kafka 4.1. The acceptance suite runs against 4.3.1. A broker can and will return a code with no name here, and that is the expected case rather than a corruption signal.

ErrorCode::Unknown(i16) round-trips and renders. It never panics, and it never collapses into a generic failure that discards the number — the number is the only thing anyone can search for.

The unit test is table-driven over every code plus one that no Kafka release defines (30000), asserting it lands in Unknown(30000) and still renders.

Where classification happens

At the point the response is decoded, not at the call site. A response carrying an error code becomes Error::Broker { code, message } — or Error::Authorization where the code is an authorization failure, since that distinction is what a UI needs and recovering it later means matching on the code again.

Per-item results keep their own errors: describe_topics over 500 names returns 500 entries, each independently Ok or Err, and one UNKNOWN_TOPIC_OR_PARTITION does not become the result of the call. See the third invariant.

The read-only gate

A client constructed read-only returns Error::ReadOnly for every mutating RPC before touching the network.

use kafka_admin::{Admin, ClusterConfig};

async fn example() -> kafka_admin::Result<()> {
let admin = Admin::connect_read_only(["localhost:9092"], ClusterConfig::default()).await?;
Ok(())
}

Under the hood this is one flag on ConnectionConfig, which also means you can build it yourself with ConnectionConfig::new().read_only() and hand it to any layer.

Enforced on the api key, not the method surface

This is the whole design, and it is the difference between a property and a convention.

The gate lives in Connection::send and matches on ApiKey — ApiKey::is_mutating. It does not live on kafka-admin's method surface.

The consequence: an admin method added tomorrow is covered without anyone remembering to cover it. ApiKey matches the protocol's closed set, so a new method necessarily sends an existing api key, and there is no way to reach the socket without passing the check. Enforcing on our own method surface would make the property depend on every future contributor remembering an annotation.

graph LR
    m["any admin/read method"] --> s["Connection::send"]
    s --> g{"read_only<br/>&& is_mutating(key)?"}
    g -->|yes| e["Error::ReadOnly<br/><i>no socket touched</i>"]
    g -->|no| w["encode + write"]

The wildcard arm is true

is_mutating is written as an allowlist of read-only keys with a _ => true fallback. The direction is the entire security property.

enum ApiKey { Fetch, Metadata }
const fn example(key: ApiKey) -> bool {
match key {
    ApiKey::Fetch | ApiKey::Metadata /* … 26 more … */ => false,
    // Deny by default. Do not replace this with `_ => false`.
    _ => true,
}
}

ApiKey is #[non_exhaustive] and carries an Unknown(i16) variant, so the wildcard is mandatory. Making it false would silently un-gate every API added by a future Kafka release and every key we have never heard of — which, given that the codec is a release behind the broker, is not a hypothetical set.

Note that this is the opposite default from the routing table's wildcard, which is Routing::Any. Mis-routing costs a redirect; mis-classifying a mutating API costs the property. Each wildcard is chosen for its own failure cost.

Two entries that look wrong

FindCoordinator is classified read-only even though on some clusters it can trigger creation of the internal __consumer_offsets topic. A read-only client cannot fetch a committed offset without it, and the alternative — parsing __consumer_offsets ourselves — is explicitly forbidden. This is a considered trade, not an oversight.

SaslHandshake and SaslAuthenticate are classified read-only. They mutate connection state rather than cluster state, and gating them would make a read-only client unable to authenticate, which is to say unable to read.

Non-obvious mutators

Several keys that read like queries are classified mutating, correctly:

• OffsetCommit, OffsetDelete — they write group state
• InitProducerId, AddPartitionsToTxn — they allocate and fence producer state
• Produce — obviously. It was classified and gated before anything in the workspace could send one, which is the point of gating on ApiKey rather than over a method surface: kafka-produce arrived already covered, and its acceptance suite asserts a read-only client refuses to produce without the gate having heard of the crate

What it does and does not protect

It does stop a UI backend from mutating a cluster its operator would rather nobody mutated, including through a bug or an unintended code path, and it does so without a round trip.

It does not replace broker-side authorization. It is a client-side safety catch, enforced in a process the operator may not control. A cluster that must not be mutated by a principal should say so with ACLs; this gate protects against the UI doing something nobody asked it to, not against a hostile client.

Verification

The acceptance test drives its assertion from ApiKey::iter rather than from a hand-written list, so every key the protocol defines is checked and new protocol keys are covered automatically as the codec learns them. That is the same reasoning as the wildcard arm, applied to the test.

The read path

Two shapes, because a UI asks two different questions. Neither is a consumer: there is no rebalance, no commit, no membership.

Forward scan

scan returns a Stream, never a Vec.

A UI browsing a partition with a hundred million records must not decide how much memory to use based on how much data the user happened to ask for, and a scan that materialises its results has lost that argument before the first record is decoded.

use futures::StreamExt;
use kafka_read::{ScanEvent, ScanSpec, StartPosition};

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let spec = ScanSpec::new("orders")
    .from(StartPosition::Earliest)
    .partitions([0, 1, 2])
    .limit(10_000);

let mut stream = Box::pin(kafka_read::scan(cluster, spec).await?);
while let Some(event) = stream.next().await {
    match event? {
        ScanEvent::Record(record) => { /* … */ }
        ScanEvent::Progress(progress) => { /* fraction() drives a progress bar */ }
        ScanEvent::Malformed { offset, .. } => { /* render, do not abort */ }
        _ => {}
    }
}
Ok(())
}

StartPosition covers Earliest, Latest, an explicit Offset(i64), and a wall-clock Timestamp resolved through ListOffsets.

Memory is bounded regardless of partition count

max_buffered_records caps the buffer across the whole scan, not per partition.

The distinction is the entire point. Interleaving records across partitions needs lookahead, and the naive implementation keeps one fetch's worth per partition — which on a thousand-partition topic is a thousand times the budget anyone intended, and it is discovered in production rather than in testing because nobody scans a thousand-partition topic on a laptop.

Ordering, stated honestly

Within a partition: exact log order, always.

Across partitions: timestamp order whenever the buffer holds at least one record from every partition still being read, which is the usual case. When the cap forces an emit before every partition is represented, ordering degrades gracefully — the record emitted is the earliest among those buffered, so the reorder is bounded by the span of the buffer rather than by the length of the topic.

ScanEvent::Progress reports when that happens, so a UI can say "approximately ordered" rather than quietly lying. Degradation you can observe is a different thing from degradation you cannot.

Backward scan

"Last N messages" is the most-used view in any Kafka UI, and it is not a forward read with a different starting point.

Reading forward from latest - N is wrong on any topic where records are not one offset apart — which is every compacted topic and every topic that has had DeleteRecords run against it.

ListOffsets(LATEST) per partition, then walk backwards in bounded chunks: read [end - step, end), keep what came out, and if it was not enough, move end back and go again. Each chunk is an ordinary forward fetch — Kafka has no backward read, so the walk lives in the planning, not in the protocol.

use kafka_read::TailSpec;

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let tails = kafka_read::tail(cluster, &TailSpec::new("orders", 500)).await?;
Ok(())
}

Three ways to get it wrong

Batch boundaries do not align to the step. A fetch from end - step begins at whatever batch contains that offset, so a chunk routinely returns records from before the window. They are filtered out, not trusted.

Compacted topics have offset gaps. Ask for the last 500 records of a partition whose offsets run 0, 7, 91, 4001 and offset arithmetic over-estimates every time. A step that assumes one record per offset walks back a handful of records per round trip and ends up re-reading the whole partition — precisely the naive behaviour this design exists to avoid. So the step grows when a chunk yields fewer records than its offset span suggested.

The loop must terminate. It stops at the partition's log start, and because the step only ever grows, a partition with a thousand-fold offset gap converges rather than crawling.

The acceptance test is built to catch exactly this: a partition with 100k records and randomised batch sizes, request the last 500, and assert both that the right 500 come back and that fewer than 5% of the partition's bytes were fetched — measured with the connection byte counters.

Fetch is deliberately session-less

crates/kafka-read/src/fetch.rs pins session_id = 0 and session_epoch = -1 — Java's FetchMetadata.LEGACY sentinel, meaning no incremental fetch session at all.

That is correct for a UI. A scan is one-shot, and an incremental session would make each scan depend on the last, which is a coupling with no benefit when the next request may be for a different topic entirely.

It is also exactly wrong for a consumer, and reshaping it is a roadmap prerequisite for group membership.

Filtering and visibility

RecordFilter runs client-side — Kafka has no server-side filtering — and Visibility chooses the isolation level:

Note that read_committed does not mean the broker filters for you — it sends the records and an AbortedTransactions list, and the client does the filtering. See Tolerant decoding.

Tolerant decoding

The module the whole read path is built around. One batch that will not decode does not fail the scan — it becomes a value the UI can render.

use bytes::Bytes;
struct Record;
struct DecodeError;
enum RecordOutcome {
    Ok(Record),
    Malformed {
        offset: i64,              // from the batch header, readable even when the records are not
        last_offset: Option<i64>, // when the header was intact enough to say
        raw: Bytes,               // so it can be dumped, hexed or reported
        reason: DecodeError,
    },
}

A UI that says "offsets 4,102–4,530 would not decode, here are the raw bytes" is useful. A UI that says "this partition failed" is not, and a library that returns the second when it could return the first has discarded information the operator needs.

Three things that look like corruption and are not

A decoder that reports these is worse than no decoder at all, because it cries wolf on every single fetch.

1. A truncated trailing batch. max_bytes cuts a fetch mid-batch by design. Every fetch ends this way whenever there is more data than the budget — which is to say, most fetches. Flagging it means claiming corruption at the end of every fetch on every healthy cluster.

This is why crates/kafka-read/src/batch.rs reads a handful of fixed offsets out of the v2 batch header directly. It is not schema duplication: it is the minimum needed to decide whether a batch is complete before handing it to a decoder that would otherwise report a truncation as corruption.

2. Control batches. Attribute bit 5 marks a transaction marker — a commit or abort record the broker itself writes into the log. It is not user data and has no key or value worth showing. Skip it.

3. Aborted transaction records. Under read_committed the broker sends them anyway and hands over an AbortedTransactions list for the client to filter with. Not filtering means showing records that were explicitly rolled back, which is a correctness bug that looks like working software.

Everything else that fails to decode becomes Malformed and the scan continues.

Granularity is a batch, decided deliberately

RecordBatchDecoder decodes a whole batch into a Vec<Record> and errors at batch granularity. Per-record tolerance is not something the crate's API offers.

The two options were to vendor the record loop or to accept batch-level Malformed. The workspace accepts batch-level, and the cost is stated plainly: a corrupt record takes its batch with it, bounded by max.message.bytes.

What matters is that this was settled up front rather than discovered while writing the scan. The design's actual claim — one bad batch does not fail the scan, and the failure carries enough information to be actionable — holds either way.

Decompression is size-bounded

Gzip's maximum expansion ratio is about 1032:1. A producer that can write a 1 MiB batch — Kafka's default max.message.bytes — can make a client allocate a gigabyte.

For a UI backend serving many clusters, that is a denial of service against every other cluster in the process, and it costs the attacker almost nothing. So decompression is bounded, using RecordBatchDecoder::decode_with_custom_compression as the hook.

The framed case keeps an input cap rather than an output one because kafka-protocol allocates per block from each block's own declared length, with no hook in between. That cap is a real bound rather than a hopeful one: snappy's expansion is limited by its format — a copy operation emits at most 64 bytes, and xerial chunks decompress to at most 32 KiB each.

Kafka's snappy is two formats, and the crate cannot tell them apart

Snappy on the wire is not one thing. The Java client frames it with snappy-java's xerial header; librdkafka — and with it most of the non-Java ecosystem — writes raw, unframed snappy. A reader has to accept both, which is why kafka-protocol autodetects.

Its autodetection is broken in 0.17.0. It reads the 16-byte magic header with try_get_bytes(16), and that call advances the buffer. When the header does not match — the raw case — the fallback then runs on a buffer whose first sixteen bytes are already gone, and fails with failed to decompress raw snappy bytes. Upstream's own fallback test passes only because its fixture is fifteen bytes long, one short of the header, so the read returns Err and consumes nothing.

The consequence for a UI is not subtle: no snappy topic written by a non-Java producer can be read at all.

So crates/kafka-read/src/decompress.rs decides the framing itself, while the buffer is still whole, and delegates only the xerial case. This is not the reimplementation the module otherwise refuses to attempt — the raw branch is a single snap call with no framing logic in it, and it is the branch that gets the better bound of the two.

This is the one place the workspace knowingly diverges from the codec crate. Revisit it when kafka-protocol fixes the detection upstream.

A record count is not a promise

RecordBatchDecoder reserves the whole Vec<Record> from the batch header's recordsCount before parsing a single record, and the only check it applies to that number is that it is not negative. The count is attacker-controlled bytes: a 99-byte batch declaring 285 million records asks for a multi-gigabyte allocation.

Decompression bounds do not help, because the reservation happens on the header, not on the payload. So the count is checked against what the batch could physically hold — its own payload when uncompressed, the decompression ceiling when compressed — and an impossible count becomes Malformed like any other unreadable batch. The allocation is then proportional to bytes already accepted rather than to a number the sender chose freely.

The divisor is six bytes per record, deliberately one below the true seven-byte floor for a v2 record, so no batch a real producer writes is ever rejected.

Verified by fuzzing

Rule 2 says a malformed record must not kill the process. The executable form of that claim is a cargo-fuzz target over RecordBatch bytes whose pass condition is simply no panic:

cargo xtask fuzz

It needs a nightly toolchain, so it has its own CI job rather than pinning the whole workspace to nightly.

The unbounded record count above is what it found on its first genuinely green run, and the shape of that finding is worth keeping in mind: libFuzzer reported it as an out-of-memory, not a panic. "No panic" is the pass condition, but a decoder can violate rule 2 without ever panicking — killing a process by allocation rather than by abort. Both count.

The unit tests hand-craft a batch with a corrupt record and assert Malformed is yielded and the scan continues; separately, they fetch with max_bytes small enough to truncate a batch and assert zero Malformed events — the truncation must be invisible.

Cancel safety

Every public async fn must be cancel-safe. Dropping the future releases buffers and either completes-and-discards the in-flight request or closes the connection. Never leave a connection with a half-read response.

Why this rule exists here specifically

A UI backend cancels futures constantly, and not because of anything exceptional:

• A user closes a tab mid-scan.
• An HTTP request times out while three admin RPCs are in flight.
• tokio::select! races a fetch against a shutdown signal.
• A Stream is dropped after the caller has seen enough records.

If any of those can leave a connection with a partially-consumed response frame, then the next request on that connection decodes the tail of the previous one. The symptom is a decode error on an unrelated call, minutes later, on a different topic — a bug that is very hard to trace back to the tab that was closed.

How the connection makes it structural

The property is not maintained by careful cleanup code. It falls out of the connection actor's shape: the caller never touches the socket.

sequenceDiagram
    participant C as caller
    participant W as writer task
    participant R as reader task
    participant B as broker

    C->>W: encoded frame + oneshot sender
    W->>B: framed request
    Note over C: future dropped here
    C--xC: oneshot receiver dropped
    B-->>R: framed response
    R->>R: look up correlation id → no waiter
    R->>R: discard, release permit
    Note over R: connection remains consistent

Drop the future and the oneshot receiver goes away. The request is still written; the response is still read in full; the reader finds no waiter and discards it. The in-flight permit is released by its guard rather than by an explicit path that a ? could skip.

The only cost is one wasted round trip. There is no state to unwind, because there was never any caller-owned socket state to begin with.

What this means for the layers above

Because kafka-conn is cancel-safe by construction, the layers above inherit it for free as long as they hold one rule: do not hold a resource across an await that a drop would need to clean up.

In practice:

• kafka-meta — the pool's per-endpoint connect mutex is held across the handshake. Dropping there releases the mutex and leaves no half-open connection, because the connection is only published to the pool once fully established.
• kafka-admin — every method is a send-and-convert. Dropping mid-call discards a response.
• kafka-read — the interesting one. A scan holds a decode buffer and in-flight fetches. Dropping the stream drops the buffer and the fetch futures, each of which is individually cancel-safe. Nothing is left behind and no connection is disturbed.

Cancel-safe is not the same as atomic

Worth stating plainly, because the rule is easy to over-read.

Dropping the future for a mutating call does not cancel the operation on the broker. create_topics whose future is dropped after the request was written will very likely still create the topic — the request was sent, and Kafka has no cancellation. What the rule guarantees is that the connection stays usable, not that the cluster is unchanged.

For a read-heavy UI this distinction rarely bites. It matters for anything mutating, where "the future was dropped" must not be read as "it did not happen".

Verification

The leak test spawns 1,000 scans, cancels them at random points, and asserts that connection count and RSS both return to baseline. A cancel path that leaked a connection, a permit or a buffer would show up as drift in one of those two numbers rather than as a test that fails cleanly, which is why the assertion is on the process rather than on a return value.

API support matrix

37 of the protocol's 87 api keys, listed in wire-code order. Every row is negotiated per connection — the version column is deliberately absent, because there is no fixed version to report. See Version negotiation.

Reading the table

Routing is the class from crates/kafka-meta/src/routing.rs. Sending to the wrong class does not produce an error — it produces a NOT_CONTROLLER or NOT_COORDINATOR retry loop that presents as a flaky cluster. See Metadata, routing and the pool.

Mutating is ApiKey::is_mutating, which is what a read-only client refuses before opening a socket. Note the entries that look surprising in both directions: OffsetCommit and OffsetDelete are mutating because they write group state, while SaslHandshake, SaslAuthenticate and FindCoordinator are not, for reasons the read-only gate chapter explains.

DeleteRecords, CreateAcls, IncrementalAlterConfigs and the quota and SCRAM alters route to any broker rather than to the controller, which is correct — these are not controller-only APIs even though they mutate.

The other 50

Not implemented, and they fall into four groups.

Consumer-group membership and the transactional write path — the real gap. Produce (0) is implemented; what remains is InitProducerId (18) and the transaction write path (AddPartitionsToTxn 20, AddOffsetsToTxn 21, EndTxn 22, TxnOffsetCommit 24), classic membership (JoinGroup 7, Heartbeat 8, LeaveGroup 9, SyncGroup 10), KIP-848 membership (ConsumerGroupHeartbeat 64), and share consumption (ShareGroupHeartbeat 72, ShareFetch 74, ShareAcknowledge 75).

This is the scope boundary, not an oversight — the library writes records, but observes groups and transactions without joining or starting one. See Non-goals and Roadmap.

Broker-internal and KRaft APIs, which a client has no business sending: Vote (48), BeginQuorumEpoch (49), EndQuorumEpoch (50), AlterPartition (52), Envelope (54), FetchSnapshot (55), BrokerRegistration (58), BrokerHeartbeat (59), UnregisterBroker (60), AllocateProducerIds (63), ControllerRegistration (66), AssignReplicasToDirs (69), the Raft voter APIs (76–78), and the share-group state APIs (79–83) that live between broker and coordinator. DescribeQuorum (51) is the one in this group a UI might plausibly want.

Superseded. AlterConfigs (29) is deliberately absent — IncrementalAlterConfigs (40) replaces it, and using the older API silently resets every config you did not mention. There is no reason to offer it.

Not yet needed. Delegation tokens (34–37), OffsetForLeaderEpoch (19), AlterReplicaLogDirs (30), UpdateFeatures (53), the telemetry APIs (67–68), ListConfigResources (70), and the share-group offset admin APIs (84–86). These are ordinary gaps — nothing structural stops them being added.

Keeping this page honest

The table is generated from two sources of truth and must be regenerated when either changes:

• crates/kafka-conn/src/api_key.rs — the wire codes and is_mutating
• crates/kafka-meta/src/routing.rs — the routing class

The docs CI job asserts that every crates/… path cited anywhere in this book exists, so a refactor that moves a file fails the build rather than leaving a citation that is confidently wrong. It does not yet verify the rows of this table against the source; that check is worth adding the first time this page goes stale.

The upstream schema gap

kafka-protocol 0.17.0 ships Kafka 4.0 message schemas. The acceptance suite runs against a 4.3.1 broker.

That gap is permanent in shape even though its contents change: a codec crate tracks a Kafka release, and brokers keep shipping. So schemas older than the broker is the normal operating condition, not an edge case, and this page lists exactly what it costs today.

Version history for context: 0.16.0 added Kafka 4.0 support and error codes through 4.1.0; 0.17.0 changed only snappy framing and ApiKey::iter. Check the crate's CHANGELOG.md for which release the schemas track before assuming anything here is still current.

1. Streams groups cannot be described at all

StreamsGroupDescribe and StreamsGroupHeartbeat (KIP-1071) have no schema in the crate. They occupy wire codes a 4.1+ broker advertises and kafka-protocol 0.17 leaves unassigned.

The codes are 88 (StreamsGroupHeartbeat) and 89 (StreamsGroupDescribe), and the shape of the gap is worth naming exactly, because the obvious mental model is wrong. This is a hole, not a truncation: the crate's ApiKey enum runs ReadShareGroupStateSummary = 87 straight to DescribeShareGroupOffsets = 90, so it knows three keys above the two it is missing. "Our schemas stop at key N" does not describe it, and a codec bump that raises the ceiling elsewhere is not automatically the thing that fills this. Upstream marked the streams-group APIs stable in KAFKA-19869, so a crate regenerated after that should pick them up.

Two keys reported as Unknown in the middle of a version table is therefore expected output, not a symptom. So is ours: None on those rows.

The consequence is concrete and it will hit any real cluster: a 4.1+ broker running Kafka Streams reports groupType=streams in ListGroups, and we cannot describe those groups.

Degrade, do not fail. The domain enum carries an Unrecognized { group_id, group_type } variant, so a streams group renders as a known-but-undescribable group rather than taking down the group list. A UI that hard-fails on an undescribable group is a UI that hard-fails on most real clusters. See The four group kinds.

There is a unit test asserting that wire code 89 — unassigned in the crate, where a 4.1+ broker advertises StreamsGroupDescribe — survives in the version table as ApiKey::Unknown(89) with ours: None.

2. ListOffsets sentinel -6 is unreachable

ListOffsets has six sentinels, not two:

-6 requires ListOffsets v11. The crate caps at v10; a 4.3.1 broker serves v11. So it is unreachable until upstream bumps — not by omission, but because there is no v11 request to encode.

The domain enum documents the gap rather than silently omitting the variant, because "this client does not support tiered-storage upload watermarks" and "this sentinel does not exist" are different answers.

The other five are surfaced distinctly. Collapsing them is how a UI reports wrong retention on a tiered cluster: EARLIEST and EARLIEST_LOCAL_TIMESTAMP differ by exactly the data that has been offloaded to remote storage, which on a tiered cluster is most of it.

3. Error codes stop at 4.1

kafka-protocol knows codes through Kafka 4.1. A 4.3 broker can return codes it cannot name.

ResponseError already models this with its own unknown handling, and our owned enum carries ErrorCode::Unknown(i16) for the same reason. An unrecognised code round-trips and renders; it never panics and never collapses into a generic failure that discards the number — the number is the only thing anyone can search for. See The error taxonomy.

4. OffsetFetch request and response ranges disagree

Not a broker gap but a codec one, and it caused a real bug.

ApiKey::valid_versions() is derived per api key, and where a request and its response have different schema ranges it reports the wider one. For OffsetFetch the response reaches v10 while the request stops at v9, so negotiating from the api key alone picks v10 and the encoder then refuses.

The fix is ApiVersions::negotiate_with, which takes the specific request and response types' own VERSIONS rather than the api key's. See Version negotiation.

InitProducerId is the second instance — request v5, response v6 — and it is the one that shows the symptom where it is most likely to be mistaken for a bug. A version table renders ours = 0..6 against a 4.2 broker's 0..5: our declared ceiling sits above the broker's, on a key where the encoder would in fact refuse v6. Nothing ever sends at v6, because Connection::send goes through negotiate_with; the discrepancy is confined to the reported range. Expect it on any key whose response schema outruns its request.

5. Raw snappy does not decode

Also a codec bug rather than a broker gap, and the most consequential one on this page — it is the only entry that makes real data unreadable rather than merely unreachable.

Kafka's snappy is two formats. The Java client frames it with snappy-java's xerial header; librdkafka, and so most of the non-Java ecosystem, writes raw unframed snappy. kafka-protocol autodetects between them and gets it wrong: it reads the 16-byte magic header with try_get_bytes(16), which advances the buffer, so the raw fallback runs on a buffer already missing its first sixteen bytes. Upstream's own fallback test passes only because its fixture is fifteen bytes, one short of the header.

Note the version history above — 0.17.0's one substantive change was to the snappy framing. This is the newest code in the dependency, and it is the code we were least entitled to assume.

Untreated, this means no snappy topic written by a non-Java producer can be read. So this is the single place the workspace takes option 3 below in its strongest form: kafka-read decides the framing itself and delegates only the xerial case, which is a knowing divergence from the codec crate rather than a gap it merely documents. See Tolerant decoding.

Revisit when upstream fixes the detection. The workaround is small and deliberately shaped to be deleted.

What to do when this blocks you

Say so. An honest blocker is a legitimate outcome; a hand-rolled schema is not.

Adding a message definition by hand to work around a missing one means maintaining a private fork of a generated artifact, diverging from upstream silently, and losing the compile-error signal that makes the error taxonomy trustworthy. The correct responses, in order of preference:

1. Wait for the upstream bump and note the gap here.
2. Contribute the schema upstream to kafka-protocol.
3. Degrade visibly — an Unrecognized variant, an Unknown(i16), a documented None — so the gap is observable rather than invisible.

Every gap on this page took option 3 while waiting on option 1.

The four group kinds

Kafka 4.x groups come in four kinds, described by different RPCs with different response shapes. Do not flatten them into one struct.

This is the single most likely thing to get wrong in a Kafka 4.x client, because every earlier version of Kafka had exactly one kind and every tutorial still assumes it.

ListGroups reports the type per group, so the flow is: list, then dispatch to the right describe RPC per group type.

The distinction is preserved, not flattened

GroupDescription is an enum whose variants carry the fields their protocol actually has — not a struct with everything optional:

struct ClassicGroupMember; struct ConsumerGroupMember; struct ShareGroupMember;
#[derive(Debug)] enum GroupState { Empty }
enum GroupDescription {
    Classic {
        group_id: String,
        state: GroupState,
        protocol_type: String,      // "consumer", "connect", …
        protocol: String,           // the assignor the group agreed on
        members: Vec<ClassicGroupMember>,
    },
    Consumer {
        group_id: String,
        state: GroupState,
        group_epoch: i32,           // KIP-848 has epochs; classic has generations
        assignment_epoch: i32,
        assignor: String,           // server-side, not client-negotiated
        members: Vec<ConsumerGroupMember>,
    },
    Share { /* … */ },
    Unrecognized { group_id: String, group_type: String },
}

Flattening these into one struct forces every field to be Option, and then every consumer has to know which combinations are possible for which kind — which is the same knowledge, moved somewhere it cannot be checked.

The differences are real, not cosmetic. A classic group negotiates its assignor between members and has a generation; a KIP-848 consumer group has the broker choose the assignor and has a group epoch and a separate assignment epoch. There is no honest single field for "the version number of this group's membership".

Unrecognized is what makes this work on real clusters

Streams groups exist on the wire, list on any 4.1+ broker running Kafka Streams, and have no schema in kafka-protocol 0.17. See The upstream schema gap.

So the enum needs a variant meaning "this group exists, I know its id and its type, and I cannot tell you more":

enum GroupDescription {
Unrecognized { group_id: String, group_type: String },
}

A UI that hard-fails on an undescribable group is a UI that hard-fails on most real clusters. Rendering the group id with "streams group — not describable by this client" is both honest and useful; returning an error for the whole group list because one entry was a streams group is neither.

The acceptance test asserts a fourth, undescribable group type surfaces as Unrecognized rather than Err.

GroupState::Other(String)

The same reasoning one level down. The named states are Empty, PreparingRebalance, CompletingRebalance, Stable and Dead, and Other(String) carries anything else the broker says.

Group states are transmitted as strings, and Kafka adds them — KIP-848 introduced Assigning and Reconciling for consumer groups. Collapsing an unknown state into Dead or into an error would be worse than reporting the string the broker actually sent.

Offset reset differs by protocol

A trap worth naming because it fails on exactly one of the two group types, which is easy to miss if your fixture only covers one:

Get it wrong and the broker returns ILLEGAL_GENERATION — against one kind only, while the other keeps working.

kafka-admin also refuses an offset reset when the group is not EMPTY, with a clear error, rather than letting the broker accept a commit that a live member immediately overwrites. That is a silent no-op from the operator's point of view, and a silent no-op in an admin tool is worse than a refusal.

Fixtures come from the container

Worth recording, because it is a trap in the test rather than the code.

librdkafka has no KIP-932 share-group support, so rdkafka cannot generate the share-group fixture the acceptance test requires — and it drags cmake and a C toolchain into CI besides.

The apache/kafka image already ships kafka-console-consumer.sh and kafka-console-share-consumer.sh. Driving those through testkit's exec helper reaches every group kind with zero build dependencies. rdkafka earns its place in the interop suite, where being a genuinely different client implementation is the entire point.

KIP index

The Kafka Improvement Proposals this library actually implements, degrades on, or deliberately does not implement. Ordered by number.

Legend: ✅ implemented · ⚠️ partial or degraded · ❌ not implemented · 🚫 blocked upstream

Each KIP is searchable by number on the Apache Kafka KIP index.

The ones worth expanding on

KIP-848 — the new consumer group protocol ✅

Kafka 4.x's default group protocol, and the reason group kinds is its own chapter. Assignment moves server-side: the broker computes it and the client acknowledges, replacing the JoinGroup/SyncGroup dance entirely.

Both halves are implemented. ConsumerGroupDescribe renders these groups completely — epochs, assignor, per-member assignment — and ConsumerGroupHeartbeat joins one, with a client-generated member id and a broker-computed assignment. This was adopted before the classic protocol on purpose: server-side assignment removes the byte-compatibility problem that makes classic hard, so it was the cheaper of the two to get right first.

KIP-932 — share groups ⚠️

Queue semantics on top of Kafka: multiple consumers on the same partition, per-record acknowledgement, no partition-exclusive ownership.

ShareGroupDescribe is implemented. Note that librdkafka has no KIP-932 support at all, so rdkafka cannot even generate a share-group fixture — the tests drive kafka-console-share-consumer.sh in the container instead.

KIP-405 and KIP-1005 — tiered storage ⚠️

Both sentinels this library can reach are surfaced distinctly rather than collapsed, and that matters more on a tiered cluster than anywhere else: EARLIEST and EARLIEST_LOCAL_TIMESTAMP differ by exactly the data that has been offloaded to remote storage, which on a tiered cluster is most of it. A UI that treats them as interchangeable reports wrong retention.

The third tiered sentinel, KIP-1023's -6, is blocked upstream.

KIP-227 — incremental fetch sessions ✅ in one crate, deliberately not the other

kafka-consume establishes and maintains sessions, which is what a steady-state consumer needs: after the first full fetch, subsequent requests carry only what changed.

kafka-read deliberately does not. crates/kafka-read/src/fetch.rs pins session_id = 0, session_epoch = -1 — Java's FetchMetadata.LEGACY sentinel — because a browse-shaped scan is one-shot, and an incremental session would make each scan depend on the last for no benefit.

The split is the point: the same KIP is right for one crate and wrong for the other, which is why they are separate crates.

KIP-98 — exactly-once ✅ · KIP-447 — exactly-once v2 ⚠️

Both halves of KIP-98 are here. On the read side, Visibility::CommittedOnly sets read_committed and the client filters aborted records using the AbortedTransactions list the broker returns — the broker does not filter for you. On the write side, kafka-produce claims a producer id, tracks per-partition sequences, and drives InitProducerId/AddPartitionsToTxn/EndTxn behind init_transactions, begin_transaction, commit_transaction and abort_transaction — including the epoch bump KIP-890 hides inside EndTxn. DescribeTransactions, ListTransactions and DescribeProducers inspect the resulting state.

KIP-447 is the gap. There is no sendOffsetsToTransaction: a consumer's offsets cannot be committed inside a producer transaction, so the consume-process-produce loop cannot be made exactly-once end to end. Both AddOffsetsToTxn and TxnOffsetCommit are already typed and routed in kafka-conn, so what is missing is the producer-side method, a group-metadata type to carry member_id/generation across the crate boundary, and an acceptance test — not wire support.

Tracked in #10.

KIP-699 — batched FindCoordinator ⚠️

The batched form resolves many coordinators in one round trip. kafka-meta uses the single-key form, which is correct on every broker version and costs an extra round trip per group on a cold cache. Worth revisiting for a UI rendering hundreds of groups at once.

Non-goals

The fastest honest answer to "can I use this". Everything here is a decision, not a gap waiting to be closed by accident — though Roadmap covers which ones are on the list to lift.

What is no longer a non-goal

This page used to open by saying kaas-lib was not a general-purpose client: a small producer with no accumulator, and no group membership at all. Both have since shipped, and the page is kept honest here rather than quietly edited, because "can I use this" deserves a straight answer.

• The producer is a real one. kafka-produce has a batching accumulator with linger and bounded buffer memory, Java-compatible murmur2 partitioning, KIP-480 sticky partitioning, every compression codec, idempotence and transactions.
• Group membership exists, in both protocols — KIP-848 and the classic JoinGroup/SyncGroup/Heartbeat path, with assignor payloads byte-identical to Java's.
• Incremental fetch sessions exist (KIP-227), behind a streaming fetcher that batches partitions per broker.

Two genuine omissions remain inside that surface, and both are decisions:

• acks=0 is not offered. It is a request the broker never answers, so a correlation-based client would leave a pending oneshot forever and report every successful write as a timeout. Refused at the config boundary rather than given a fire-and-forget path.
• Classic groups advertise range, roundrobin and cooperative-sticky, not eager sticky. StickyAssignor carries its state in the subscription's user_data as a struct with no schema in kafka-protocol, and hand-rolling a wire format is what this codebase does not do. Cooperative-sticky has no such problem, so incremental rebalancing is available; a group whose other members are pinned to eager sticky alone fails at join time with INCONSISTENT_GROUP_PROTOCOL, which is loud rather than subtle.

rdkafka remains the more mature option, and the honest tradeoff is maturity against toolchain: it wraps librdkafka and wants cmake and a C toolchain, where this is Rust apart from the lz4 and zstd codecs, which reach C through lz4-sys and zstd-sys.

kafka-read is browse-shaped, and stays that way

This is a scoping decision between two crates rather than a missing feature. If you want a consumer, use kafka-consume. kafka-read's scan and tail answer a UI's questions — show me this topic from here, show me what just happened — and they are bounded, one-shot, and never commit an offset.

Consequences worth being explicit about, all of them about kafka-read specifically:

• No incremental fetch sessions. Every fetch is a full fetch. Correct for one-shot scans, wrong for a steady-state consumer — which is why kafka-consume has them and this does not. See KIP-227.
• No offset commits from the read path. kafka-admin can commit offsets as an admin operation (resetting a group's position) and kafka-consume commits as a group member; kafka-read never does it as a side effect of reading.
• No partition assignment or ownership. Two scan calls on the same partition both read it. Nothing coordinates them, because nothing is supposed to — coordination is what group membership is for.

We do not parse __consumer_offsets

Use OffsetFetch. The internal format is not a stable interface.

It is tempting — reading the internal topic directly gives you every group's offsets in one scan instead of one OffsetFetch per group. It is also a format Kafka changes between releases without notice, because it is internal, and a client that parses it is a client that breaks on upgrade in a way nobody can debug from the outside.

This is why FindCoordinator is classified read-only by the gate despite being able to trigger __consumer_offsets creation on some clusters: the alternative is worse.

We do not act as a broker

kafka-protocol is depended on with default-features = false and the client feature, which drops the broker half of the codegen — every response encoder and request decoder across 87 message types.

That is not a size optimisation, it is a statement of scope. The KRaft and broker-internal APIs are not "not yet implemented"; they are not ours to send.

We do not share code with the kaas broker

Stated as a non-goal because it looks like obvious reuse and is not.

kaas is a Kafka-compatible broker by the same author, and it has a codec. Using it here would be a mistake: kaas-lib is the natural conformance harness for kaas, and two implementations sharing a codec share its bugs. A mutual misreading of the spec encodes and decodes consistently, passes green, and hides precisely the class of wire bug the harness exists to catch.

So testkit's bootstrap addresses sit behind a trait rather than hardcoding the Apache image, and the codec dependency stays kafka-protocol.

We do not hide upstream gaps

Where kafka-protocol cannot express something, the library says so rather than working around it. ErrorCode::Unknown(i16), GroupDescription::Unrecognized, ours: None in the version table, the documented-but-unreachable ListOffsets -6 sentinel — all of these are visible degradation by choice.

A hand-rolled schema to fill a gap would be a private fork of a generated artifact, and it would destroy the compile-error signal that makes the error taxonomy trustworthy. See The upstream schema gap.

We do not mock brokers

Every milestone is verified against a real broker in a container. There are no mocked responses, no recorded fixtures replayed as if they were a cluster, and no todo!() standing in for an unfinished path.

The cost is that the meaningful tests need Docker and take minutes. The benefit is that "the tests pass" and "it works against Kafka" are the same statement. See Verification.

Verification

A green cargo build is not evidence. Every milestone has an acceptance command, and it must pass before the milestone is done.

Four layers, in increasing order of cost and of what they prove.

Unit — the fast gate

cargo xtask ci is fmt, clippy across all targets with -D warnings, and the unit tests. No Docker, so it stays fast enough to run on every save, and it is what CI runs on every push.

.cargo/config.toml sets rustflags = ["-D", "warnings"], so warnings fail the build locally rather than only in CI. The lints from rule 2 live in [workspace.lints] at the root, which means a new crate inherits them by adding [lints] workspace = true to its manifest rather than by repeating #![deny(...)] attributes and eventually forgetting one.

The unit tests that matter here are table-driven over things that are easy to get quietly wrong: every error code including one no Kafka release defines, the version-negotiation clamp in both directions, the routing class of every api key, and allow_auto_topic_creation being false.

Acceptance — real brokers, no mocks

Every acceptance test wears #[testkit::integration_test]: #[ignore]d by default so cargo test stays useful without a Docker daemon, and capped at two minutes of wall clock per test — a test that exceeds the deadline fails, and cargo xtask refuses a hand-written #[ignore] that would dodge it. Each test boots apache/kafka:4.3.1 in KRaft mode through testkit.

The acceptance tests are written to fail on the specific wrong implementation rather than to confirm the right one. A few examples of what that means in practice:

• Backward scan asserts not just that the last 500 records of a 100k-record partition come back correctly, but that fewer than 5% of the partition's bytes were fetched — measured with the connection byte counters. A naive implementation that reads the whole partition returns the right answer and fails this test.
• Auto topic creation requests metadata for a nonexistent topic against a broker with auto.create.topics.enable=true, then uses a second client to assert the topic was not created. Asserting on our own response would prove nothing.
• Truncated batches fetch with max_bytes small enough to cut a batch mid-stream and assert zero Malformed events. The truncation must be invisible, and a decoder that reports it passes a naive test and fails this one.
• KIP-368 re-authentication runs a broker with connections.max.reauth.ms around 10 seconds and asserts a connection survives past twice that window while still serving requests. The only way to prove re-auth works is to let a session expire.
• The read-only gate drives its assertion from ApiKey::iter rather than a hand-written list, so new protocol keys are covered automatically.
• Per-item results describe 50 topics where 2 do not exist and assert 48 Ok plus 2 Err, not a global Err.

Multi-broker fixtures (cluster(3)) exist because leader spread, log dirs and reassignment are not observable on one broker.

The suite runs --no-fail-fast. Cargo otherwise stops at the first test binary that fails, and each one here is a whole milestone — one broken assertion in kafka-read's forward scan is enough to leave the leak suite unrun and M11's status simply unknown. Finding out which milestones are red is the entire point of the command, so it pays for all of them; the exit status still carries any failure.

Fuzz — rule 2, made executable

cargo xtask fuzz

A cargo-fuzz target over RecordBatch bytes whose pass condition is simply no panic. That is rule 2 stated as a program rather than as an intention: a malformed record from one topic must not kill a server hosting other clusters.

It needs a nightly toolchain, so it gets its own CI job rather than pinning the whole workspace to nightly for one target.

Its first genuinely green run found a real bug, and found it as an out-of-memory rather than a panic: a 99-byte batch declaring 285 million records, against a decoder that sized its Vec from that number before parsing anything. Worth internalising — rule 2 can be violated without any unwrap in sight, by allocation rather than by abort, and libFuzzer counts that as a finding even though "no panic" is the stated pass condition. See Tolerant decoding.

Interop — the silent-wrongness class

cargo xtask interop

Produce with rdkafka, read with kafka-read, and the reverse. This is the only layer that catches bugs where both ends of our code agree with each other and disagree with the rest of the world:

• murmur2 partitioning
• snappy xerial framing
• header encoding
• tombstones (a null value must round-trip as null, not as empty bytes)

The snappy path is asserted explicitly, and it is the assertion that paid. kafka-protocol 0.17.0 rewrote snappy to emit Java/xerial framing — 0.16 and earlier were mutually incompatible with the Java client — and it decodes by autodetecting between that and raw snappy. It is the newest code in the dependency and the one we were least entitled to assume was right.

It was not. rdkafka writes raw unframed snappy, upstream's autodetection consumes the bytes it is sniffing, and the interop case failed the first time it ran — a bug no unit test in this workspace could have found, because both ends of every unit test are our own code and our own encoder emits xerial. That is precisely the silent-wrongness class this layer exists for. See The upstream schema gap.

The interop crate is deliberately outside the workspace: rdkafka builds librdkafka from C source and wants cmake, which is a fine thing to require of the cross-client job and a terrible thing to require of cargo xtask ci.

Live clusters

Beyond the container fixtures, the workspace has livetest — a binary that runs the library against real Kafka clusters rather than ephemeral containers. It emits partial reports on failure and ranks topics by record count, so a run against a production-shaped cluster produces something readable rather than a stack trace. See Testing against a real cluster.

This is also where the conformance-harness idea from Non-goals becomes concrete: pointing the same suite at a kaas broker instead of apache/kafka:4.3.1 turns it into a typed parity check with real diffs.

What CI runs when

The slow three are manual because they are minutes rather than seconds. That is a deliberate trade and it has a cost: the acceptance tests are the ones that actually decide whether a milestone is done, so they have to be run — by a person, before saying so.

Workspace layout

Ten crates in the workspace, plus one deliberately outside it. Six of them publish, in lockstep at a single version.

The dependency graph

graph TD
    admin[kafka-admin] --> meta[kafka-meta]
    read[kafka-read] --> meta
    produce[kafka-produce] --> meta
    consume[kafka-consume] --> read
    meta --> conn[kafka-conn]
    conn --> kp[["kafka-protocol"]]
    live[livetest] --> admin
    live --> read
    live --> produce
    live --> consume
    live --> tk[testkit]
    admin -.dev.-> tk
    read -.dev.-> tk
    meta -.dev.-> tk
    produce -.dev.-> tk
    consume -.dev.-> tk

Strictly layered, no cycles, no sideways edges. kafka-admin and kafka-read do not know about each other; both reach the wire only through kafka-meta.

The one edge that is not straight down is kafka-consume → kafka-read, and it is a deliberate reuse rather than a layering slip: the consumer decodes record batches with the same tolerant decoder the scan path uses, so a batch that will not parse behaves identically on both. Duplicating that decoder to keep the graph tidier would mean two tolerant decoders drifting apart, which is the more expensive kind of tidiness.

Why interop is outside the workspace

rdkafka builds librdkafka from C source and wants cmake and a C toolchain. That is a fine thing to require of a job whose entire purpose is cross-client interoperability, and a terrible thing to require of cargo xtask ci — CI on a minimal runner image has already been red once over exactly this.

So the crate stands alone with its own [workspace] table and is built by cargo xtask interop.

Lints are inherited, not repeated

[workspace.lints] at the root carries rule 2:

[workspace.lints.clippy]
unwrap_used              = "deny"
expect_used              = "deny"
panic                    = "deny"
indexing_slicing         = "deny"
as_conversions           = "deny"
cast_possible_truncation = "deny"
cast_sign_loss           = "deny"

[workspace.lints.rust]
unsafe_code                   = "forbid"
missing_debug_implementations = "warn"

A new crate opts in with [lints] workspace = true in its manifest. That is deliberately better than repeating #![deny(...)] attributes at each crate root, because the failure mode of the attribute approach is a crate that quietly forgets one.

Test code is exempted per crate with a #![cfg_attr(test, allow(...))] block — tests may unwrap freely, and requiring them not to would make every assertion three lines long.

Note indexing_slicing and the cast lints. Denying a[i] in library code is unusual and it is the right call here: a panic from an out-of-bounds index on a malformed response is exactly the failure rule 2 exists to prevent, and it is the easiest one to write by accident.

The codec dependency

kafka-protocol = { version = "0.17", default-features = false,
                   features = ["client", "messages_enums",
                               "gzip", "snappy", "lz4", "zstd"] }

Every part of that line is deliberate:

• default-features = false + client drops the broker half of the codegen — every response encoder and request decoder across 87 message types. We never act as a broker. See Non-goals.
• messages_enums is not in default. We need it: RequestKind and ResponseKind are what the read-only gate and the generic dispatch layer match on.
• The four compression features are already in default. Naming them matters only because defaults are off.

It is declared once in [workspace.dependencies] so an upstream bump is a one-line change rather than a five-manifest change.

Where to start reading

1. crates/kafka-conn/src/conn.rs — the connection actor, and the shape everything else assumes.
2. crates/kafka-conn/src/versions.rs — small, and it explains why the whole workspace is built around a codec that is a release behind.
3. crates/kafka-meta/src/routing.rs — 209 lines that encode knowledge otherwise scattered across every call site.
4. crates/kafka-read/src/batch.rs — the tolerant decoder, the most subtle file in the workspace.

Rust and toolchain

Edition 2024, rust-version = "1.97", pinned by rust-toolchain.toml to 1.97.1 with rustfmt, clippy, rust-src and rust-analyzer. resolver = "3".

kafka-conn

Framing, correlation, version negotiation, TLS, SASL and the connection actor — everything between a TCP socket and a typed Kafka request.

Module map

What this crate owns that everything else borrows: ApiKey and ErrorCode, the two protocol vocabularies that would otherwise leak everywhere. Both carry an Unknown variant, because the codec ships Kafka 4.0 schemas and the brokers we target are newer. They live here rather than in kafka-meta because every crate — including this one — has to classify a broker's answer, and a workspace with two error types pushes a From conversion into every call site.

The one deliberate exception to rule 1: Connection::send is generic over kafka_protocol::protocol::Request. This crate is the wire boundary, and a parallel request trait here would convert protocol types into protocol types for no gain. Everything above is held to the rule without exception. See The domain boundary.

The re-export: kafka_conn::protocol carries the codec's Decodable, Encodable, Message, Request, StrBytes, HeaderVersion, plus compression, indexmap, messages and records. Crates above reach the codec through here so the version is pinned in one manifest. Re-exporting is not licence to expose it in a signature.

The subtle files:

• codec.rs — two header traps, both producing off-by-a-few-bytes failures. The response header version is not the request's api version, and ApiVersions responses always use response header v0 even on a flexible connection. Both go through helpers rather than being computed.
• versions.rs — negotiate vs negotiate_with. The first reads ApiKey::valid_versions(), which is right for a report and wrong for encoding, because a request and its response can have different ranges. OffsetFetch is the live example.
• api_key.rs — is_mutating is an allowlist of read-only keys with _ => true. The direction is the whole security property.
• scram.rs — real SASLprep via stringprep, and a constant-time server signature check.

Where the boundary sits: this crate knows nothing about clusters. Give it an address and it gives you request/response against that one broker. Leadership, coordinators, retries and pooling are all kafka-meta's problem.

Start reading at conn.rs's module docs, then versions.rs end to end — it is short, and it explains why the rest of the workspace is shaped the way it is.

Related chapters: The connection actor, Version negotiation, TLS, SASL and re-authentication, The error taxonomy, The read-only gate.

kafka-meta

The layer that knows what a cluster looks like. Everything above sends through Cluster, which resolves the right broker, retries on the errors that mean "your view is stale", and keeps an immutable snapshot readers can take without blocking.

Module map

The two tables. routing.rs and the error taxonomy are first-class artifacts, each in one file, because both encode knowledge that otherwise scatters into individual call sites and then quietly diverges. The error table lives one crate down in kafka-conn and is re-exported here, so the two sit together at the layer that acts on them.

routing.rs is 209 lines and worth reading in full. Six classes, not the four the summary suggests — Routing::Specific splits into Caller (a broker the caller names, for DescribeLogDirs) and PartitionLeader (a broker the snapshot names, for Fetch). Same routing class, completely different resolution.

Its wildcard arm is Routing::Any, and that is safe in a way the read-only gate's wildcard is not: mis-routing costs a redirect, mis-classifying a mutating API costs the property.

snapshot.rs and ArcSwap. The snapshot is immutable and swapped wholesale; reads never block and never wait on a refresh. It carries its own fetch timestamp (age()) because a UI rendering "as of 4 seconds ago" is honest and one rendering stale data as live is not.

TopicId wraps uuid::Uuid rather than exposing it, so the Fetch v13+ topic-id path does not leak a codec-adjacent type upward.

The one-word regression with a destructive blast radius. Every MetadataRequest sets allow_auto_topic_creation: false explicitly. The schema default is true and the crate honours it, so following "Default plus builders" literally produces a UI that creates a topic every time someone typos a name into a search box. A unit test asserts it.

pool.rs and the Kubernetes case. Endpoint is an enum — Node(i32) for a broker known by id, Bootstrap(String) for the addresses we were given — because when every known broker goes unreachable the pool must fall back to bootstrap and re-resolve. A cluster rolling every broker onto new addresses is a normal Kubernetes event, and a pool that only remembers metadata addresses never recovers from one.

Connecting happens under a per-endpoint async mutex, not a global one, so a slow handshake to a dead broker does not stall healthy ones and twenty callers for the same broker open one socket.

Where the boundary sits: this crate builds and sends kafka-protocol requests but returns owned types. Cluster::send is the seam every layer above uses; nothing above it opens a socket or picks a broker.

Start reading at routing.rs — it is the shortest complete statement of what this crate is for — then cluster.rs's send path.

Related chapters: Metadata, routing and the pool, The error taxonomy.

kafka-admin

The admin RPCs — 31 of the 37 api keys this library sends. Pure translation: build a request from owned types, send it through Cluster, convert the response back.

Module map

PerItem is the whole API shape

Every call naming several resources returns PerItem<Id, T> — Vec<(Id, Result<T, Error>)>, never Result<Vec<T>, Error>.

use kafka_admin::Admin;
async fn example(admin: &Admin) -> kafka_admin::Result<()> {
for (name, result) in admin.describe_topics(["orders", "shipments"]).await? {
    match result {
        Ok(topic) => println!("{name}: {} partitions", topic.partitions.len()),
        Err(error) => println!("{name}: {error}"),
    }
}
Ok(())
}

Describing 500 topics while 3 are mid-deletion returns 497 descriptions and 3 errors. The alternative makes a UI unusable on exactly the clusters that need one. errs and oks in types.rs are the helpers for splitting a PerItem when a caller genuinely wants one side.

Note that the outer Result still exists and still means something: it is the transport failing, not any individual item.

groups.rs is the biggest file for a reason

Four group kinds, three describable, all with different response shapes. See The four group kinds — this file is where that chapter lives in code, including the Unrecognized variant that keeps a streams group from taking down a group list.

It also holds the offset-reset path, where the classic protocol wants generation_id = -1 and KIP-848 wants member_epoch = -1, and where a reset against a non-EMPTY group is refused with a clear error rather than accepted as a commit a live member will immediately overwrite.

topics.rs and the pagination fallback

describe_topics prefers DescribeTopicPartitions — how the 4.x Java AdminClient describes topics, and it paginates. Unfiltered Metadata returns the whole cluster in one response, which on a 10k-topic cluster is a multi-megabyte payload on every refresh.

The fallback to Metadata has to handle two different causes arriving at the same place: the broker is too old to offer the API, or our schemas are too old to encode it. Hence Admin::supports and the Error::UnsupportedApi match.

offsets.rs and the six sentinels

Surfaced distinctly, not collapsed. Five are reachable; -6 (EARLIEST_PENDING_UPLOAD_TIMESTAMP, KIP-1023) needs ListOffsets v11 and the codec caps at v10 — the domain enum documents the gap rather than omitting it silently.

The private version helpers

admin.rs carries four pub(crate) helpers worth knowing about before adding a method:

• supports(ApiKey) — does this cluster offer the key at all
• negotiated_version(ApiKey) — for reporting
• negotiated_for::<R>() — for encoding, because a request's shape can change with its version and the codec rejects a field set outside its own range rather than ignoring it. "Set both the old and the new field" is an encode failure, not a compatibility trick.
• request_timeout_ms() — the pool's timeout in the milliseconds admin RPCs want

Where the boundary sits

No sockets, no broker selection, no retry — all of that is kafka-meta's. This crate is request construction and response translation, and the volume of it is the price of rule 1.

Start reading at types.rs for the vocabulary, then topics.rs for the simplest complete round trip, then groups.rs when you need the hard case.

kafka-read

The read path: browse a topic forwards, or read its tail. Shaped for a UI rather than for a consumer group — there is no rebalance, no commit, no membership.

Module map

batch.rs is the subtle one

Read its module docs before touching anything here. Three things in a fetch response look like corruption and are not — a truncated trailing batch, a control batch, and aborted-transaction records — and a decoder that reports them is worse than no decoder at all, because it cries wolf on every fetch of every healthy cluster.

The header submodule reads fixed byte offsets out of the v2 batch header directly (BASE_OFFSET, BATCH_LENGTH, MAGIC, ATTRIBUTES, LAST_OFFSET_DELTA, PRODUCER_ID). That is not schema duplication: it is the minimum needed to decide whether a batch is complete before handing it to a decoder that would otherwise report truncation as corruption.

See Tolerant decoding.

scan.rs — a Stream, never a Vec

Memory is bounded by max_buffered_records across the whole scan, not per partition. The naive implementation keeps one fetch's worth per partition, which on a thousand-partition topic is a thousand times the intended budget — and is discovered in production, because nobody scans a thousand-partition topic on a laptop.

Cross-partition ordering degrades gracefully rather than silently: when the buffer cap forces an emit before every partition is represented, the reorder is bounded by the buffer span rather than by the topic length, and ScanEvent::Progress reports that it happened.

backward.rs — not a forward read with a different start

Reading forward from latest - N is wrong on any topic where records are not one offset apart, which is every compacted topic and every topic that has had DeleteRecords run against it.

The step grows when a chunk yields fewer records than its offset span suggested. That is what stops a compacted partition with thousand-fold offset gaps from crawling backwards a handful of records per round trip and re-reading the whole log.

decompress.rs — bounded, and the one place we diverge from the codec

Gzip, LZ4 and zstd decompress through a Read wrapped in take(), so the limit applies during decompression and the oversized allocation never happens.

Snappy needs more care, because Kafka's snappy is two formats: the Java client writes snappy-java's xerial framing, librdkafka writes raw unframed snappy. kafka-protocol 0.17 autodetects and gets it wrong — it sniffs the magic header with a call that advances the buffer, so the raw fallback runs on bytes it has already eaten. Left alone, that makes every snappy topic written by a non-Java producer unreadable.

So this module picks the framing while the buffer is intact and delegates only the xerial case, which stays bounded on its compressed input because upstream allocates per block from each block's own declared length. The raw branch is one snap call, and it gets the better bound of the two: the block declares its decompressed size up front, so the check is exact.

That is a knowing divergence from the codec crate — the only one in the workspace — and it is written to be deleted when upstream fixes the detection.

fetch.rs — deliberately session-less

session_id = 0, session_epoch = -1: Java's FetchMetadata.LEGACY sentinel, no incremental fetch session. Correct for one-shot UI scans, wrong for a steady-state consumer, and a roadmap prerequisite for group membership.

min_bytes is 1 rather than 0. Zero would also work, but Kafka treats min_bytes = 0 as "return immediately even with nothing", which turns a scan into a spin when a partition is briefly empty.

Topic identification switches on the negotiated version: Fetch v13+ uses a Uuid, below that a name. Both paths exist here.

Where the boundary sits

This is the only place in the workspace that parses bytes from an untrusted producer. Everything it returns is owned — Record holds Bytes and String, never StrBytes.

Start reading at batch.rs's module docs, then record.rs for RecordOutcome, then scan.rs.

Related chapters: The read path, Tolerant decoding.

kafka-produce

The write path: encode a record batch, route it to the partition leader, and report where it landed. The first half of lifting the library past its admin-first scope.

Module map

Three decisions worth knowing before reading the code

acks=0 is refused at the type level

Acks has no None variant, so the mode cannot be selected and then fail at runtime. It is not squeamishness: acks=0 is a request the broker sends no response to at all, and Connection correlates every in-flight request on a HashMap<i32, oneshot::Sender<_>>. An acks=0 produce would register a waiter nothing ever resolves, and every successful write would surface to the caller as a timeout.

The alternative was a fire-and-forget path that drops the correlation entry at send time. That was rejected because it punches a hole in the connection actor's invariant that every in-flight request has a waiter, because a mode whose whole character is discarding results sits badly with a library that treats partial failure as a result, and because idempotence needs the response to advance its sequence numbers. What acks=0 actually buys — not waiting on the leader — is what a batching accumulator provides safely.

A rejection is not an ambiguous failure

This is the crate's central safety property, and it is a type rather than a boolean so that a new failure path has to declare which kind it is:

Collapsing the two is a bug in either direction. Retry everything and you duplicate a record on every timeout, with no error anywhere. Retry nothing and an ordinary leader election becomes a delivery failure.

The second is not hypothetical — it is what the library did until a live run against a second broker implementation caught it, on a freshly created topic whose leader had not settled. Note also that the backoff matters as much as the count: three immediate retries all re-read the same stale metadata and fail identically, so the crate reuses RetryPolicy rather than counting attempts itself.

The encoder splits batches where you do not expect

RecordBatchEncoder decides where one batch ends and the next begins by walking records while offset - sequence stays constant. Offsets necessarily increase, so the obvious thing — a constant NO_SEQUENCE on every record — makes that difference increase too, and every record is emitted as its own batch, each with its own 61-byte header and its own CRC.

The records all arrive, in order, and read back correctly. It is a throughput bug wearing a correctness result, and the only thing that catches it is an assertion on lastOffsetDelta in the encoded bytes. encode.rs counts the sequence up from NO_SEQUENCE, which is what the wire format implies anyway: the batch header stores a base sequence plus a per-record offset delta, and the decoder reconstructs the sequence as their sum.

murmur2 is checked against a different implementation, not against itself

A partitioner that is nearly Java's returns a partition for every key, round trips through our own reader, and passes any test written against ourselves. It just puts keys where a Java or C client would not look for them, which breaks co-partitioned joins and compacted-topic semantics silently and much later.

So partition.rs's own tests assert properties — determinism, range, tail handling for every length residue, spread — and the byte-exactness assertion lives in the interop crate, where rdkafka produces 1000 keys with partitioner=murmur2_random and every one must land where we say it does. That setting is explicit for a reason: librdkafka's default partitioner is not the Java-compatible one, so leaving it unset would compare our murmur2 against a different hash entirely.

The accumulator is an actor, and that is what makes cancellation tractable

Every piece of batching state — the open batch per partition, the closed ones queued behind it, and which partitions have a request on the wire — lives in one task and is touched by nothing else. Callers reach it through a channel, so dropping a send future drops a oneshot::Receiver and nothing more: a cancelled caller cannot leave a half-updated batch behind for the next one to trip over. The record it already enqueued is still produced; only the result is discarded.

At most one batch per partition is on the wire at a time. Different partitions proceed concurrently — ordering is a per-partition property — but within one partition the next batch waits for the previous answer. This is what makes retry safe: the moment a rejected batch is re-sent while a later batch for the same partition is already in flight, the log's order stops matching the caller's, with no error and no log line. Doing it per partition rather than per connection keeps the guarantee while still letting six partitions on one broker fill six batches concurrently.

It also explains why linger defaults to zero and should usually stay there. Records arriving during a round trip accumulate into the next batch on their own, so batching scales with load rather than with the setting.

Idempotence is routed differently from transactions

kafka-meta's routing table sends InitProducerId to the transaction coordinator, which is right for a transactional producer and wrong for an idempotent-only one: it has no transactional id, and the coordinator is resolved by that id — there is nothing to look one up with. Java sends this to any broker, and so do we.

The table is keyed on api key alone and cannot express "depends on whether a field is null", so this is a documented exception in idempotence.rs rather than a table change.

Transactions add three rules that are each quiet when broken: AddPartitionsToTxn must precede the first produce to each partition; the client ceiling on it is v3, because v4 (KIP-890) replaced the flat request with a transactions array and the clamp lives on the Rpc impl so no call site has to remember it; and PRODUCER_FENCED is terminal — another producer sharing the transactional id has bumped the epoch, so retrying is an infinite loop.

What is not here

acks=0 — a decision with a reason above rather than a gap. Batching, idempotence and transactions all landed in phase 2; see Roadmap for what is actually outstanding, and Producing records for the user-facing surface.

kafka-consume

The long-running read path: incremental fetch sessions, a fetcher that batches per broker, and three tiers of membership over one engine. The other half of lifting the library past its admin-first scope.

Module map

One engine, three ways to decide what it reads

GroupConsumer and ClassicConsumer both wrap Consumer. The fetch path, the sessions, the decoding and the offset plumbing are identical; the only thing membership changes is where the assignment comes from.

That is why the manually-assigned mode is not a degraded group consumer. It is the substrate — and it is independently the right answer for pinning a reader to a partition, which is a thing UIs and single-instance jobs genuinely want.

graph TD
    gc[GroupConsumer<br/>KIP-848] --> c[Consumer]
    cc[ClassicConsumer<br/>JoinGroup/SyncGroup] --> c
    c --> f[BrokerFetcher]
    f --> s[FetchSession]
    c --> o[offsets]

The session epoch rules, which are easy to get subtly wrong

A consumer fetches the same partitions over and over. KIP-227 lets the broker remember the assignment, so every request after the first sends only what changed — in steady state, nothing at all.

• (0, 0) opens a session. Not (0, -1): that is the legacy sentinel meaning "no session at all", and sending it forever is how a consumer silently re-sends its whole assignment on every fetch while appearing to work perfectly.
• After the broker answers with a session id, every request uses (session_id, epoch + 1).
• A partition that leaves the assignment goes into forgotten_topics_data once, on the next request. Leaving it out instead means the broker keeps fetching a partition nobody is reading.
• FETCH_SESSION_ID_NOT_FOUND and INVALID_FETCH_SESSION_EPOCH mean the broker dropped the session — a restart, or eviction under cache pressure. Both are recovered by opening a new one with the full assignment, and neither is ever surfaced to the caller. A broker restart must not kill a consumer.

kafka-read's scan and tail deliberately keep the legacy sentinel. They are one-shot: a session would make each scan depend on the last and leave state on the broker for a client that is not coming back.

The fetch count scales with brokers, not partitions

kafka-read's fetcher takes one topic and one topic id per call, which is exactly right for scanning one partition and wrong for a consumer. A consumer holding twelve partitions across two topics on three brokers should send three requests per round, not twenty-four — so fetcher.rs groups the active assignment by leader and asks each broker once.

A broker with nothing assigned still gets one request while it holds a session, so the forgotten list can drain; only then is the fetcher dropped.

KIP-848: revoke, then acknowledge

ConsumerGroupHeartbeat replaces JoinGroup, SyncGroup and Heartbeat outright, and the broker computes the assignment — which removes the single largest source of subtle incompatibility in the classic protocol, because there is no assignor payload to get byte-identical.

What the client still owns is the reconciliation, and it is ordered:

listener.on_revoke  →  auto-commit  →  drop the partitions  →  acknowledge

Acknowledging an assignment whose predecessor has not yet been revoked means two consumers hold the same partition at once — duplicate delivery, with no error anywhere and nothing in any log to explain it. So a rebalance is two beats: the first learns the target and revokes, the second acknowledges what is now owned.

The acceptance test asserts union and intersection, not record counts. A reconciliation that acknowledges before revoking still delivers every record; counting records would pass while the bug is live, and only the empty intersection catches it.

The epoch sentinels are not interchangeable

Using -1 for a static member throws away the whole point of group.instance.id: the assignment is handed to somebody else instead of waiting for the restart.

A half-done rebalance is state, not a moment

Rule 5 says dropping a poll future must be safe, and a rebalance is the place that is hardest. So a reconciliation that has been computed but not carried out is held on the consumer (rebalance::Pending) rather than run inline inside the heartbeat: a poll dropped mid-callback does not skip it, because the next poll finds it and finishes it, still ahead of the acknowledging beat.

The cost is that on_revoke may run twice for the same partitions, and that is the tolerable half of the trade — a listener that flushes twice writes the same bytes twice, while a listener that never fires loses them. It is documented as at-least-once rather than papered over.

The ordering inside the callback is the other half: the caller flushes first and the offset commit follows, so a committed offset always trails data the caller has already written. Committing first inverts exactly that, and the window is as long as the caller's flush.

The classic protocol's two hard constraints

The assignor payload has to be byte-identical to Java's. The group leader computes the assignment, and the leader is whichever member the coordinator picked — possibly a Java client decoding what we encoded. There is no negotiation of the format: it is ConsumerProtocolSubscription and ConsumerProtocolAssignment, and a field misread produces a group where somebody's assignment is silently empty. kafka-protocol ships both as real schemas, so none of it is hand-rolled.

Every member needs its own Cluster. JoinGroup blocks on the coordinator, and a Kafka broker will not read a second request from a socket until it has answered the first — so two members of one group sharing a connection deadlock, and it presents as a plain timeout. This is a property of the protocol rather than of this client, and GroupConsumer does not have it.

Cooperative rebalancing withholds, and that is the point

Under range/roundrobin a rebalance is eager: everyone revokes everything and takes what they are given. Under cooperative-sticky (KIP-429) the leader computes a sticky target and then withholds every partition whose owner is changing — round one assigns it to nobody, the losers revoke and re-join, and round two hands it over. A partition is never assigned to its next owner while its previous owner still holds it. Skipping the withholding step is the bug that delivers every record in a moved partition twice, silently.

Eager sticky is the one assignor deliberately missing: StickyAssignor carries its state in the subscription's user_data as a struct with no schema in kafka-protocol, and hand-rolling a wire format is what this codebase does not do. A group whose other members are pinned to eager sticky alone fails at join time with INCONSISTENT_GROUP_PROTOCOL, which is loud rather than subtle.

Offsets: the sentinel, and who is allowed to use it

A manually-assigned consumer still wants its position remembered, and the protocol expresses "not a member" with sentinels rather than a separate api: generation_id = -1 in a classic group, member_epoch = -1 in a KIP-848 one. Same wire field, same value — the one case where the two protocols agree. The member id must be empty; a made-up one is rejected with UNKNOWN_MEMBER_ID, which reads like a membership bug in a client that deliberately has no membership.

The anonymous form is honoured only while the group has no members, precisely so a detached client cannot scribble over a live group's positions. So a member commits under its own identity — member id, current epoch or generation, instance id if static. Getting this wrong is quiet in both directions, and an auto-commit whose result nobody checks is refused in silence.

NOT_COORDINATOR arrives inside a successful response

coordinator.rs is 74 lines and exists because of one asymmetry: Cluster::dispatch retries on Err, and a coordinator that has moved does not produce one. The round trip succeeds, and NOT_COORDINATOR arrives as a field inside the response — top-level on a heartbeat, per partition on an OffsetCommit. The routing layer has finished with the request by the time anything decodes that field, so nothing invalidates the cached coordinator and nothing asks again.

Every KIP-848 acceptance test failed this way, in under ten seconds — which is itself the tell: a retry budget being consulted would have spent it. So the re-ask lives above the decode, as a deadline rather than an attempt count, because the condition is not "the request failed" but "ask again in a moment".

kafka-produce reached the same conclusion for the transaction coordinator and keeps its own copy. Two private helpers rather than one shared public one is deliberate for now: this is a lockstep release, and a new public method on kafka-meta that kafka-consume calls in the same version is exactly what cargo publish --workspace --dry-run refuses to verify.

Where to start reading

1. session.rs — small, and the epoch state machine explains the shape of every fetch this crate sends.
2. consumer.rs's fetch_once — leader grouping, per-partition error handling, and where a malformed batch advances the position instead of stalling it.
3. group.rs's reconciliation — 60 lines that decide the ordering the whole membership story rests on.
4. classic.rs — longest file in the crate, and the assignors are the part worth reading against Java's own source.

Verification

cargo test -p kafka-consume -- --ignored boots real brokers through testkit. The group suites are the interesting ones: three KIP-848 members covering every partition with an empty intersection, and a mixed classic group with one Rust member and one kafka-console-consumer.sh, which is the case that makes byte-compatible assignor payloads non-optional.

testkit

Broker fixtures for the acceptance suite. Every milestone is verified against a real broker in a container — there are no mocked brokers in this workspace, by rule — and this crate is what makes that affordable.

publish = false.

Module map

The usual shapes

use testkit::{BrokerConfig, SaslMechanism, Security};

async fn example() -> testkit::Result<()> {
// The common case.
let broker = testkit::single_broker().await?;
let addr = broker.bootstrap_csv();

// Three nodes, for anything replica-aware.
let cluster = testkit::cluster(3).await?;

// Or configured.
let sasl = testkit::single_broker_with(
    BrokerConfig::new()
        .with_security(Security::SaslPlaintext)
        .with_mechanism(SaslMechanism::Plain)
        .with_user("alice", "alice-pw"),
)
.await?;
Ok(())
}

apache/kafka:4.3.1 in KRaft mode.

Tests take &dyn Cluster, never a concrete type

This is the crate's most important design decision and it is not abstraction for its own sake.

kaas-lib is the natural conformance harness for the kaas broker, and that only works if the acceptance suite can be pointed at something other than apache/kafka:4.3.1 without touching a single test. A hardcoded image in the fixtures would quietly foreclose that — not with an error, but by making the alternative expensive enough that nobody does it.

use std::fmt;
trait Cluster: fmt::Debug + Send + Sync {
    /// Bootstrap addresses reachable from the *test process* — for container
    /// fixtures the host-mapped port, not the in-container one.
    fn bootstrap(&self) -> &[String];
    fn nodes(&self) -> usize;
    // …plus exec, for driving the Kafka shell tools inside a node
}

ExternalCluster is the other half of the bargain: it makes "point the suite at a broker that is already running" a supported mode today, so the seam gets exercised rather than rotting into a trait nobody has implemented twice.

exec matters more than it looks

exec(node, argv) runs a command inside a fixture node, and several fixtures are generated by the Kafka shell tools rather than by a Rust client.

The group-kinds fixture is the reason. rdkafka cannot create a KIP-932 share group — librdkafka has no share-group support at all — so no Rust client in the workspace can generate that fixture. The apache/kafka image already ships kafka-console-consumer.sh and kafka-console-share-consumer.sh, and driving those through exec reaches every group kind with zero build dependencies.

The config builder exists from the start

single_broker_with(BrokerConfig) was built before anything needed it, because three later milestones do: SASL_PLAINTEXT/PLAIN and SASL_SSL/SCRAM-SHA-512 brokers, an authorizer-enabled broker for ACLs, and aggressive compaction settings for the backward-scan test.

The same argument applies to cluster(3). Retrofitting multi-broker fixtures after the fact is painful, and leader spread, log dirs and reassignment are not observable on one broker.

Start reading at harness.rs — it is short, and its module docs state the conformance-harness argument in full.

livetest

A binary that points kaas-lib at a real Kafka cluster — shared, long-lived, not ours, running a build we did not choose, holding data produced by clients we did not write.

publish = false.

Not a replacement for the container acceptance suite, which owns the cases needing a broker configured a particular way, killed mid-request, or fed a damaged log segment. This is the other half: the cases that only appear when the cluster is somebody else's.

Module map

Four commands

livetest probe    # read-only inventory + version table. Touches nothing.
livetest smoke    # admin round trip, prefixed resources only
livetest read     # scan and tail, asserting the decoder against real data
livetest sweep    # delete anything this tool left behind

probe is a conformance check in disguise

Its output is a sorted, diffable report. Run it against two clusters and diff the results, and you have a typed parity check — which is exactly the conformance-harness idea made concrete. Point it at kaas and at Apache Kafka and the diff is the answer.

Notes go to stderr and facts go to stdout, so livetest probe > out.txt captures exactly the diffable body and nothing else.

The report comes first, even on failure

struct Outcome; 
fn emit(outcome: Outcome) {
    // notes to stderr, facts to stdout, then the pass/fail result
}

A partial report says how far the run got and what the cluster looked like on the way, which is the entire diagnostic value. A failure that discards what it learned before failing is a stack trace, and a stack trace from a protocol mismatch against a cluster you cannot attach a debugger to is worth very little.

read ranks topics by record count, so a run against a production-shaped cluster reads the topics most likely to exercise the decoder rather than whichever five sort first alphabetically.

Everything is namespaced and swept

Every resource livetest creates carries a prefix (kaaslib-live by default), and sweep refuses to touch a name without it. Running a tool that creates topics against a shared cluster is only acceptable if cleanup is mechanical and cannot over-reach.

KAAS_TEST_READ_ONLY=1 turns on the read-only gate and the target then refuses any operation that would need to create something, with a clear error naming the environment variable rather than a permission failure from the broker.

Configuration

The live-cluster skill resolves all of these from Kubernetes — see Testing against a real cluster.

Start reading at target.rs, which is where a cluster stops being an environment variable and becomes a configured Cluster.

xtask

Repo-wide chores, as a binary in the workspace rather than a Makefile or a shell script. publish = false, and its only dependency is anyhow.

cargo xtask ci             # fmt + clippy + unit tests, no Docker
cargo xtask integration    # the #[ignore]d acceptance tests, needs Docker
cargo xtask fmt-check      # just the formatting gate
cargo xtask fuzz           # the record-batch fuzz target, needs nightly
cargo xtask interop        # cross-client checks against rdkafka, needs cmake
cargo xtask docs           # build this book (--serve for live reload)

Why these are separate commands

The split is not cosmetic — each one is separated from ci for a specific reason, and the reasons are worth knowing before adding a seventh.

ci is deliberately the unit gate. No Docker daemon, so it stays fast enough to run on every save, and it is what CI runs on every push. cargo build succeeding is not evidence of anything; integration is what decides whether a milestone is done.

integration is the slow half. Every test boots a real broker in a container, so it is minutes rather than seconds. Manual in CI.

fuzz needs nightly. cargo-fuzz does not work on stable, and pinning the whole workspace to nightly for one target would drag every other crate along with it. It gets its own toolchain invocation and its own CI job.

interop needs cmake. rdkafka builds librdkafka from C source. That is a fine requirement for the cross-client job and a terrible one for ci — CI on a minimal runner image has already been red once over exactly this — so the interop crate lives outside the workspace entirely.

docs needs three binaries on PATH: mdbook, mdbook-mermaid and mdbook-linkcheck. The docs job in .github/workflows/ci.yml pins all three. mdbook build runs the linkcheck backend too, so a broken cross-reference fails the build rather than shipping as a 404.

The version pins

mdbook stays on the 0.4.x line: mdbook-linkcheck 0.7.7 and mdbook-mermaid ≥ 0.17.0 target different mdbook major lines — 0.17.0 is built against mdbook 0.5's preprocessor protocol and fails against 0.4 — so 0.16.2 is the newest mermaid preprocessor that works here.

Bump all three together when moving to mdbook 0.5, and keep ci.yml and docs-publish.yml in lockstep.

Adding a task

main.rs is a match on env::args().nth(1) with a bail! default that lists the known tasks. Keep that list current — it is the only help text there is.

Connecting

Everything starts from a Cluster: the metadata cache, connection pool, routing and retry policy behind one cheap-to-clone handle. Clone it freely — every clone shares the same cache and the same connections.

use kafka_meta::{Cluster, ClusterConfig};

async fn example() -> kafka_meta::Result<()> {
let cluster = Cluster::connect(["broker-1:9092", "broker-2:9092"], ClusterConfig::default()).await?;
Ok(())
}

Admin::connect and Admin::connect_read_only build one for you; admin.cluster() hands it back for the read path.

Give it more than one bootstrap address

Bootstrap addresses are re-resolved when every known broker goes unreachable. A cluster that rolls all its brokers onto new addresses is a normal Kubernetes event, and a pool that only remembers addresses from its last successful metadata fetch never recovers from one.

A single bootstrap address is fine when it is a stable service DNS name that load-balances; it is a liability when it is one pod IP.

ClusterConfig

use std::time::Duration;
use kafka_meta::ClusterConfig;

fn example() {
let config = ClusterConfig {
    refresh_interval: Duration::from_secs(30),
    max_staleness: Duration::from_secs(5),
    ..ClusterConfig::default()
};
}

Kafka's own client default for metadata refresh is five minutes. A UI wants fresher than that, and metadata for a large cluster is not cheap, so 30 seconds is the compromise — with on-demand invalidation doing the real work whenever a NOT_LEADER_OR_FOLLOWER comes back.

ConnectionConfig

use std::time::Duration;
use kafka_conn::ConnectionConfig;

fn example() {
let connection = ConnectionConfig::new()
    .with_client_id("cluster-ui")
    .with_request_timeout(Duration::from_secs(30))
    .with_connect_timeout(Duration::from_secs(10))
    .with_max_in_flight(5);
}

Set client_id to something recognisable. It appears in broker request logs and in quota attribution, and "which client is hammering this cluster" is a question someone will eventually ask about your service.

max_in_flight defaults to 5, matching Kafka. The broker processes one connection's requests in order regardless, so this trades head-of-line blocking for memory rather than buying parallelism. Zero is clamped to 1.

TLS

use kafka_conn::{ConnectionConfig, TlsConfig};

fn example() -> kafka_conn::Result<()> {
// System trust roots.
let tls = TlsConfig::system();

// Or a private CA.
let tls = TlsConfig::with_ca_pem(std::fs::read("ca.pem")?);

// Or mutual TLS.
let tls = TlsConfig::system()
    .with_client_certificate(std::fs::read("client.pem")?, std::fs::read("client.key")?);

let connection = ConnectionConfig::new().with_tls(tls);
Ok(())
}

with_server_name is the one you will need unexpectedly. Brokers advertise the names in their own advertised.listeners, and those routinely do not resolve from where the client is running — behind a Kubernetes service, a load balancer, or a port-forward. Overriding the name sent in SNI and verified against the certificate is what makes that work without disabling verification.

SASL

use kafka_conn::{ConnectionConfig, SaslConfig, SaslMechanism, TlsConfig};

fn example() {
let sasl = SaslConfig::new(SaslMechanism::ScramSha512, "ui-service", "hunter2");

let connection = ConnectionConfig::new()
    .with_tls(TlsConfig::system())
    .with_sasl(sasl);
}

PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512.

PLAIN over a plaintext transport sends a recoverable password in the clear, and the library knows it — that combination requires allow_plaintext_password() to be called explicitly rather than being silently permitted. If you find yourself reaching for it outside a test fixture, reach for TLS instead.

Re-authentication is automatic. On any cluster with connections.max.reauth.ms set, the connection re-issues SaslAuthenticate before the session expires; without that the broker kills the connection and the symptom looks like a network fault. See TLS, SASL and re-authentication.

Read-only clients

use kafka_conn::ConnectionConfig;
use kafka_meta::ClusterConfig;

fn example() {
let config = ClusterConfig {
    connection: ConnectionConfig::new().read_only(),
    ..ClusterConfig::default()
};
}

Every mutating api key now returns Error::ReadOnly before a socket is touched. Admin::connect_read_only is the shorthand.

This is a client-side safety catch, not a replacement for broker ACLs — see The read-only gate for what it does and does not protect.

Inspecting what was negotiated

Useful when a call fails with UnsupportedApi and you want to know which side is the ceiling:

use kafka_conn::{ApiKey, Connection};
async fn example(conn: &Connection) {
for entry in conn.versions().entries() {
    println!("{} broker={:?} ours={:?}", entry.api_key, entry.broker, entry.ours);
}
}

ours: None means the codec has no schema for that key at all. broker_ahead() is true whenever the broker offers something newer than we can encode — which, given the upstream gap, is the normal case.

Admin operations

Every call naming several resources returns PerItem<Id, T> — Vec<(Id, Result<T, Error>)>. Handle the items; the outer Result is the transport failing, not any individual resource.

use kafka_admin::{Admin, ClusterConfig};

async fn example() -> kafka_admin::Result<()> {
let admin = Admin::connect(["localhost:9092"], ClusterConfig::default()).await?;
Ok(())
}

Topics

use kafka_admin::{Admin, NewTopic};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
// Create. NewTopic::new(name, partitions, replication_factor)
for (name, result) in admin.create_topics([NewTopic::new("orders", 6, 3)]).await? {
    match result {
        Ok(created) => println!("{name}: {} partitions", created.partitions),
        Err(error) => println!("{name}: {error}"),
    }
}

// List, describe, delete.
let names = admin.list_topics().await?;
let described = admin.describe_topics(["orders"]).await?;
let deleted = admin.delete_topics(["scratch"]).await?;

// Grow a topic. Partitions can only ever increase.
admin.create_partitions([("orders".to_owned(), 12)]).await?;
Ok(())
}

describe_topics prefers DescribeTopicPartitions and paginates, falling back to Metadata when the broker or our codec cannot offer it. On a 10k-topic cluster that difference is a multi-megabyte payload per refresh.

validate_topics exists — use it to check a creation would succeed without performing it, which is what a UI's "check" button should call.

Configs

use kafka_admin::{Admin, ConfigChange, ConfigResource};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
let configs = admin.describe_configs([ConfigResource::topic("orders")]).await?;

admin
    .alter_configs([(
        ConfigResource::topic("orders"),
        vec![ConfigChange::set("retention.ms", "604800000")],
    )])
    .await?;
Ok(())
}

This is IncrementalAlterConfigs underneath, and that matters: the legacy AlterConfigs silently resets every config you did not mention. kaas-lib does not expose the legacy API at all, so this class of accident is not reachable from here.

describe_configs_documented additionally returns the broker's own documentation strings, which is what a config editor wants for tooltips.

Offsets

use kafka_admin::{Admin, OffsetSpec};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
let latest = admin
    .list_offsets([("orders".to_owned(), 0)], OffsetSpec::Latest)
    .await?;
let range = admin.topic_offset_range("orders").await?;

// Or a different spec per partition.
let mixed = admin
    .list_offsets_with([("orders".to_owned(), 0, OffsetSpec::Earliest)])
    .await?;
Ok(())
}

Six sentinels, five reachable:

On a tiered cluster, Earliest and EarliestLocalTimestamp differ by exactly the data that has been offloaded to remote storage — which is usually most of it. Treating them as interchangeable is how a UI reports wrong retention.

Groups

use kafka_admin::{Admin, GroupDescription};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
for listing in admin.list_groups().await? {
    // listing carries the group type — classic, consumer, share, or something else
}

for (id, result) in admin.describe_groups(["analytics"]).await? {
    match result {
        Ok(GroupDescription::Classic { members, .. }) => { /* generation-based */ }
        Ok(GroupDescription::Consumer { group_epoch, .. }) => { /* KIP-848 */ }
        Ok(GroupDescription::Share { .. }) => { /* KIP-932 */ }
        Ok(GroupDescription::Unrecognized { group_type, .. }) => {
            // A streams group, most likely. Render it; do not fail.
        }
        Err(error) => println!("{id}: {error}"),
    }
}

// None means "every partition the group has committed for".
let committed = admin.fetch_offsets("analytics", None).await?;
Ok(())
}

Handle Unrecognized. Streams groups list on any 4.1+ broker running Kafka Streams and cannot be described by this build — see The four group kinds. A UI that treats it as an error hard-fails on most real clusters.

Resetting a group's offsets works through reset_offsets / delete_offsets, and it refuses when the group is not EMPTY rather than letting the broker accept a commit that a live member will immediately overwrite. A silent no-op in an admin tool is worse than a refusal.

Security

use kafka_admin::{Admin, AclFilter, QuotaFilter};

async fn example(admin: &Admin) -> kafka_admin::Result<()> {
// Both filters default to matching everything.
let acls = admin.describe_acls(&AclFilter::default()).await?;
let quotas = admin.describe_client_quotas(&QuotaFilter::default()).await?;
let scram = admin.describe_scram_credentials(["alice"]).await?;
Ok(())
}

ACLs, client quotas and SCRAM credentials, describe and alter. create_acls and delete_acls are per-item like everything else.

Cluster and storage

use kafka_admin::Admin;
async fn example(admin: &Admin) -> kafka_admin::Result<()> {
let cluster = admin.describe_cluster().await?;
let dirs = admin.describe_all_log_dirs().await?;
let sizes = admin.topic_sizes().await?;
Ok(())
}

topic_sizes joins DescribeLogDirs against Metadata for per-topic size. It does not double-count replicas — an RF=3 topic reports its single-replica size, not three times it, and there is an acceptance test asserting exactly that because getting it wrong produces a plausible-looking number.

Partitions and transactions

use kafka_admin::Admin;
async fn example(admin: &Admin) -> kafka_admin::Result<()> {
let ongoing = admin.list_partition_reassignments().await?;
let in_progress = admin.reassignments_in_progress().await?;

let txns = admin.list_transactions().await?;
let producers = admin
    .describe_producers([("orders".to_owned(), 0)])
    .await?;
Ok(())
}

Transactions and producers are describe-only — this library observes transaction state without starting one. See Non-goals.

Errors worth matching on

use kafka_conn::Error;
fn example(error: Error) {
match error {
    Error::ReadOnly { api_key } => { /* this client refuses mutations */ }
    Error::UnsupportedApi { api_key, broker, ours } => { /* which side is the ceiling? */ }
    Error::Authorization(code) => { /* ask your admin */ }
    Error::Decode { .. } => { /* this is our bug — report it */ }
    _ => {}
}
}

See The error taxonomy.

Producing records

kafka-produce writes records: it encodes a v2 record batch, routes it to the partition leader, and reports where it landed. Everything else on this page is about the two things a producer has to be honest about — where a record went, and whether it really got there.

[dependencies]
kafka-produce = "0.4"

Constructing one

Two constructors, and the difference is not cosmetic.

use kafka_produce::{ClusterConfig, Producer, ProducerConfig};

async fn example() -> kafka_produce::Result<()> {
// Its own connections, clamped to the in-flight count its guarantees allow.
let producer = Producer::connect(
    ["localhost:9092"],
    ClusterConfig::default(),
    ProducerConfig::new(),
)
.await?;

// Or share a cluster handle you already have — from `admin.cluster()`, say.
let cluster = producer.cluster().clone();
let second = Producer::new(cluster, ProducerConfig::new());
Ok(())
}

connect clamps the connections it opens to max_in_flight — one without idempotence, five with it, because the broker tracks exactly five in-flight sequence windows per partition. new cannot: the cluster it is handed is shared with whatever else is using it, and throttling a pool the admin and read paths also use would slow them down to protect a guarantee they do not need.

That is safe because ordering does not rest on that number. The accumulator holds at most one batch per partition on the wire, so a re-sent batch can never overtake a later one regardless. The clamp is defence for the connection layer, not the mechanism.

A Producer is cheap to clone, and every clone shares the metadata cache, the connection pool, the sticky partitioner's state and the accumulator. Sharing the accumulator is what makes batching work across clones: two clones producing to one partition fill the same batch, not two.

One record

use kafka_produce::{Producer, ProducerRecord};
async fn example(producer: &Producer) -> kafka_produce::Result<()> {
let meta = producer
    .send(
        ProducerRecord::new("orders")
            .with_key("customer-7")
            .with_value(r#"{"total":42}"#)
            .with_header("content-type", "application/json"),
    )
    .await?;

println!("{}-{} @{}", meta.topic, meta.partition, meta.offset);
Ok(())
}

RecordMetadata::timestamp is an Option and is usually None: on a CreateTime topic the timestamp stored is the one you supplied, so there is nothing for the broker to report back and guessing would be a fabrication. A topic configured message.timestamp.type=LogAppendTime fills it in.

The record

A record with no value is a tombstone. value: None and value: Some(Bytes::new()) are different records, and on a compacted topic the first deletes the key while the second stores nothing under it. kafka-read preserves the same distinction coming back, and the round-trip test asserts it — so never normalise one into the other.

One upstream limitation, stated where you would hit it: a duplicate header name cannot be written. ProducerRecord keeps duplicates and kafka-read returns them faithfully, but kafka_protocol's record type holds headers in an IndexMap, so a repeated name collapses to its last value on the way to the wire with no error. Reading records a Java producer wrote is unaffected; only writing them is impossible. Routing around it means hand-rolling the record format, which is the one thing this codebase does not do.

Many records: enqueue, not a loop of send

send accepts one record and waits for it, so a loop of send().await keeps exactly one record in flight and batches nothing. To get the throughput, enqueue and await the handles together:

use kafka_produce::{Producer, ProducerRecord};
async fn example(producer: &Producer) -> kafka_produce::Result<()> {
let mut pending = Vec::new();
for i in 0..10_000 {
    pending.push(
        producer
            .enqueue(ProducerRecord::new("orders").with_value(format!("{i}")))
            .await?,
    );
}

for delivery in pending {
    match delivery.await {
        Ok(meta) => println!("{}-{} @{}", meta.topic, meta.partition, meta.offset),
        Err(error) => eprintln!("not delivered: {error}"),
    }
}
Ok(())
}

enqueue returns as soon as the record is buffered; the Delivery resolves when the broker has answered for the batch the record travelled in. Dropping a Delivery is allowed and does not cancel the write — the record has been accepted, and only the result is discarded.

enqueue waits when the buffer is full, and that wait is the backpressure. Without the buffer_memory bound, a producer whose broker has stopped acknowledging accepts records until the process dies.

linger defaults to zero, and that is not a reason to raise it. A partition holds one batch on the wire at a time, so records arriving during a round trip accumulate into the next batch on their own. Batching scales with load rather than with the setting: an idle producer pays no latency, and a busy one batches anyway. Raise it only for a producer whose records arrive in bursts smaller than one round trip.

flush waits for everything buffered to be acknowledged. Per-record errors still belong to their own Delivery; flush reports only that the flush itself could not be carried out.

Where a record lands

The murmur2 implementation is the Java one, checked against rdkafka in the interop crate rather than against ourselves: a partitioner that is nearly Java's returns a partition for every key and passes any test written against our own reader — it just puts keys where a Java or C client would not look for them, which breaks co-partitioned joins and compacted-topic semantics silently and much later. partition_for_key is public if you need to compute the same answer outside a producer.

Durability

use kafka_produce::{Acks, ProducerConfig};

fn example() {
let config = ProducerConfig::new().acks(Acks::Leader);
}

Acks::Leader is lossy exactly once — if the leader fails before a follower replicates the record, the record is gone and the caller was told it arrived. It also acknowledges before the record is readable: a consumer reads only up to the high watermark, which does not advance until the ISR has the record, so there is a window where send has returned an offset that a scan of that partition will not yet show. Code that reads its own writes back wants Acks::All.

acks=0 is not offered

There is no None variant on Acks, so the mode cannot be selected and then fail at runtime. acks=0 is a request the broker sends no response to at all, and the connection actor correlates every in-flight request on a HashMap<i32, oneshot::Sender<_>> — an acks=0 produce would register a waiter nothing ever resolves, and every successful write would surface to the caller as a timeout. What the mode actually buys, not waiting on the leader, is what enqueue provides safely. The full argument is in the crate's code tour.

What happens when it fails

Two kinds of failure look similar and permit completely different things:

Collapsing the two is a bug in either direction. Retry everything without sequence numbers and you duplicate a record on every timeout, with no error anywhere; retry nothing and an ordinary leader election becomes a delivery failure.

Rejections are retried under ProducerConfig::retry after refreshing the metadata that made us ask the wrong broker. The delay is the point rather than the count — three immediate retries all re-read the same stale answer and fail identically.

Idempotence

On by default. The producer claims a producer id and numbers every record, so the broker recognises a re-sent batch and answers with the original offsets instead of appending it twice. That is what makes an ambiguous failure retriable, and it is why an ordinary leader election is something the producer rides out rather than something the caller sees.

ProducerConfig::idempotent(false) is for brokers that cannot issue a producer id. It does not make the producer faster; it makes it lossier.

Compression

use kafka_produce::{Compression, ProducerConfig};

fn example() {
let config = ProducerConfig::new().compression(Compression::Zstd);
}

None (default), Gzip, Snappy, Lz4, Zstd. Codec choice is per producer and applies to the whole batch; the consumer side needs no configuration, because the codec travels in the batch header.

Note the build implication rather than a runtime one: Lz4 and Zstd reach C through kafka-protocol's lz4-sys and zstd-sys, so a downstream build wants a C compiler. Gzip and Snappy are pure Rust. This is the one place "no C in the dependency tree" is not literally true, and it is stated rather than papered over.

Transactions

A transactional producer is always idempotent, and setting a transactional id changes what the producer is: it claims a fenced producer id, and claiming it fences any earlier producer holding the same id. That is the point of the id rather than a side effect — it is how a restarted application takes over cleanly from the instance it replaced.

use kafka_produce::{Producer, ProducerConfig, ProducerRecord};

async fn example(cluster: kafka_produce::Cluster) -> kafka_produce::Result<()> {
let producer = Producer::new(
    cluster,
    ProducerConfig::new().transactional_id("billing-writer-1"),
);

// Once, before any transaction: claims the id and fences the previous holder.
producer.init_transactions().await?;

producer.begin_transaction()?;
producer.send(ProducerRecord::new("orders").with_value("a")).await?;
producer.send(ProducerRecord::new("invoices").with_value("b")).await?;

match producer.commit_transaction().await {
    Ok(()) => {}
    Err(error) => {
        eprintln!("committing failed: {error}");
        producer.abort_transaction().await?;
    }
}
Ok(())
}

Things worth knowing before you rely on it:

• begin_transaction is local. The protocol has no "begin" request; the coordinator first learns of the transaction when the producer enrols a partition in it, which happens automatically on the first write to each.
• Commit and abort flush first. A record still sitting in the accumulator when the marker is written is not in the transaction, and would appear afterwards as an ordinary uncommitted write.
• Aborting deletes nothing. It writes a marker, so a read_committed reader never sees the records and a read_uncommitted one does. That asymmetry is the protocol's. Read them back with Visibility::CommittedOnly to see what a committed reader sees.
• PRODUCER_FENCED is terminal. Another producer with the same transactional id has bumped the epoch, and every later request of this one will fail identically. Retrying is an infinite loop; the correct response is to stop.

Configuration reference

Every setter is a consuming builder, so a config is one expression.

The batching defaults are Java's, deliberately: they are the numbers every operator's intuition is calibrated against.

A record accounted larger than max_request_size is refused at enqueue with MESSAGE_TOO_LARGE before it is buffered, so it fails alone rather than taking a batch with it. A record larger than batch_size is still sent, in a batch of its own — which is the only way it can be sent at all.

Note that delivery_timeout is a field in the request, honoured by the leader while it waits on its followers. It is not the connection's own request timeout, which bounds how long we wait for the socket.

Cancel safety

Dropping a send future does not cancel the write. Once the record has been accepted into the accumulator it will be sent, and dropping only discards the result. Dropping while still waiting for buffer space does cancel it, and in that case the record was never accepted.

That is the same rule the whole workspace follows — see Cancel safety.

Reading it back

Consuming records for a consumer, or Reading records for the browse-shaped scan and tail. A record written with Acks::All is readable as soon as send returns; one written with Acks::Leader may not be.

Consuming records

kafka-consume is the long-running read path: incremental fetch sessions, a streaming fetcher, and three ways of deciding which partitions you own.

[dependencies]
kafka-consume = "0.4"

Which shape do you want?

GroupConsumer and ClassicConsumer both wrap a Consumer rather than replacing it: the fetch path, the sessions and the decoding are identical, and the only thing membership changes is where the assignment comes from. That is why the manual mode is not a degraded group consumer — it is the same engine with the assignment supplied by the caller.

If what you actually want is a page of a topic rather than a stream of it, you want kafka-read instead. A scan is bounded and reports progress because a UI is drawing a page; a consumer runs until told to stop, and its interesting operations — seek, pause, resume — are about changing its mind mid-stream, which a bounded scan never does.

A manually-assigned consumer

use kafka_consume::{Consumer, ConsumerConfig, Position};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer = Consumer::new(cluster, ConsumerConfig::new());

consumer
    .assign(
        [("orders".to_owned(), 0), ("orders".to_owned(), 1)],
        Position::Earliest,
    )
    .await?;

loop {
    for record in consumer.poll().await? {
        println!(
            "{}-{} @{}: {:?}",
            record.topic, record.partition, record.offset, record.value
        );
    }
}
}

Consumer::connect takes bootstrap addresses if you have no Cluster yet.

assign replaces the assignment rather than adding to it, and partitions that leave it are forgotten in the next fetch — which is what stops the broker holding session state for partitions nobody is reading.

An empty poll is a normal answer, not an error. A consumer at the log end is caught up; it returns after max_wait_ms with nothing. A poll loop that treats empty as a failure is a poll loop that fails on every healthy cluster.

Records are decoded with the same tolerant decoder the scan path uses. A batch that will not decode does not stall the partition and does not end the stream: the position steps past it and polling continues. Unlike ScanEvent::Malformed, the consumer does not surface those bytes to you — Tolerant decoding covers the difference.

Changing its mind mid-stream

use kafka_consume::Consumer;
fn example(consumer: &mut Consumer) -> kafka_consume::Result<()> {
consumer.seek("orders", 0, 4_200)?;    // takes effect on the next fetch
consumer.pause("orders", 1);           // stop fetching, keep the partition
consumer.resume("orders", 1);          // continue from where it stopped

let next = consumer.position("orders", 0);   // Option<i64>
let behind = consumer.lag("orders", 0);      // Option<i64>, None until a fetch reports
Ok(())
}

A seek discards anything already buffered for that partition — a seek that still delivered the old records would not be a seek. A paused partition keeps its position and its place in the assignment, so resume does not re-resolve anything.

Offsets

A manually-assigned consumer can borrow a group's offset storage without joining the group:

use kafka_consume::{Consumer, ConsumerConfig, Position};

fn handle(record: &kafka_consume::Record) -> kafka_consume::Result<()> { Ok(()) }
async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer = Consumer::new(cluster, ConsumerConfig::new().group_id("reporting"));
consumer.assign([("orders".to_owned(), 0)], Position::Earliest).await?;

// Resume where the last run stopped, rather than where `assign` started.
consumer.seek_to_committed().await?;

for record in consumer.poll().await? {
    handle(&record)?;   // …and only then commit
}

for ((topic, partition), result) in consumer.commit().await? {
    if let Err(error) = result {
        eprintln!("{topic}-{partition}: commit failed: {error}");
    }
}

let stored = consumer.committed().await?;   // HashMap<(String, i32), CommittedOffset>
Ok(())
}

Three things about this that bite if you assume otherwise:

• A committed offset is the offset of the next record to read, not the last one handled. Storing the last record's offset re-delivers it forever. commit stores the consumer's current positions, which are already that.
• Borrowing the storage is not joining the group. The commit goes out anonymously, with an empty member id and the -1 non-member sentinel. The coordinator honours that form only while the group has no members — precisely so a detached client cannot scribble over a live group's positions. Point a standalone consumer at a group that has live members and every partition comes back UNKNOWN_MEMBER_ID. Group members commit as themselves; GroupConsumer::commit and ClassicConsumer::commit do that for you.
• The result is per partition. commit returns Vec<((String, i32), Result<()>)> — the same invariant as everywhere else in this library. An auto-commit whose result nobody checks is refused silently, so check it when it matters.

Do not read __consumer_offsets yourself. committed uses OffsetFetch; the internal topic format is not a stable interface.

Group membership (KIP-848)

use kafka_consume::{ConsumerConfig, GroupConsumer};

fn still_running() -> bool { true }
async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer =
    GroupConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders", "refunds"])
        .await?
        .auto_commit(true)                    // the default
        .instance_id("billing-pod-3");        // static membership, optional

while still_running() {
    for record in consumer.poll().await? {
        println!("{}-{} @{}", record.topic, record.partition, record.offset);
    }
}

consumer.leave().await?;
Ok(())
}

poll is the heartbeat. It beats when one is due, reconciles any new assignment, and only then reads. A member that stops polling — because it is doing slow work between batches, or because its task was descheduled — is a member the coordinator eventually evicts, and the partitions go to somebody else. Keep the loop tight and do slow work elsewhere.

Nothing is owned until the first heartbeat comes back with an assignment, so the first poll or two returning empty is expected rather than a symptom. assignment() says what is owned right now, and member_id() is the id this client generated for itself — KIP-848 inverts the classic protocol here, where the broker issues it.

instance_id makes the member static (KIP-345): a restart inside session.timeout.ms parks the assignment rather than triggering a rebalance, which is the difference between a rolling deploy that shuffles every partition and one that does not.

leave() releases the assignment — or parks it, for a static member — after running the rebalance listener and committing. Dropping the consumer without calling it leaves the group waiting out the session timeout before it notices.

The rebalance listener

Auto-commit flushes the offsets this crate tracks. A caller that keeps its own per-partition state — a windowed aggregate, a write-behind buffer, a file handle per partition — has state the library knows nothing about, and by the time poll returns the partition is gone and another member may already own it.

So the callback fires before the revocation takes effect, while this member still owns the partitions and the broker is still waiting for the acknowledgement that gives them away:

listener.on_revoke  →  auto-commit  →  drop the partitions  →  acknowledge

use futures::future::BoxFuture;
use kafka_consume::{ConsumerConfig, GroupConsumer, RebalanceListener, Result, RevokedPartition};

struct FlushOnRevoke;

impl RebalanceListener for FlushOnRevoke {
    fn on_revoke(&mut self, revoked: Vec<RevokedPartition>) -> BoxFuture<'_, Result<()>> {
        Box::pin(async move {
            for partition in revoked {
                // `position` is what auto-commit is about to store — write your
                // own checkpoint at the same offset, not at the last record's.
                println!(
                    "flushing {}-{} at {}",
                    partition.topic, partition.partition, partition.position
                );
            }
            Ok(())
        })
    }
}

async fn example(cluster: kafka_consume::Cluster) -> Result<()> {
let consumer =
    GroupConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders"])
        .await?
        .on_rebalance(FlushOnRevoke);
Ok(())
}

on_assign fires after the new partitions are owned and readable, and defaults to doing nothing — gaining a partition needs no protection.

Two properties to design around:

• on_revoke is at-least-once. Dropping a poll future must be safe, so a rebalance that has been computed but not finished is held on the consumer and retried by the next poll — which means a poll cancelled during on_revoke runs it again with the same partitions. A listener that flushes twice writes the same bytes twice; a listener that never fires loses them. Make it idempotent.
• An error from the listener does not stop the rebalance. By the time it runs, the group has already moved on and the broker is waiting for the acknowledgement. Refusing to revoke would leave this member holding partitions the group has given away — the double-ownership the revoke-then-acknowledge ordering exists to prevent. An Err is logged at warn and the rebalance proceeds.

Classic groups

Only for brokers older than 4.0, or a mixed group where a Java client is pinned to group.protocol=classic. GroupConsumer is the default on 4.x and is strictly less work.

use kafka_consume::{Assignor, ClassicConsumer, ConsumerConfig};

async fn example(cluster: kafka_consume::Cluster) -> kafka_consume::Result<()> {
let mut consumer =
    ClassicConsumer::subscribe(cluster, ConsumerConfig::new(), "billing", ["orders"])
        .await?
        .assignors([Assignor::CooperativeSticky, Assignor::Range]);

for record in consumer.poll().await? {
    println!("{}-{} @{}", record.topic, record.partition, record.offset);
}
Ok(())
}

Every member needs its own Cluster. That is not a style preference but a hard requirement of this protocol: JoinGroup blocks on the coordinator, and a Kafka broker will not read a second request from a socket until it has answered the first — so two members of one group sharing a connection deadlock, and it presents as a plain timeout with nothing in any log to explain it. GroupConsumer has no such constraint.

The advertised order is a vote, not a demand: the coordinator intersects every member's list, each member votes for the first of its own that survived, and the most-voted protocol wins. Advertising exactly one assignor forces the issue, at the cost of failing to join any group that does not share it — INCONSISTENT_GROUP_PROTOCOL, at join time, loudly.

Eager sticky is deliberately absent: StickyAssignor carries its state in the subscription's user_data as a struct with no schema in kafka-protocol, and hand-rolling a wire format is what this codebase does not do. cooperative-sticky has no such problem, so incremental rebalancing is available.

One protocol difference worth knowing for the listener: the classic protocol revokes eagerly, so every rebalance hands on_revoke the whole assignment rather than only the partitions that end up moving. That is what range-style rebalancing does; it is not this client rounding up.

Configuration

use kafka_consume::{ConsumerConfig, Visibility};

fn example() {
let config = ConsumerConfig::new()
    .group_id("reporting")
    .visibility(Visibility::All)
    .max_wait_ms(200);

// The rest are public fields on an owned type:
let tuned = ConsumerConfig {
    partition_max_bytes: 4 * 1024 * 1024,
    ..ConsumerConfig::new()
};
}

Note the default: a consumer is CommittedOnly, while a ScanSpec is Visibility::All. A consumer is usually a pipeline stage that should not see records a transaction abandoned; a scan is usually a human looking at what is actually in the log. read_committed does not mean the broker filters for you — it sends the records plus an AbortedTransactions list and the client does the work.

ConsumerConfig and ProducerConfig still use bare setter names (.group_id(…), not .with_group_id(…)). The rest of the workspace uses the with_ prefix, and these predate it; renaming them breaks callers, so it is a deliberate future pass rather than a trickle. See STYLE.md.

Fetch sessions, for free

Every consumer keeps a KIP-227 incremental fetch session per broker. The first request establishes the assignment and every request after it sends only what changed, which in steady state is nothing at all — so a consumer holding twelve partitions across three brokers sends three fetches per round, not twenty-four, and each carries almost no request body.

A broker that drops the session (a restart, or eviction under cache pressure) answers FETCH_SESSION_ID_NOT_FOUND or INVALID_FETCH_SESSION_EPOCH. Both are recovered by opening a new session with the full assignment, and neither is ever surfaced to you: a broker restart must not kill a consumer.

kafka-read's scan and tail deliberately keep the legacy sentinel and open no session, because they are one-shot and would otherwise leave broker state behind for a client that is not coming back.

Cancel safety

Dropping a poll future may discard a fetch that was in flight. It never advances a position for records you did not receive, so the worst case is re-fetching the same records. For a group member, a rebalance computed but not carried out is held on the consumer, so a dropped poll cannot skip the callback — the next one picks it up, still ahead of the acknowledging heartbeat.

See Cancel safety.

Reading records

Two APIs answering two different questions. Neither is a consumer — see The read path.

Both take a &Cluster. Get one from Cluster::connect, or from admin.cluster().

The tail — "what just happened"

The most-used view in any Kafka UI, and it is a backward walk rather than a forward read.

use kafka_read::TailSpec;

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
// Last 500 records per partition, all partitions.
let tails = kafka_read::tail(cluster, &TailSpec::new("orders", 500)).await?;

for tail in &tails {
    println!(
        "partition {}: {} records, {} malformed batches, {} fetches",
        tail.partition,
        tail.records.len(),
        tail.malformed,
        tail.fetches,
    );
}

// Or narrow it.
let spec = TailSpec::new("orders", 100).partitions([0, 3]);
let tails = kafka_read::tail(cluster, &spec).await?;
Ok(())
}

This returns a Vec rather than a stream, deliberately: you asked for a bounded number of records, and the implementation reads roughly that many bytes rather than the whole partition. On a compacted topic with large offset gaps it still converges — the step grows when a chunk yields fewer records than its offset span suggested.

The scan — "show me this topic"

use futures::StreamExt;
use kafka_read::{ScanEvent, ScanSpec, StartPosition};

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let spec = ScanSpec::new("orders")
    .from(StartPosition::Earliest)
    .partitions([0, 1, 2])
    .limit(10_000);

let mut stream = Box::pin(kafka_read::scan(cluster, spec).await?);
while let Some(event) = stream.next().await {
    match event? {
        ScanEvent::Record(record) => {
            println!("{}:{} {:?}", record.partition, record.offset, record.value);
        }
        ScanEvent::Progress(progress) => {
            if let Some(fraction) = progress.fraction() {
                println!("{:.0}%", fraction * 100.0);
            }
        }
        ScanEvent::Malformed { offset, last_offset, reason, .. } => {
            eprintln!("offsets {offset}..={last_offset:?} did not decode: {reason}");
        }
        _ => {}
    }
}
Ok(())
}

Box::pin because the returned stream is not Unpin.

Start positions:

Handle Malformed, do not ignore it

This is the point of the whole design. A batch that will not decode becomes an event carrying the offsets it covered and the raw bytes, and the scan continues.

use kafka_read::ScanEvent;
fn example(event: ScanEvent) {
match event {
    ScanEvent::Malformed { offset, last_offset, raw, reason } => {
        // Render "offsets 4102–4530 would not decode", offer the hex.
        // Do NOT abort the scan, and do NOT treat this as a transport error.
    }
    _ => {}
}
}

Granularity is a batch, not a record — a corrupt record takes its batch with it, bounded by max.message.bytes. That was a deliberate choice, not a limitation discovered late; see Tolerant decoding.

What you will not see as Malformed: a batch truncated by max_bytes (normal on every fetch), control batches (transaction markers), or aborted records under CommittedOnly. Those are filtered silently, because reporting them means crying wolf on every fetch of every healthy cluster.

Transactions and visibility

use kafka_read::{ScanSpec, Visibility};

fn example() {
// Default: read_uncommitted. Aborted records are visible.
let all = ScanSpec::new("orders").visibility(Visibility::All);

// read_committed. Aborted records filtered client-side.
let committed = ScanSpec::new("orders").visibility(Visibility::CommittedOnly);
}

read_committed does not mean the broker filters for you — it sends the records plus an AbortedTransactions list, and the client does the work. That is Kafka's design, not a shortcut here.

Filtering

use kafka_read::{RecordFilter, ScanSpec};
fn example(filter: RecordFilter) {
let spec = ScanSpec::new("orders").filter(filter);
}

RecordFilter runs client-side, after decoding — Kafka has no server-side filtering, so a filter reduces what you iterate, not what crosses the network. Use partitions and limit to reduce bytes; use the filter to reduce noise.

Cancelling

Drop the stream. That is the whole protocol.

Dropping mid-scan releases the buffer, drops the in-flight fetch futures, and leaves every connection consistent — no half-read responses, nothing to unwind. See Cancel safety.

use futures::StreamExt;
async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let spec = kafka_read::ScanSpec::new("orders");
let mut stream = Box::pin(kafka_read::scan(cluster, spec).await?);
while let Some(event) = stream.next().await {
    // …stop whenever you like; just drop it
    break;
}
Ok(())
}

Memory

Bounded by ScanSpec::max_buffered_records (default 10,000) across the whole scan, not per partition. Scanning a thousand-partition topic uses the same budget as scanning one — which is the difference between a UI backend that survives a large cluster and one that does not.

Lowering it tightens memory at the cost of cross-partition ordering: a smaller buffer forces more emits before every partition is represented, which widens the bounded reorder window that ScanEvent::Progress reports.

Testing against a real cluster

The container acceptance suite owns everything needing a broker configured a particular way, killed mid-request, or fed a damaged log segment. It cannot tell you what happens against a cluster that is shared, long-lived and not yours, running a Kafka build you did not choose, holding data written by clients you did not write.

That is what livetest is for, and it is not a nice-to-have: the first live run against Strimzi found six protocol bugs that every unit test passed straight over.

The commands

cargo build -p livetest

livetest probe    # read-only inventory + negotiated version table. Touches nothing.
livetest smoke    # admin round trip: create, describe, alter, verify, delete
livetest read     # scan and tail real topics, asserting the decoder
livetest sweep    # delete anything this tool left behind

read takes options:

livetest read --topic orders --expect 15000
livetest read --max-topics 5 --limit 20000

Configuration

Everything comes from the environment:

The live-cluster skill

This repository ships a live-cluster skill that resolves all of the above from Kubernetes, so you do not assemble them by hand:

eval "$(.claude/skills/live-cluster/resolve-target.sh strimzi)"
cargo run -q -p livetest -- probe

resolve-target.sh <strimzi|kaas> [plain|tls|authed] reads the Kafka CR status or the Service and prints export lines. For a TLS listener it also extracts the cluster CA into a temp file and points KAAS_TEST_CA_FILE at it. It never resolves credentials.

The two targets

When a run fails against kaas and passes against strimzi, the default conclusion is that kaas is incomplete, not that kaas-lib is broken. Check the version table first: kaas advertises roughly half the api keys Strimzi does. See Non-goals for why the two projects share no code — this diff is exactly the check that separation exists to enable.

probe is a conformance diff

Its output is sorted and diffable, with notes on stderr and facts on stdout:

eval "$(.claude/skills/live-cluster/resolve-target.sh strimzi)"; livetest probe > strimzi.txt
eval "$(.claude/skills/live-cluster/resolve-target.sh kaas)";    livetest probe > kaas.txt
diff strimzi.txt kaas.txt

That diff is the parity check. livetest probe > out.txt captures exactly the body and nothing else, which is why the stream split exists.

Safety on a shared cluster

Everything created is prefixed, and sweep refuses to touch a name without the prefix. Running a tool that creates topics against a cluster other people depend on is only acceptable if cleanup is mechanical and cannot over-reach.

For a cluster you must not modify at all:

export KAAS_TEST_READ_ONLY=1
livetest probe

That turns on the read-only gate — mutating api keys are refused before a socket is opened — and the target additionally refuses any command that would need to create something, with an error naming the environment variable rather than a permission failure from the broker.

Always run sweep after a smoke run, including a failed one.

Reading a failed run

The report is printed before the pass/fail result, even when the run failed. A partial report says how far it got and what the cluster looked like on the way, which is the entire diagnostic value — a protocol mismatch against a cluster you cannot attach a debugger to is not something a stack trace will explain.

read ranks topics by record count, so a run against a production-shaped cluster exercises the topics most likely to stress the decoder rather than whichever five sort first alphabetically. A compression bug, a header encoding bug or a topic-id bug shows up in the topic with fifteen million records, not in the empty one.

Roadmap

Phase 2 has landed. The library was admin-first with a browse-shaped read path; it now has a real producer and both consumer-group protocols, which is what "general-purpose Kafka client" was shorthand for.

The milestone breakdown with acceptance criteria lives in PLAN.md (M0–M19). This page is what shipped, what is deliberately still missing, and what nothing in this repository can move.

What phase 2 delivered (M12–M19)

The producer — kafka-produce.

Both traps this page flagged before M12 was written got resolved rather than discovered:

• acks=0 is not offered. A request with no response would leave a pending oneshot in the connection actor forever, so every successful write would report a timeout. It is refused at the config boundary rather than given a fire-and-forget path.
• The max_in_flight warning turned out to be the wrong worry. At most one batch per partition is on the wire regardless, so ordering does not depend on the setting at all. The clamp — one without idempotence, five with — is defence for the connection layer, not the mechanism keeping the log in order.

The consumer — kafka-consume.

M18 was conditional on the classic protocol being needed, and it was: the acceptance suite runs a mixed group with one Rust member and one kafka-console-consumer.sh, which is the case that makes byte-compatible assignor payloads non-optional.

Two gaps found after the fact and since closed: a caller had no way to flush per-partition state before revocation, so on_rebalance now runs on_revoke while the member still owns the partitions and before the auto-commit; and the classic path advertised only eager assignors, so cooperative-sticky joined them.

Publishing. The crates are on crates.io, releasing in lockstep at a single version — see RELEASING.md. kafka-consume joins the published set at 0.3.0.

Next

Nothing here is structural. These are ordinary gaps with no blocker beyond someone doing them.

• Java's StickyAssignor. The classic path ships three of the four assignors PLAN.md lists — range, round-robin and cooperative-sticky. Plain sticky is the eager one that keeps assignments stable across a rebalance without the two-round handover, and it is the remaining name a mixed group might vote for.
• KIP-699 batched FindCoordinator. The v4+ coordinator_keys shape is already used, but with one key per request. Batching is one round trip instead of one per group, which matters for a UI rendering hundreds.
• DescribeQuorum — the one KRaft-adjacent API a cluster UI plausibly wants. Present in the ApiKey enum and reachable through generic dispatch; there is no typed method.
• Delegation token management. Only the ACL resource type exists today; Create/Renew/Expire/DescribeDelegationToken do not.
• OffsetForLeaderEpoch, UpdateFeatures, ListConfigResources — routing entries only, no user-facing surface.

Blocked upstream

Not roadmap items, because nothing in this repository can move them. See The upstream schema gap:

• Streams groups (KIP-1071) — no schema in kafka-protocol 0.17, so a 4.1+ cluster running Kafka Streams reports groupType=streams in ListGroups and we surface it as Unrecognized rather than describing it.
• ListOffsets -6 (EARLIEST_PENDING_UPLOAD_TIMESTAMP) — needs v11; the codec caps at v10. The other five sentinels are surfaced.
• Error codes past Kafka 4.1 — surfaced as Unknown(i16) until upstream names them.