Introduction

kaas is a from-scratch, Apache Kafka 3.7 wire-compatible broker built to run on Kubernetes — clients and tools connect unchanged, while Kubernetes primitives and a shared filesystem replace Kafka's own distributed-systems machinery.

If you know Apache Kafka, you already know how to use kaas. Java clients, librdkafka, franz-go, Kafka Streams applications, and the stock shell tools (kafka-topics.sh, kafka-console-producer.sh, …) all talk to it as if it were a Kafka 3.7 cluster. What changes is everything behind the socket — and this book is organised around exactly that: what stays the same (the wire contract, proven in Part II) and what is replaced (the machinery, explained in Part I).

One idea, three substitutions

Kafka's hardest operational problems — quorums, replication, partition movement — follow from one assumption: brokers own local disks and must protect the data on them. kaas drops that assumption. Brokers are (nearly) stateless pods; partition data lives on shared ReadWriteMany volumes that every broker mounts; durability is the storage layer's job and coordination is Kubernetes' job. That single move replaces three pieces of Kafka machinery, and the whole book keeps referring back to them:

1. KRaft (or ZooKeeper) → a Kubernetes Lease. The controller is whichever broker currently holds the kaas-controller Lease. There is no quorum and no replicated state machine; the Kubernetes API server is the metadata store.
2. Replication & ISR → a single writer per partition. Exactly one broker leads — and writes — each partition at a time, fenced by epochs so a deposed leader cannot corrupt the log. There are no followers: acks=all is an fsync to the shared volume, not N replicas, and a broker "takes over" a partition by opening its files, not by copying data.
3. Internal topics → files on the shared volume. Consumer offsets (__consumer_offsets) and transaction state (__transaction_state) are plain JSON files in a cluster-state directory instead of replicated internal topics.

Each substitution is a real trade. Non-goals states plainly what you give up; Part I walks through how each replacement works and why it is safe on the storage kaas targets.

A translation table

Where the things you manage in Kafka live in kaas:

What ships

Two binaries and a Helm chart: the broker, an operator that reconciles the custom resources into on-disk state and Kubernetes plumbing (and stays entirely off the produce/fetch hot path), and the chart that deploys both. The release line is v0.3.x-preview — pre-v1, kaas makes no backwards-compatibility promises between releases; see Releasing for the exact upgrade contract.

The parity target

kaas targets Apache Kafka 3.7 for wire-protocol and Kafka Streams parity — and the book is structured to prove that claim rather than assert it. Part II carries a generated API matrix that CI pins to the actual wire surface, a KIP index that says implemented / partial / non-goal honestly, and a verification story built on Apache's own shell tools, run against every release.

How to read this book

• Evaluating kaas? Read this page, then Non-goals — the fastest honest answer to "can it replace my cluster" — then Getting Started to try it.
• Operating it? Getting Started, then Part I in order starting at the system overview, then Part IV for the chart and the storage substrate requirements your filer must meet.
• Contributing? Parts I and II are the semantics; Part III is the crate-by-crate tour of where they live in the source.

Getting Started

Deploy kaas onto a Kubernetes cluster with the Helm chart, or run a single broker locally in dev mode with in-memory storage.

Deploy on Kubernetes

Prerequisites: Kubernetes ≥ 1.27, Helm ≥ 3.8, and — for more than one broker — a ReadWriteMany StorageClass with NFSv4-class semantics (what "NFSv4-class" means precisely, and which providers qualify, is Storage substrate requirements). A single-broker cluster can run on a plain ReadWriteOnce local-path class.

helm install my-kaas oci://ghcr.io/kaas-rs/charts/kaas \
  --version 0.3.1-preview \
  --namespace kafka --create-namespace \
  --set storage.className=<your-rwx-class> \
  --set broker.replicaCount=3

That deploys a broker StatefulSet, the operator, and the shared volume(s). The chart's defaults give you an anonymous in-cluster listener on 9092; TLS, SCRAM, OAuth (SASL/OAUTHBEARER against an OIDC issuer, disabled by default on 9096), and external access are chart values — see Helm chart & listener configuration.

Create a topic

Topics are Kubernetes resources:

apiVersion: kaas.rs/v1alpha1
kind: KafkaTopic
metadata:
  name: test
  namespace: kafka
spec:
  partitions: 3

If you'd rather stay in Kafka's own tooling, that works too — kafka-topics.sh --create (or any Admin-API client) creates the same KafkaTopic resource for you, so both routes end in one place and kubectl get kafkatopics always shows the truth.

Retention is enforced: a topic without an explicit retention config ages out after 7 days (Kafka's own default) — set retentionMs: -1 to keep data forever.

Talk to it

Any Kafka client, unchanged:

kubectl -n kafka port-forward svc/my-kaas-kaas 9092:9092 &
echo "hello" | kcat -b localhost:9092 -t test -P
kcat -b localhost:9092 -t test -C -o beginning -e

From here, the system overview explains what you just deployed — which pod does what, and where your bytes actually went.

Run locally in dev mode

The broker binary detects dev mode by the absence of the MY_POD_NAME env var (which the StatefulSet always sets): storage flips to in-memory, no Kubernetes API is needed, and the broker treats itself as leader of every partition.

cargo run -p kaas

Point any Kafka client at localhost:9092. Nothing is persisted and nothing is cluster-aware — it's a protocol-correct scratchpad for client development and codec work, not a single-node production mode.

Build the book and the code

cargo build --workspace          # toolchain is pinned; rustup auto-installs
cargo test  --workspace --all-features
cargo xtask docs --serve         # this book, live-reloading

System overview

The moving parts at a glance: broker pods, the operator, the shared RWX volume, the Kubernetes API — and where Apache Kafka clients plug in.

A production Apache Kafka cluster is brokers plus distributed-systems machinery: a KRaft controller quorum (ZooKeeper before it) electing leaders and replicating a metadata log, ISR-tracked replicas guarding every partition, and internal topics carrying offsets and transaction state. kaas keeps Kafka's wire contract — Apache Kafka clients (Java, librdkafka, franz-go) connect unchanged, verified against the Kafka 3.7 parity matrix (see Part II) — but replaces that machinery with Kubernetes primitives and a shared filesystem. Three substitutions carry the whole design:

1. KRaft / ZooKeeper quorum → a Kubernetes Lease plus custom resources. Controller election is a Lease; topics, users, and ACLs live in CRs instead of a metadata log.
2. Replication / ISR → a single writer per partition on a shared ReadWriteMany volume. Durability comes from the storage substrate, not from followers, and epoch fencing guards against split brain where ISR membership would.
3. Internal topics (__consumer_offsets, __transaction_state) → plain JSON files on the same shared volume.

Kubernetes is the only control plane — there is no peer gossip protocol and no replicated state machine. The full rationale for each divergence lives in Non-goals.

flowchart TB
    k8s["Kubernetes API<br/>Leases · CRDs · Services · RBAC"]
    clients["Apache Kafka clients<br/>Java · librdkafka · franz-go<br/>Produce / Fetch / Metadata / SASL …"]

    subgraph deployment["kaas deployment"]
        operator["kaas-operator<br/>(Deployment, 1 replica)"]
        subgraph brokers["kaas brokers (StatefulSet, N replicas)"]
            b0["kaas-0<br/>controller — holds the<br/>kaas-controller Lease"]
            b1["kaas-1"]
            b2["kaas-2"]
        end
        pvc[("Shared RWX PVC — NFSv4<br/>/data/__cluster/<br/>assignment.json · credentials.json · acls.json<br/>txn_state/ · producer_fences/ · marker_queue/<br/>__consumer_offsets/<br/>/data/&lt;topic&gt;/&lt;partition&gt;/<br/>segments · manifest.json · producer-state.snapshot")]
    end

    operator -- "reconcile CRs" --> k8s
    brokers -- "watch Lease + CRs" --> k8s
    operator -- "writes credentials.json, acls.json,<br/>partition dirs" --> pvc
    brokers -- "append / read segments,<br/>assignment.json" --> pvc
    b1 -- "heartbeat gRPC :9094" --> b0
    b2 -- "heartbeat gRPC :9094" --> b0
    clients -- "Kafka wire protocol,<br/>per-listener ports" --> brokers

The brokers

A StatefulSet with stable pod ordinals (kaas-0, kaas-1, …). Each pod is a single broker process that:

• serves client traffic on its listeners — the Helm chart declares one per listeners[] entry, each with its own TLS and authentication settings;
• serves peer heartbeats on :9094 (gRPC, controller-bound);
• exposes /healthz + /readyz on :8080 (kubelet probes + diagnostics);
• mounts the shared RWX volume at /data — every broker sees every other broker's segment files, which is what makes leadership takeover a file open, not a data copy.

The operator

A Deployment, single replica, leader-elected. It reconciles three CRDs into on-disk config files and Kubernetes plumbing:

The operator does not sit on the data path: brokers serve traffic even if the operator is crash-looping. Why that holds is the subject of Broker/operator runtime independence.

The shared substrate

NFSv4 in production (csi-driver-nfs or similar), local-path for single-node dev. kaas asks three things of the filesystem, and leans on each in a specific place:

1. Same-directory rename atomicity — the manifest and every cluster file are written tmp + fsync + rename.
2. Exclusive-create atomicity (open with O_CREAT|O_EXCL) — exactly one racer wins where a file is the lock.
3. Close-to-open consistency — a transaction-state file written and closed by one broker reads back complete on the next broker that opens it.

On top of the three, fsync must be honest: the group-commit cycle's sync_all() is the acks=all promise, so a substrate that lies about durability breaks everything.

The full contract — what those three guarantees do and don't buy, and the rules for code that touches the volume — is The RWX substrate contract; storage requirements and the provider matrix are covered in Operations.

Reading order

Part I follows the three substitutions.

Start with Broker/operator runtime independence — the ground rule that shapes everything else: the operator provisions, brokers serve, and the hot path never waits on Kubernetes. Controller, leases & assignment.json then covers the first substitution: how a Lease election plus one JSON file replace the KRaft quorum as the source of partition leadership.

The next four chapters are the second substitution — the data plane. Storage engine hot path shows how a broker makes acks=all affordable on networked storage; File-handle ownership how leadership moves between brokers as a file open; The RWX substrate contract the three filesystem guarantees all of it stands on; The volume pool how the substrate itself scales out.

Consumer-group coordination and Transactions & idempotence are the third substitution: what Kafka keeps in __consumer_offsets and __transaction_state, kaas keeps in JSON files on the shared volume.

The closing chapters face the operator of the cluster: Listeners, authentication, authorization, Kubernetes integration, Honest readiness & rollout pacing, and Observability.

Implementation notes (for contributors)

• The broker binary is bins/kaas; the operator is bins/kaas-operator.
• Listeners reach the broker as the KAAS_LISTENERS JSON env var; the chart synthesizes one entry per .Values.listeners[] item (gh #126).
• Fixed ports: 8080 health, 9094 inter-broker heartbeat gRPC.

Broker/operator runtime independence

Why the operator is a startup/admission component, not a hot-path dependency — the Produce/Fetch path makes zero Kubernetes API calls.

This is the most important architectural fact in kaas, and the easiest one to misread from the deployment layout: kaas ships a broker and an operator, so it looks like a classic operator-managed system where the operator sits in the middle of everything. It doesn't. The relationship mirrors one you already know from Apache Kafka, where a broker that loses sight of the controller keeps serving from the metadata it has — it just stops learning anything new. In kaas, the operator and the Kubernetes API server behind it play that role. The operator is a startup and admission component:

• Brokers read KafkaTopic CRs at startup and watch them for new topics and partition expansion — but the read is non-fatal. A missing or unreachable API server only blocks new topic creation; existing topics keep serving.
• The Produce/Fetch hot path makes zero Kubernetes API calls. Ownership lookups are in-memory, against the broker's view of assignment.json.
• Authentication and authorization state (credentials.json, acls.json) is read from the shared volume with hot-reload — the operator writes those files when KafkaUser CRs change, but a broker never asks Kubernetes a question to authenticate a client.
• Brokers serve traffic while the operator is crash-looping, upgrading, or deleted. What degrades without the operator: new KafkaTopic/KafkaUser CRs stop being materialized, and external-listener plumbing (Certificates, TLSRoutes) stops reconciling. What does not degrade: every already-created topic, credential, and ACL.

One honest caveat, and it is not a Kubernetes dependency: an OAUTHBEARER listener validates tokens against signing keys fetched from the OAuth issuer — an external service — and is fail-closed before the first successful fetch. An unreachable issuer blocks new authentications on that listener; every other listener, and every already-authenticated connection, is unaffected.

The one Kubernetes dependency on the control path

Brokers do keep a few long-lived Kubernetes watches: the kaas-controller Lease (controller election), the KafkaTopic CR watch (topic catalog), and the headless-Service endpoint watch (peer discovery). All of them feed the control plane — assignment recomputation — not the data plane. If the API server goes away, the current controller keeps its Lease view, assignment.json stays where it is, and partition leadership simply stops changing until the API server returns. Clients notice nothing.

"Until the API server returns" is load-bearing, and it costs the topic watch real machinery to honour. Kubernetes ends watch streams for entirely routine reasons — a relist, an API-server rollout, a network blip — so the watch rebuilds its stream with exponential backoff instead of treating stream end as completion; a watch that exits on the first routine disconnect stops tracking topics permanently, with no error to log. The watch also treats every relist as a full reconcile: the topic set the fresh stream reports is diffed against what the watch last knew, and anything missing is retracted. Without that diff, a topic deleted while the watch was disconnected would never produce a delete event and would linger forever — the broker serving Metadata for a topic that no longer exists, and the controller assigning its partitions to brokers that then fail to open them.

Admin writes go through CRs — deliberately

The broker isn't strictly read-only against Kubernetes: the Kafka admin APIs are served by writing CRs, so that the operator remains the single materializer of topic and user state:

• Topic admin APIs work the KafkaTopic CR: CreateTopics (and Metadata-driven auto-creation) mints one, CreatePartitions patches spec.partitions, IncrementalAlterConfigs patches spec.config per key, DeleteTopics deletes it, and AlterReplicaLogDirs records a partition move in the CR's status.
• User admin APIs edit an existing KafkaUser CR: CreateAcls/DeleteAcls mutate spec.authorization.acls, and the SCRAM admin API (KIP-554) patches spec.authentication.scram.

The operator then materializes the change — partition directories, .config.json, credentials.json, acls.json — exactly as if you had edited the CR yourself. This keeps one writer for on-disk state while still serving the Kafka admin surface. (It's also why broker RBAC carries the full verb set on kafkatopics — including create and delete — plus kafkatopics/status and read/update,patch on kafkausers; see Kubernetes integration.)

The line not to cross

If a change adds a broker→operator runtime dependency — a CR watch that blocks request handling, a reconcile the hot path waits on — that's an architectural change, not an implementation detail. The invariant to preserve: a broker that has already started serves Produce/Fetch with the Kubernetes API server unreachable.

Implementation notes (for contributors)

• Admin CR writes route through crates/kaas-broker/src/topic_cr_writer.rs (topics), crates/kaas-broker/src/acl_cr_writer.rs (ACLs), and crates/kaas-broker/src/user_cr_writer.rs (KIP-554 SCRAM rotation, gh #252).
• The self-restarting, relist-reconciling topic watch is run_topic_watch (gh #202): backoff 1 s → 30 s, reset on any event; the relist diff keys off the Event::InitApply topic set and only returns Ok(()) on cancellation.

Controller, leases & assignment.json

Controller election via a Kubernetes Lease, and assignment.json on the shared volume as the single source of truth for partition leadership.

In Apache Kafka, the controller is elected by the metadata quorum — KRaft today, ZooKeeper before it — and partition leadership reaches the brokers through a replicated metadata log. kaas replaces that entire machine with two much smaller parts (the first of the three substitutions): election is a Kubernetes Lease, and propagation is a JSON file on the shared volume.

The "controller" is just a broker holding the kaas-controller Lease — there is no separate process and no Raft quorum. The Lease's leaseTransitions counter is the cluster's epoch source: it increments exactly when the holder changes, and a releasing controller re-sends it so the epoch fence never rewinds.

sequenceDiagram
    participant L as Kubernetes Lease<br/>kaas-controller
    participant C as kaas-0<br/>(controller)
    participant A as assignment.json<br/>/data/__cluster/
    participant B as kaas-1<br/>(peer broker)

    B->>C: heartbeat gRPC :9094<br/>bidi stream, 1 s PING cadence
    C->>L: acquire (server-side apply)<br/>holderIdentity = kaas-0,<br/>leaseTransitions +1 on takeover
    L-->>C: epoch = leaseTransitions
    Note over C: recompute triggers:<br/>first Lease win · KafkaTopic change ·<br/>broker join/leave (2 s alive-set poll)
    C->>C: balancer: partition +<br/>consumer-group assignments
    C->>A: write tmp + fsync + rename<br/>{controller_epoch, assignment_version,<br/>brokers (alive/draining/dead),<br/>partitions, consumerGroups}
    C-->>B: heartbeat push: ASSIGNMENT_CHANGED
    B->>A: re-read (1 s mtime poll,<br/>push is the fast path)
    B->>B: reject if controller_epoch<br/>< Lease epoch<br/>(stale-controller fence)
    B->>B: partition takeover diff:<br/>take over → open FDs + recover<br/>relinquish → close FDs
    B->>B: consumer-group takeover diff<br/>+ orphan sweep

The file on the volume is where you read the assignment (from any broker pod, or via kafka-topics.sh --describe on the client side) — there is no Kubernetes-object reflection of it. There is no per-partition Lease either: the singleton controller Lease is the only Kubernetes coordination primitive, and everything downstream of it travels through assignment.json on the shared volume.

What the controller does

The Lease holder takes on four extra responsibilities:

• Observes peer brokers via the heartbeat gRPC stream every broker dials into it. A broker that stops heartbeating ages out of the alive set, and a broker shutting down cleanly announces itself first: the drain flag set at SIGTERM rides its next heartbeat, so the controller moves its partitions while it is still healthy enough to hand them over, rather than waiting for a timeout.
• Computes assignments — partition leadership over the alive set, and consumer-group placement over the full registered broker set (alive, draining, and dead rows alike): the group-coordinator hash divides by that set, so dead rows are retained deliberately — see Broker fencing.
• Writes assignment.json, epoch-prefixed, tmp + fsync + rename. Every broker rejects an assignment whose epoch is stale, so a deposed controller coming back from a GC pause can't roll the cluster backwards. Since the fencing work the file also carries every registered broker's health tri-state (alive / draining / dead), which is what DescribeCluster v2 reports as fenced brokers.

When it recomputes

The alive set the balancer feeds on is the set of heartbeat-connected brokers that report themselves healthy and not draining — a broker's own 1 s liveness tick, trusted unconditionally, minus anyone who has announced shutdown. The one exception is self-preservation: the controller always pins itself into the set, so a cluster can never compute an empty assignment out from under itself. Kubernetes endpoint readiness is only the bootstrap fallback for a freshly elected controller that no broker has dialed into yet, so a controller elected mid-rollout doesn't compute an empty assignment. How a broker earns — and loses — its place in the alive set is the subject of Honest readiness & rollout pacing.

How partitions get placed

Apache Kafka decides placement once, when a topic is created, and the answer lives in the metadata log forever. kaas has no such log: the assignment is recomputed from scratch every time an input changes. That buys self-healing — a partition on a departed broker is simply somewhere else in the next file — but it means the balancer has to choose to be stable, because nothing outside it remembers the last answer.

It does that by deciding as little as possible:

1. Keep every partition whose current leader is still alive.
2. Place only what's left, giving each new partition to the broker holding the fewest partitions of that topic.
3. Even out the brokers, cluster-wide, until no broker leads more than one partition more than any other.
4. Even out each topic across brokers, by trading partitions between brokers rather than moving them one way — so the balance from step 3 survives untouched.

Step 1 comes first for a reason. Deriving the whole layout and then noticing what didn't change gets the same answer in a steady state, but it makes every recompute a fresh opinion about every partition — so creating one topic could hand a dozen unrelated partitions to different brokers. Each of those is a genuine cost here: the new leader opens the log, replays the tail, and the old leader keeps acknowledging writes until it notices it's been replaced.

Step 4 is separate from step 3 because an even cluster and an even topic are different properties. Three brokers each leading 25 partitions look perfectly balanced, and a 3-partition topic can still have all three of its partitions on one of them — every producer and consumer for that topic talking to one broker while the other two sit idle. Apache gets this for free by assigning each topic round-robin from its own starting offset; kaas has to ask for it explicitly.

Two rules keep the moves cheap. Partitions that were just placed move before partitions inherited from the previous assignment, since the first cost nothing and the second cost a takeover. And the per-topic pass swaps rather than moves — one partition each way between two brokers — so it can never disturb the cluster-wide balance it runs after.

How peers follow

Non-controller brokers watch assignment.json via file notification plus a 1 s poll; the heartbeat stream's ASSIGNMENT_CHANGED push is the fast path, the poll the backstop. On every accepted assignment the broker diffs the new leadership map against what it currently serves, opening or relinquishing partitions in the storage engine to match (see File-handle ownership), and does the same for consumer groups (see Consumer-group coordination).

Everything that needs a leadership answer — the Metadata response, the Produce/Fetch ownership check, /healthz's partitions_led — sources from the broker's view of assignment.json. There is no second authority to disagree with.

Local-dev mode

When the broker starts outside a pod (the MY_POD_NAME env unset), the cluster runtime isn't started at all: storage flips to in-memory and a local shim answers "yes, I lead" for every partition. This is a dev-loop convenience, not a single-node production mode — nothing is persisted.

Implementation notes (for contributors)

• Controller-side logic lives in crates/kaas-controller: heartbeat_server.rs (serves proto/heartbeat.proto), balancer.rs (assignment computation), assignment_writer.rs (epoch-prefixed write).
• Recompute wiring — the topic-watch callback (gh #74) and the 2 s broker-set watcher (gh #77) — is in bins/kaas/src/cluster.rs.
• The broker-side assignment watcher and stale-epoch rejection live in crates/kaas-broker/src/coordinator.rs; making assignment.json the single leadership authority was the gh #75 cleanup. The deposed-controller race is pinned down by crates/kaas-controller/tests/stale_controller_race.rs.
• Dev-mode selection is in bins/kaas/src/main.rs; the always-leader shim is crates/kaas-broker/src/local_lease.rs.

Storage engine hot path

Group-commit fsync, segment files, the manifest — and how a Produce and a Fetch request travel the broker end to end.

If you know Apache Kafka, you know its produce path leans on two comfortable assumptions: writes land in the OS page cache, and durability comes from replication — acks=all means "on enough in-sync replicas", not "on disk", which is why almost nobody sets log.flush.interval.messages in production. kaas has neither assumption available. There are no followers: the only copy of a partition lives on a shared ReadWriteMany volume (the ground rules for that bet are the RWX substrate contract), so acks=all has to mean fsynced to that volume — and on an NFS-class mount, every fsync is a COMMIT round-trip over the network. That round-trip is the dominant cost in the whole broker, and the hot path is shaped around issuing as few of them as possible: one fsync per group of concurrent batches, an index that is never fsynced on the hot path, and a manifest that is almost never rewritten.

A Produce request, end to end

sequenceDiagram
    participant C as Client
    participant L as Listener
    participant H as Produce handler
    participant P as Partition
    participant K as Committer task

    C->>L: Produce (RecordBatch bytes)
    L->>L: read_frame → decode_request_header
    L->>H: dispatch by api_key<br/>(per-listener pre-auth gate)
    H->>H: topic exists? · do I lead it?<br/>heartbeat fresh (self-fence)? · ACL Write?
    Note over H: any gate fails →<br/>NOT_LEADER_FOR_PARTITION (6) /<br/>TOPIC_AUTHORIZATION_FAILED (29)
    H->>P: engine.append(topic, partition, epoch, acks, bytes)
    activate P
    Note over P: under the partition mutex
    P->>P: sticky flush_err? · epoch fence
    P->>P: idempotence classify (PID, epoch, seq)<br/>duplicate → echo cached base_offset, no write
    P->>P: segment full? → roll_fast
    P->>P: rewrite baseOffset → HWM,<br/>append_batch (pwrite, bytes verbatim)
    P->>P: snapshot.store (ArcSwap) — lock-free readers
    P->>P: pending ≥ flush_interval_messages?<br/>→ requested_flush_seq += 1
    deactivate P
    P-->>K: flush request — capacity-1 channel, coalesces
    opt acks = -1 and this append crossed the flush threshold
        Note over P: all concurrent appenders to this<br/>partition park on the same Notify —<br/>one fsync cycle serves them all
        K->>K: lock: target = requested_flush_seq
        K->>K: spawn_blocking → lock + log.sync_all()<br/>30 s watchdog → sticky Stalled on timeout
        K-->>P: completed_flush_seq = target,<br/>notify_waiters()
    end
    H-->>C: ProduceResponse(base_offset,<br/>throttle_time from one post-append quota check)

The RecordBatch bytes are never parsed on this path — only two fixed-size header peeks (producer info for idempotence, offsets for assignment); the same opaque bytes the client sent land verbatim on disk. The quota check runs once per request over the summed byte count, after the appends, and feeds throttle_time_ms.

A Fetch request, end to end

sequenceDiagram
    participant C as Client
    participant H as Fetch handler
    participant E as Storage engine

    C->>H: Fetch (decode + dispatch, same front door as Produce)
    H->>H: topic exists? · do I lead it? · ACL Read?
    Note over H: read cap = last stable offset (read_committed)<br/>or high watermark (read_uncommitted)
    H->>E: read(topic, partition, fetch_offset, max_bytes)
    E->>E: walk closed segments + active,<br/>copy batch bytes into memory
    E-->>H: Bytes (opaque, undecoded)
    H->>H: trim_to_offset(read_cap) — whole batches only
    H->>H: aborted_transactions_in_range<br/>→ AbortedTransaction list (read_committed)
    H-->>C: FetchResponse(session_id = 0,<br/>records bytes verbatim)

Two response-shape facts:

• Stateless fetch sessions: session_id = 0 on every response — Apache's documented contract for "broker doesn't support sessions", so clients fall back to full Fetch data per request. KIP-227 incremental sessions are a future optimisation, not a correctness gap.
• The response is materialized bytes, copied from the segment files; a sendfile/splice zero-copy path is a future optimisation (the codec keeps records byte-opaque exactly so that a splice path stays possible).

Concurrency model: inside one Partition

What runs under the partition mutex, what the per-partition committer task does, and what segment roll defers to a background task:

flowchart TB
    subgraph appender["append() — any producer connection"]
        direction TB
        lock["take the partition mutex"] --> classify["sticky-error check · epoch fence ·<br/>idempotence classify"]
        classify --> roll{"segment full?"}
        roll -- yes --> rollfast["roll_fast: fsync log ·<br/>create new active · pointer swap<br/>(old FDs move into deferred closure)"]
        roll -- no --> write
        rollfast --> write["append_batch → pwrite on active log"]
        write --> snap["snapshot.store (ArcSwap)"]
        snap --> trigger["pending ≥ flush_interval_messages?<br/>requested_flush_seq += 1"]
        trigger --> unlock["drop lock"]
        unlock --> wait["acks = -1 and triggered:<br/>park on Notify until<br/>completed_flush_seq ≥ mine"]
    end

    subgraph committer["committer task — one per partition"]
        direction TB
        recv["wait for a flush request"] --> target["lock: target = requested_flush_seq<br/>(skip if already completed)"]
        target --> fsync["spawn_blocking { lock(); log.sync_all() }<br/>fsync runs holding the partition mutex"]
        fsync --> watchdog{"done within 30 s?"}
        watchdog -- yes --> ok["completed_flush_seq = target<br/>notify_waiters()"]
        watchdog -- "timeout / io error" --> dead["sticky flush_err = Stalled —<br/>partition fails fast on next append;<br/>orphaned fsync drains in background"]
    end

    subgraph deferred["deferred finalize — spawn_blocking after roll"]
        fin["fsync index · close old log + index FDs"]
    end

    trigger -- "capacity-1 channel — coalesces" --> recv
    ok -. "notify" .-> wait
    rollfast -.-> fin
    readers["lock-free readers: high_water() / log_start()<br/>load the ArcSwap snapshot — never touch the mutex"] -.-> snap

Three properties that make this work:

1. Group commit: N concurrent appenders share one fsync cycle — the capacity-1 flush channel coalesces requests, and every waiter with flush_seq ≤ completed wakes on the same notify_waiters().
2. Lock-free reads: high-watermark / log-start observation goes through the ArcSwap snapshot, so a stuck NFS fsync can't stall metrics callbacks.
3. Fsync watchdog: a hung NFS server trips the 30 s timeout, sets a sticky Stalled error, and the partition fails fast instead of hanging appenders forever.

Note the committer holds the partition mutex for the fsync window (readers are unaffected via the ArcSwap; concurrent appenders queue on the lock). Fsyncing a cloned FD outside the mutex was built, measured, and deliberately reverted: it moved no throughput (the produce ceiling is set by write round-trips, not the lock — see Performance) and it made a subtle invariant durability-critical — the flush sequence published to waiting producers had to be the one sampled before the sync. Don't rebuild it without new evidence.

On-disk layout

/data/__cluster/                  ── cluster-wide files
    assignment.json
    credentials.json
    acls.json
    txn_state/
        slot-0.json               ── 50 slots, hash(transactional_id) % 50
        ...
    producer_fences/
        from-kaas-0.json          ── one per broker; cross-broker producer
        ...                          epoch fence broadcast
    producer_ids/
        kaas-0.json               ── per-broker producer-ID block high-water
    marker_queue/
        to-kaas-1/                ── per-broker txn marker inbox
            <pid>-<epoch>.json
    __consumer_offsets/
        <group_id>.json           ── per-group offset file

/data/<topic>/
    .config.json                  ── operator-written; retention / segment
                                     bytes / compaction knobs
    .topic-id.json                ── topic-incarnation stamp (see the
                                     substrate contract chapter)

/data/<topic>/<partition>/
    manifest.json                 ── { epoch, highWatermark, logStartOffset }
    producer-state.snapshot       ── idempotent-producer dedupe window
    recovery-checkpoint.json      ── fsynced-prefix hint (see below)
    00000005-00000000000000000000.log   ── epoch=5, base_offset=0
    00000005-00000000000000000000.index ──   8-byte (rel_offset, file_pos)
    00000005-00000000000000001000.log   ── epoch=5, base_offset=1000
    ...

(With a volume pool, a topic bound to a pool volume keeps this exact layout under /vols/<name>/ instead of /data/; the topic-level .config.json / .topic-id.json are written per involved root.)

__cluster/ is the cluster-state directory — kaas's file-shaped replacement for Kafka's internal topics. It can live on its own volume, separate from topic data; see the RWX substrate contract.

Each segment is a pair of files; the filename carries both the leader epoch and the base offset, so a stale ex-leader's writes can never physically collide with a new leader's segment:

{epoch:08x}-{base_offset:020d}.log     ── append-only log of v2 RecordBatches
{epoch:08x}-{base_offset:020d}.index   ── sparse offset index, 8 bytes/entry

The manifest is written on partition open (takeover routes through open) and on close/relinquish — not on segment roll and not per append — so its highWatermark can lag in-memory state; recovery treats the log as authoritative and reconciles on open. The recovery scan validates each batch against its own CRC32C and stops at the first batch that is short, structurally impossible, or fails its checksum; the log is then truncated to the last valid byte before any write handle is used, so a torn tail from a crash can never be appended past (which would strand acknowledged records behind unparseable garbage). Truncation only ever removes bytes the scan proved unparseable — a clean log is never touched. Who runs that recovery, and when, is the next chapter — file-handle ownership & takeover.

Recovery checkpoint — scanning only the tail

Re-scanning the whole active segment on every takeover is the dominant cost of broker startup on NFS (O(active segment) at the substrate's read bandwidth). A per-partition recovery checkpoint — Kafka's recovery-point idea — bounds it. recovery-checkpoint.json records {segment_base, byte_pos, high_watermark}: everything up to byte_pos of the named active segment is fsynced, with that log-end-offset. The committer refreshes it once the fsynced log grows a threshold (64 MiB) past the last one, and a clean close writes it at EOF. On open, recovery resumes the scan from byte_pos instead of byte 0 whenever the checkpoint still names the current active segment; if it doesn't (a roll happened since, or the file is missing/stale/truncated) it falls back to a full scan — always correct, and cheap with a bounded segment size.

The clean-shutdown fast path falls out for free and on one code path: a graceful close leaves the checkpoint at EOF, so "scan from checkpoint to EOF" reads zero bytes — the same path a crash takes, which scans only the gap. A missing checkpoint is harmless (full scan), but the checkpoint is not purely advisory: it seeds the recovered high watermark, and the scan trusts everything before its byte_pos without re-verifying it. That is why a fenced leader's close deliberately skips writing the checkpoint — a stale one from a superseded leader could seed the wrong watermark — and why corruption before the checkpointed position is trusted, not detected.

The index is sparse: one (rel_offset: i32, file_pos: i32) entry every index.interval.bytes of log data (4 KiB default). Lookup binary-searches to the closest entry ≤ the target offset, then scans the log forward. The index is not fsynced on the hot path — it's rebuildable from the log during takeover recovery, so only the log's durability is on the acks=all promise.

Byte opacity: the broker never parses records

Exactly three places on the Produce path touch the RecordBatch bytes, and none of them decodes a record:

1. The request decoder carries the records as an opaque, zero-copy slice into the frame buffer.
2. The idempotence check peeks the fixed-size batch header for producer ID / epoch / sequence — header only.
3. The segment append peeks the head for (base_offset, last_offset_delta, max_timestamp) — header only.

After that, the same opaque bytes the client sent land verbatim on disk (with the base offset rewritten in place): the log file IS the wire format, which is what makes Fetch a byte copy rather than a re-encode. Fetch is symmetric — batch bytes come back off disk undecoded. The invariant is enforced by tripwire counters that must stay at zero (see Observability) and by an integration test.

The durability dial

KAAS_FLUSH_INTERVAL_MESSAGES (default 1 = honest acks=all: every batch waits for a group-commit fsync cycle) mirrors Apache Kafka's log.flush.interval.messages. Raising it trades durability for throughput by letting the committer skip cycles until N messages are pending — same semantics, same trade, as Apache. On NFS substrates where the COMMIT round-trip dominates, this and the group-commit coalescing are the two levers that matter (see Performance).

Retention, DeleteRecords, and compaction — the honest state

Per-topic policy flows from KafkaTopic.spec.config through the operator-written .config.json into the broker, where a background cleaner enforces it:

• Retention runs on a timer (five minutes by default, the same cadence and meaning as Apache's log.retention.check.interval.ms). Each pass drops closed segments that have aged past retention.ms or pushed the partition over retention.bytes, whichever reaps more. A topic that sets neither gets the cluster default — seven days, as in Apache. -1 on either knob opts out and retains forever. The active segment is never reclaimed.
• Segment rolls happen at segment.bytes (1 GiB) or segment.ms (seven days), whichever comes first, per topic. The time-based roll is not a detail: retention only ever deletes closed segments, so a topic that never fills a segment would otherwise keep everything regardless of its retention setting.
• DeleteRecords (API key 21) reclaims on demand rather than on a timer: it advances logStartOffset to a caller-chosen point and unlinks the closed segments that point fully covers. Topic deletion is the third path.
• Compaction is still missing. Its knobs min.compaction.lag.ms (KIP-58) and delete.retention.ms (KIP-354) round-trip through CRs and DescribeConfigs but gate nothing yet. When the compactor lands, tombstone expiry will be per-batch (Apache is per-record) — a deliberate consequence of never opening batches. Status is tracked honestly on the KIP-58 and KIP-354 pages.

Why a quiet topic used to keep everything

A subtlety worth stating plainly, because it makes retention look broken when it isn't: retention can only delete closed segments. The segment currently being written is off limits — Kafka works the same way — so a topic only starts reclaiming space once it has rolled at least one segment behind it.

That is why segment.ms matters as much as retention.ms. A topic producing a few KB a minute will not reach a 1 GiB segment this decade. Without a time-based roll it has exactly one segment, that segment is the active one, and no retention setting can touch it — the topic grows forever while kafka-configs.sh cheerfully reports a 24-hour retention. With segment.ms at seven days, the same topic rolls weekly and its retention applies from then on.

Both knobs are per topic and both apply to a live log: changing them takes effect within one sweep, without restarting a broker.

How old is a segment?

Time-based retention has to answer that per segment, and the answer is less obvious than it looks. Apache keeps a time index alongside each segment and reads the largest record timestamp out of it. kaas has no time index, and its manifest — the small JSON file beside a partition's segments — records the epoch, high watermark and log start offset, but no per-segment list. So there is nowhere on disk for a segment's largest timestamp to live.

kaas therefore uses the timestamp it does know for free. While a segment is being written, the broker tracks the largest record timestamp it has seen, and stamps that onto the segment when it rolls closed. A segment closed by this broker, in this process, is dated by its records — exactly like Apache.

Everything else is dated by the log file's modification time. That covers every segment a broker inherits when it restarts or takes a partition over from a peer. It is the same fallback Apache uses when a segment has no usable timestamp, and it is well-behaved here: a closed segment is never written again, so its mtime stops moving the moment it rolls.

The practical difference is small but worth knowing: a producer that back-dates its records — setting timestamps far in the past — will see them aged out promptly by the broker that wrote them, and aged by wall-clock arrival time by a broker that inherited them.

Retention only ever runs on the broker that leads the partition. This is not an optimisation: on a shared volume, a second broker deleting the same segments is a data race, and deleting a file another broker still holds open doesn't free the space at all — the filesystem just renames it out of the way. See file-handle ownership.

Implementation notes (for contributors)

• Issue trail: the group-commit Produce path is gh #80/#81/#82; the lock-free ArcSwap read path is gh #134; the 30 s fsync watchdog is gh #95; stateless fetch sessions are gh #4; wiring the retention cleaner's interval loop was gh #250, and the compactor is still open on gh #158.
• Byte-opacity peeks: kaas-storage/src/idempotence.rs (PID / epoch / sequence) and kaas-storage/src/segment.rs (offsets / timestamp); the codec side is kaas-codec decoding records: Option<bytes::Bytes> as a zero-copy slice. Enforced by the tripwire counters and the bins/kaas/tests/byte_opacity.rs integration test.
• Topic-config plumbing: crates/kaas-storage/src/topicconfig.rs; RetentionCleaner and its policy sources: crates/kaas-storage/src/cleaner.rs; the interval loop and the leadership gate: bins/kaas/src/main.rs.
• The producer-fence files and marker queue in the layout above belong to the transaction machinery (gh #108 phase 2, gh #175) — see Transactions & idempotence.
• The index is read into memory on open; an mmap-backed index is future work noted in segment.rs (the workspace forbids unsafe code, so it would need a vetted dependency).

File-handle ownership & takeover

Only a partition's current leader holds open file descriptors — the rule that makes deletes actually free disk on NFS instead of silly-renaming.

In Apache Kafka, log dirs are broker-local, so file handles are nobody's problem: every replica holds its own segments open, and deleting a segment is a local unlink no other machine can observe. On kaas's shared volume, every broker mounts the same partition directories the storage hot path writes — and NFS adds a rule local filesystems don't have: removing a file that any client still holds open "silly-renames" it into a hidden .nfsXXXX entry that pins the parent directory (EBUSY on removal) until every descriptor closes. Left unmanaged, segment cleanup stops reclaiming disk and topic deletion loops forever.

kaas's answer extends its single-writer-per-partition model to file descriptors: only a partition's current leader holds open log and index handles. Every other broker knows the segments as metadata only — size, base offset, and leader epoch, all readable from the filename without opening the file. Handles are opened on takeover and dropped on relinquish or close; even opening a partition at broker startup only stats the segment files.

The payoff is day-to-day, not just in failure cases: segment retention, DeleteRecords, and segment-roll cleanup all unlink files on the leader — the only broker with the descriptors open — so removal genuinely frees space instead of leaving .nfsXXXX ghosts.

Takeover and relinquish

When assignment.json moves a partition (the controller chapter covers who writes it and why):

• New leader: takeover opens the log + index handles, restores the idempotent-producer snapshot, and runs segment recovery — scanning the active segment forward, verifying each batch's CRC, truncating any torn tail to the last valid byte, and reconciling the manifest's possibly-stale high watermark against what's actually on disk. Recovery runs at takeover time precisely because the manifest is allowed to lag (see Storage engine hot path). It then raises the partition's epoch to the one the assignment carries, rolls to a fresh segment file stamped with that epoch, and persists the manifest before accepting a single write.
• Old leader: relinquish persists the manifest and producer snapshot one last time, then closes the handles — unless the manifest on disk already carries a higher epoch, in which case it has been superseded and writes nothing but still releases its descriptors.

That roll is what makes the epoch-prefixed filenames load-bearing rather than decorative. A departing leader does not find out it lost the partition until its own copy of assignment.json catches up, so there is a window where it still believes it leads and keeps accepting produce requests. Because the new leader appends to a different file, the two can't interleave writes into one — the deposed leader's records land in a segment belonging to a dead epoch and are discarded, instead of being spliced byte-wise into the live log. Its in-flight produce requests start failing as soon as the epoch bump lands, since the append path rejects any write carrying an epoch below the partition's current one.

This is the shape Apache Kafka gets from replication and a leader epoch cache: a follower that becomes leader starts a new epoch, and records written by the old leader past the divergence point are truncated away. kaas has no followers to truncate against, so it separates the writers at the filesystem instead — same guarantee, different mechanism.

Manifest + producer snapshot

Two sibling files ride along with every partition's segments:

• manifest.json — (epoch, highWatermark, logStartOffset), written temp + fsync + rename. Persisted on partition open and on close/relinquish — not per append, and not on segment roll — so recovery treats the log itself as authoritative.
• producer-state.snapshot — the idempotent-producer dedupe window, written on segment roll and relinquish, restored on takeover. Without it a leadership move would drop the per-producer sequence history, and in-flight producer retries would be misclassified as OUT_OF_ORDER_SEQUENCE_NUMBER instead of duplicates.

Topic delete: the handle-close path

flowchart TD
    del["kubectl delete kafkatopic T"] --> watch["broker topic watch<br/>fires the delete event"]
    watch --> abandon["broker drops T from its topic registry ·<br/>abandons T's partitions:<br/>close handles, persist NOTHING"]
    abandon --> notify["assignment recompute triggered<br/>(reason: topic deleted)"]
    notify --> ctl["controller: balancer drops T's partitions,<br/>writes new assignment.json"]
    ctl --> apply["every broker applies it —<br/>relinquish is already a no-op<br/>for the abandoned partitions"]
    apply --> reclaim["operator reclaims /data/T:<br/>periodic sweep, or the identity check<br/>if T is recreated"]
    reclaim --> disk["directory unlink succeeds —<br/>no .nfsXXXX silly-rename, disk freed"]

Note the asymmetry between abandon and relinquish. A relinquish is a leadership handover: persist the manifest and producer snapshot, then release. A delete is not — the topic is gone, and the operator reclaims its directory by renaming it aside, so a recreated topic of the same name gets a fresh directory at the same path. A well-meaning close racing that sequence lands the dead incarnation's high watermark and dedupe window in the new incarnation's directory. Abandon therefore drops the handles and writes nothing, and dropping the in-memory partition forces the next takeover to re-open from disk rather than keep serving the deleted topic's state.

Both the delete and the recreate arrive on the same watch stream, so the abandon is ordered before the recreate's apply — no coordination needed. (What protects a recreated topic even without that ordering is the topic-identity stamp — see the RWX substrate contract.)

Graceful SIGTERM drain

The broker's shutdown path relinquishes every open partition — persisting each manifest and producer snapshot one final time and closing the active segment's handles, so the next leader doesn't inherit a silly-rename fight on takeover.

There is no controlled-shutdown RPC, but the drain is announced: before teardown the broker sets its drain flag, which rides the next heartbeat, and the controller moves its partitions off while the broker is still healthy enough to hand them over — the proactive "I'm draining" hint, rather than a heartbeat-timeout discovery. See Broker fencing for the draining state's other half.

This whole chapter is one discipline in service of a larger contract — the rules any code touching the shared volume must obey. That contract is the next chapter: The RWX substrate contract.

Implementation notes (for contributors)

• The leader-only FD rule is gh #76: TakeoverDriver calls the engine's take-over, which opens handles before recovery; relinquish closes them. Partition open at startup stats without opening.
• abandon_topic ≠ relinquish (gh #219) — don't "unify" them: abandon must close handles and persist nothing, or a late close writes the dead incarnation's HWM and dedupe window into a recreated topic's directory.
• The producer snapshot lives in crates/kaas-storage/src/producer_snapshot.rs.
• The SIGTERM drain is in bins/kaas/src/main.rs (gh #61, gh #139); partition keys are parsed from the right so slash-bearing topic names split correctly.

The RWX substrate contract

Apache Kafka gets durability and failover from replication. Every partition has a leader and a set of in-sync followers, and a record isn't acknowledged until it is on enough of them. Lose a broker and a follower is promoted — the data was already there.

kaas makes a different bet. It runs on Kubernetes and stores every partition on a shared ReadWriteMany volume with one writer per partition and no followers at all. Durability comes from the shared filesystem (plus whatever redundancy the storage itself provides), and failover means a surviving broker opens the same files the dead one was writing. There is no second copy to fall back on, because the shared volume is the copy. (Why give up replication? See the non-goals — in short, it trades a replication protocol and a consensus log for a much smaller system that leans on Kubernetes and the filesystem instead.)

That bet buys a lot, but it moves the hard problems onto the filesystem. kaas never speaks NFS — it does POSIX file operations on whatever the CSI driver mounted — so the real requirement is not a protocol but a semantic contract: three guarantees the substrate must honor, spelled out below. NFS is the reference substrate the contract was debugged against, and the floor it describes; correctness depends on respecting what that floor actually promises, which is less than most code assumes. The two previous chapters — the storage hot path and file-handle ownership — showed the machinery; this page is the contract that machinery obeys. Most of the subtle bugs in kaas's history are what happens when a piece of code forgets it.

Which filesystems qualify? Any RWX filesystem that delivers the three guarantees: the NFSv4.1 family (Linux kernel NFS, Azure NetApp Files, FSx for NetApp ONTAP, EFS, Azure Files NFS), and coherent POSIX filesystems like CephFS, which exceed the contract (stronger than close-to-open, and no silly-rename behavior — though the coding rules below still apply in full: read-modify-write and check-then-act are non-atomic on every shared filesystem). Metadata-engine filesystems over object storage (JuiceFS-class) satisfy the contract semantically but bring their own operational tax and object-PUT fsync latency. Disqualified regardless of protocol: SMB/CIFS (divergent open/rename/locking semantics), FUSE object-store mounts whose rename is copy+delete (s3fs, gcsfuse, Mountpoint for S3), and any substrate that lies about fsync — including an async NFS export, which is "still NFS" and still breaks everything. The risk direction is always a substrate weaker than the contract, never stronger.

What lives on the shared volume

Everything a broker needs to serve, and everything the cluster needs to coordinate, is a file other brokers can read:

• Partition logs — …/<topic>/<partition>/ with the usual segment and index files, the same idea as Kafka's on-disk log. Only a partition's current leader has them open.
• The assignment file — who leads what. This is kaas's equivalent of the partition-to-leader map Kafka keeps in its metadata log (or, pre-KRaft, in ZooKeeper). One broker — the elected controller — writes it; every broker reads it.
• The state Kafka keeps in internal topics — consumer offsets, transaction state, producer fences — are plain files here rather than __consumer_offsets and the transaction log. A broker that becomes a coordinator reads the same file the previous one wrote. This cluster-wide state lives in its own directory (__cluster/ by default), and can live on its own volume: the broker and operator honor KAAS_CLUSTER_DIR, and the chart's storage.controlPlane mounts a dedicated control-plane volume so a runaway topic filling the data volume degrades into a per-topic produce error instead of taking cluster coordination down with it.

Because these files are read and written across brokers, every one of them is exposed to the guarantees — and the non-guarantees — below.

What the substrate actually guarantees

Three things, and only three — this is the contract's floor, which NFS defines:

1. A rename within one directory is atomic. A reader sees the old target or the new one, never a half-written mix. This is the load-bearing primitive.
2. An exclusive create (open with O_CREAT|O_EXCL) is atomic. Exactly one racer creates the file; the rest are told it already exists.
3. Close-to-open consistency. Once one host closes a file, the next host to open it sees the complete contents. This is how one broker reads what another wrote.

That is the whole toolbox. Everything you might wish were atomic is not:

• Recursive delete is not atomic. It is a sequence of unlinks that can be observed, and interrupted, half-done.
• Read-modify-write is not atomic. Two writers interleave.
• Check-then-act is not atomic. "If it doesn't exist, create it" is a race — and "open a partition: make the directory, open the files, recover the tail" is exactly that shape.
• Deleting a file another host has open is not clean. NFS renames it to a hidden .nfsXXXX file and keeps the parent directory busy until every open handle closes. (This one is NFS-specific — CephFS behaves like a local filesystem here — but the file-handle discipline it forced is kept on every substrate, because it is what makes leader-side deletes actually free disk.)

You cannot make a recursive delete atomic on a shared filesystem. So the goal is not "make everything atomic" — it is the following.

The contract

1. Build durable state changes out of the atomic primitives. Write a temp file, flush it, then rename it over the target. Never mutate a file in place where another host can catch it half-written.

2. If an operation can't be a single atomic step, make it idempotent and drive it to completion by retry. On a shared volume it will race another actor or get interrupted, so "try once, log on failure" is a latent stuck state. Name the desired end-state, then re-drive toward it until it is reached.

3. Give every piece of state a single writer, fenced by epoch. If only one broker ever writes a partition, there is no concurrent writer to race — and an epoch stamp lets a new leader reject a zombie's late writes. This is Kafka's leader-epoch idea, applied to files.

Every rule here was paid for by a production bug; the contributor notes at the end of this chapter catalogue them and give the review checklist that keeps new code honest.

Rule 1 in practice

Every metadata file kaas persists — the per-partition manifest, the producer-state snapshot, the assignment file, the operator-written topic config — is written to a temp name, flushed, and renamed into place, so a reader sees either the previous version or the next one and never a torn write. Segment logs are never edited in place either: they are append-only, and a segment roll creates a new (epoch-stamped) file and swaps a pointer. A new persisted file goes through the same temp-then-rename path, no exceptions.

Rule 2 is the one that gets forgotten

A multi-step operation that isn't retried turns a momentary hiccup into a permanent fault. The mental model is name the end-state, then converge to it — never assume one attempt either fully succeeds or is someone else's problem. "Log a warning and move on" is the anti-pattern this rule exists to kill.

Rule 3 is kaas's core model

Only a partition's leader writes its log, and segment filenames carry the leadership epoch, so a stale leader's late write lands in a file the new leader ignores — Kafka's zombie fencing, done through the filesystem (see file-handle ownership & takeover). Where this breaks down is when a second actor touches state the single writer owns — for instance the operator deleting a topic's directory while a broker still has that partition open. That is outside rule 3, and it is exactly where races live.

Names are not identities

The three rules govern concurrent access. One class of bug slips past all of them, because no two writers ever run at once: state addressed by a name that gets reused.

A partition's directory is addressed by name — /data/<topic>/<partition>/. Delete a topic and recreate it under the same name (Kafka Streams' application-reset does this on every run) and the new topic silently inherits the old one's segments, high watermark, and idempotence dedupe window. No race, no second writer — the second incarnation simply moved into the first one's house. The visible symptom was a producer whose very first batch came back OUT_OF_ORDER_SEQUENCE_NUMBER — or worse, was accepted and silently discarded as a duplicate of a record written by a producer that no longer existed.

The same shape shows up wherever an identifier is recycled: a producer ID reissued after a broker restart lands a fresh producer on a dead one's sequence history, which is why kaas allocates producer IDs from a persisted, per-broker block (see transactions & idempotence).

Apache Kafka answers this with topic IDs (KIP-516) and producer-ID blocks. kaas answers it the same way: the operator stamps each topic directory with the CR's Status.TopicID (.topic-id.json) and reclaims the directory when the stamp belongs to a previous incarnation — a reconcile-time check, so it needs no delete event and no ordering between watchers. An unstamped directory is always adopted, never reclaimed: "unknown identity" must not be destructive.

Why this matters more for kaas than for Kafka

In Apache Kafka a broker's local disk is one replica among several: a corner case on one node is masked by the others, and the filesystem is rarely the point where the cluster coordinates. In kaas the shared volume is the only copy and the coordination point for the whole cluster, so a filesystem race isn't masked — it is the failure. This contract is what keeps "single writer on shared storage" as safe in practice as "replicated across brokers" is in Kafka.

Implementation notes (for contributors)

Any change that touches the shared volume gets reviewed against four questions before it ships:

1. Is this a single atomic primitive — a same-directory rename, or an exclusive create?
2. If not, is it idempotent and safe to re-drive until it completes?
3. Is there exactly one writer, fenced by epoch, against the rest?
4. Does this path assume a name identifies its state uniquely — over time, not just at this instant?

If all answers are "no," the code has a latent race — no matter how cleanly it passes on a single broker backed by a local disk.

The checklist is not theoretical. Each entry below shipped as a real bug, and each is one rule ignored:

Note the fifth entry was found by this document's own checklist, applied to a fix for the first: rename-aside solved a rule-3 race and quietly introduced a rule-2 one, because "rename, then delete, then re-create" is three steps and only the rename is atomic. That is the argument for writing the contract down — the next reviewer, holding it, can catch the next one before it ships.

The volume pool: log dirs & placement

If you know Apache Kafka, you know its storage model: every broker owns local disks, listed in log.dirs, and a partition's replicas live on the specific brokers that host them. kaas keeps the Kafka protocol but inverts that model. kaas brokers are (nearly) stateless processes on Kubernetes: partition data lives on shared ReadWriteMany volumes that every broker mounts, any broker can serve any partition, and durability comes from the storage layer instead of replication (there are no followers — see the RWX substrate contract for the ground rules that makes this safe).

Out of the box, all of that shared storage is one volume. The volume pool lets you mount several named volumes and control, per topic, which one holds its data. kaas deliberately describes this in Kafka's own vocabulary: one pool volume = one log dir (the KIP-113 concept). kafka-log-dirs.sh --describe against a kaas cluster lists every pool member, exactly as it would list a JBOD broker's disks.

Why you would want more than one volume

• Throughput, on cloud filers. On the storage kaas targets in the cloud (FSx for NetApp ONTAP, Azure NetApp Files, Azure Files provisioned tiers), I/O budget is provisioned per volume. Brokers add CPU and network, never disk bandwidth — the pool is how the substrate itself scales.
• Tiering. Put source-of-truth topics on a premium volume and recreatable ones (Kafka Streams changelog/repartition topics) on a cheap one — the same reason Kafka users mix disk classes, expressed per topic instead of per broker.
• Blast radius. A volume that fills up or fails takes down the topics placed on it, not the cluster.

One property has no Apache Kafka equivalent: because every volume is mounted by every broker, placement and leadership are independent. Pinning a topic to one volume constrains where its bytes live, never which broker leads it. JBOD on local disks can't do that.

Declaring a pool

The pool is Helm chart configuration (storage.pool[]). Each member becomes its own PVC, mounted on every broker and on the operator at /vols/<name>:

storage:
  className: nfs          # the default data volume (log dir "default")
  size: 100Gi
  pool:
    - name: bulk
      size: 500Gi
      className: standard-files   # per-member class = per-member substrate/QoS
      defaultEligible: true       # may receive topics with no explicit binding
      labels:
        class: standard           # matched by volumeSelector (below)
    - name: premium
      size: 100Gi
      className: premium-files
      defaultEligible: false      # reserved: only topics that ask for it
      labels:
        class: premium
      # cordoned: true            # drain mode: accepts no NEW placements

Things to know:

• The data volume is always a member, under the reserved name default. An empty pool: [] is exactly the classic single-volume layout.
• Members are addressed by name, never by position — removing one never renumbers the rest.
• defaultEligible: false makes a member reserved: topics land on it only by naming or selecting it. This is what keeps auto-created topics (Streams creates repartition topics without asking you) off premium storage.
• Adding or changing members is a chart upgrade and therefore a rolling restart — a capacity operation, the same cadence as adding brokers. Creating a topic never waits on volume provisioning.
• Every member must satisfy the same substrate contract (NFSv4-class semantics). Mixing, say, NFS and CephFS members is fine; a member that lies about fsync is not a slower tier, it is a correctness bug.

Binding topics to volumes

kaas manages topics as Kubernetes custom resources (KafkaTopic) — the Strimzi-style pattern; topics created over the Kafka Admin API get a CR created for them. The binding is one optional field on the topic:

apiVersion: kaas.rs/v1alpha1
kind: KafkaTopic
metadata:
  name: orders
spec:
  partitions: 12
  storage:
    volumes: [premium]        # pin: every partition on `premium`

Three shapes, one field:

• Pin — volumes: [premium]: hard isolation, one volume's budget.
• Stripe — volumes: [bulk-1, bulk-2]: partitions are spread round-robin across the set.
• Unset — the topic uses the default set: default plus every member with defaultEligible: true.

If you'd rather not hard-code infrastructure names into topic definitions, select members by label instead (the nodeSelector idea, applied to storage):

spec:
  storage:
    volumeSelector:
      class: premium          # every key/value must match the member's labels

volumes and volumeSelector are mutually exclusive. A binding that names an unknown member, or a selector that matches nothing, fails the topic's reconcile loudly — Ready=False with reason InvalidVolumeBinding — rather than silently placing data somewhere you didn't intend.

How placement behaves

Placement is decided once, when a partition is created, and recorded in the topic's status:

status:
  volumeAssignments:
    "0": premium
    "1": premium
    "2": premium
  partitionsOutsideSpec: 0

Editing the binding later never moves data. New partitions (from expansion) follow the new set; existing partitions keep serving where they are and are counted in status.partitionsOutsideSpec — visible drift instead of surprise I/O. This is deliberate: kaas has no replication layer to move data behind your back, so an inter-volume move is a raw copy, and raw copies only happen when you ask for one (next section).

If a placement record ever points at a member that no longer exists, brokers fall back to the default volume rather than failing the partition — resolution can never make a partition unopenable.

Moving data: cordon & drain

Draining a member (to decommission it, or to move a hot topic) is a three-step, explicitly operator-driven flow:

1. Cordon it (cordoned: true on the member, chart upgrade). From Kafka 4.3's vocabulary (KIP-1066): a cordoned log dir accepts no new partition placements — even from topics that name it — while existing partitions keep serving in place.

2. Move the partitions off. The convenient path is an annotation on each affected topic:

   kubectl annotate kafkatopic orders kaas.rs/migrate-to-volume=bulk
   

   Each broker then walks the partitions it leads through the move — close, copy to the target volume, flip the placement record, reclaim the source — one partition every few seconds. The annotation is level-triggered and idempotent: partitions already on the target are skipped, and you remove the annotation once status.volumeAssignments shows the move complete. (The underlying Kafka API is AlterReplicaLogDirs, which you can also drive directly; the destination is the log-dir path as shown by kafka-log-dirs.sh --describe.)

   What clients see: producers and consumers of a partition get a brief retriable LEADER_NOT_AVAILABLE window while its files are copied — standard client retries absorb it. If the record flip fails, the copy is rolled back; data location and placement record never disagree.

3. Remove the member from storage.pool[] once it hosts nothing, and delete its PVC.

Observing the pool

• kafka-log-dirs.sh --describe — every member with its partitions, and (v4 of the API, KIP-827) per-dir TotalBytes / UsableBytes.
• Metrics: kaas.log.dir.total.bytes and kaas.log.dir.usable.bytes, labelled per log dir — the pool's headroom on a dashboard.
• kubectl get kafkatopics.kaas.rs -o wide plus the status fields above for placement and drift.

Choosing backends

• FSx ONTAP / ANF (incl. manual-QoS capacity pools): each member is an independently provisioned throughput budget — the pool multiplies substrate bandwidth. The target case.
• EFS Elastic: the filesystem scales its own throughput and bills per byte; a pool adds mounts and cost with no spreading gain — use the pool for tiering only.
• CephFS (Rook): qualifies with no filer required; measure MDS and OSD-journal latency before promoting a member to a hot tier.
• Single-filer homelab NFS: members share one filer's budget; the pool is for layout/tier testing only.

Implementation notes (for contributors)

The flow of placement truth, and where it lives in the source:

operator reconciler: round-robin over the eligible set
  (crates/kaas-operator-controllers/src/kafkatopic_controller.rs)
  → KafkaTopic.status.volumeAssignments
    → broker topic watch (crates/kaas-k8s/src/kube_watchers.rs)
      → TopicRegistry, the engine's PlacementResolver
        (crates/kaas-broker/src/topic_registry.rs)
        → partition path resolution in the storage engine
          (crates/kaas-storage/src/disk.rs)

Segment files, the manifest, the producer-state snapshot, and the recovery checkpoint live inside the partition directory and follow it between volumes. Topic-level files (.config.json, the topic-identity stamp) are written to every root hosting the topic's partitions, and the topic-incarnation check and orphan sweep run per root — a deleted topic is reclaimed wherever its partitions were placed.

Consumer-group coordination

Deterministic hash routing of group coordination, two-tier ownership via assignment.json, and group takeover on assignment change.

In Apache Kafka, "which broker coordinates group G?" is a storage question: partitionFor(groupId) hashes the group ID into the internal __consumer_offsets topic, and whoever leads the resulting partition is the group coordinator. Group metadata and committed offsets are records in that topic; coordinator failover is partition-leadership failover.

kaas has no __consumer_offsets topic. It is one of the two internal topics kaas replaces with plain JSON files on the shared volume — the third substitution from the introduction; the other is __transaction_state, covered in Transactions & idempotence. Committed offsets live in one file per group, and coordinator routing hashes the group ID directly into the broker set instead of into a topic's partitions.

The coordinator is a pure function

Routing is a stateless computation over the group ID, the broker set, and the alive set — no lookup table, no election, no I/O:

• Hash: FNV-1a (32-bit) over the group ID, modulo the number of brokers. Clients never compute the coordinator themselves — they ask via FindCoordinator, exactly as against Apache Kafka — so any deterministic hash works as long as every broker computes the same one.
• Stable divisor: the modulus is pinned to the full broker set the controller knows about — including draining and dead brokers — not the alive count. Holding the divisor constant keeps existing assignments stable across rolling restarts; modding by the alive count would reshuffle roughly (N−1)/N of all groups on every pod up/down event.
• Preferred-slot-down fallback: when the hashed broker isn't alive, a deterministic alternate is picked from the alive subset, so coordination keeps working through a rolling restart.

The same machinery routes transaction coordination: hash(transactional.id) picks the transaction coordinator, which is what gates cross-broker transaction handling (see Transactions & idempotence).

Two-tier ownership

A broker answers "do I coordinate group G?" in two tiers:

1. Explicit entries win: if the controller's assignment.json carries a consumerGroups[] entry for G, that broker is the coordinator. This is the controller's group-balancing output — and the forward-compat lever for sticky rebalancing.
2. Hash fallback otherwise: the pure function above, over the full broker set.

For stable broker sets the two tiers converge, so the hash is the load-bearing path in steady state.

They can still disagree, though, and the disagreement is the reason both tiers need care. An explicit entry sticks to its broker across alive-set changes, so over time it can name a broker the hash would not. That makes losing an entry a real event: the group silently falls through to the hash, and if the hash says someone else, the group's clients get NOT_COORDINATOR and rebuild from scratch — on a perfectly healthy group, with no broker having restarted.

So an entry is retired only when its broker leaves the cluster, never because the group went quiet. The controller learns which groups are active from what brokers report, and a group drops out of that report for reasons that have nothing to do with coordination: its members all left for a moment, a heartbeat window went stale, an idle group was swept from memory. None of those mean the coordinator should move.

The same rule binds every component that asks the question. Anything deciding "do I still coordinate this group?" — including the takeover pass below — has to consult both tiers, because a group living on the hash tier alone (any brand-new group, for one) otherwise looks unowned to whoever reads only the explicit list.

And it binds the controller hardest of all, in a way that is easy to miss: the entry it writes has to agree with the hash it is replacing. A brand-new group has no entry, so brokers serve it from the hash. The moment the controller writes the group's first entry, that entry becomes the answer. If the controller picked the coordinator with a different function than the hash tier uses — even a perfectly good one — then writing it moves the group, and the group's clients see NOT_COORDINATOR and rebuild, once, shortly after they start, for no reason a user could ever diagnose. Both sides resolve through the same function over the same broker list, so the first entry confirms rather than relocates.

Group takeover and the orphan sweep

When the assignment changes — a broker joins, leaves, or dies — every broker sweeps the groups it holds in memory and drops the ones the current assignment no longer routes to it. Groups it gains are not eagerly migrated: the new coordinator's first JoinGroup creates the group lazily and loads its persisted offsets from the group's file on the shared volume.

Note which set the sweep walks: every group resident in memory, not the groups this broker believes it coordinates. Those are different sets, and the second one is precisely the wrong one — filtering by ownership before sweeping hides the disowned groups the sweep exists to evict.

The sweep is what keeps memory bounded across alive-set churn, and it is what keeps kafka-consumer-groups.sh --list honest: the AdminClient unions ListGroups across all brokers, so a single broker holding one forgotten in-memory group would make a deleted group reappear cluster-wide.

Coordinator changes are logged on both brokers involved — the one losing the group and the one gaining it — with the assignment version that caused the move, so a client-visible rebalance can be traced back to the recompute that triggered it.

Ownership also filters the read side: ListGroups on a broker only returns groups it currently coordinates, and DescribeGroups for a group owned elsewhere answers NOT_COORDINATOR — a stale entry on a non-coordinator broker is never visible to clients.

Where group state lives

• Membership, generation, protocol state: in-memory on the coordinator broker, exactly as in Apache Kafka. It is lost on coordinator failover — consumers rejoin and rebalance, which is what a coordinator change looks like to clients of Apache Kafka too.
• Committed offsets: one JSON file per group, /data/__cluster/__consumer_offsets/<groupID>.json, on the shared volume. Durable across failover: where an Apache Kafka coordinator replays its __consumer_offsets partition to materialize offsets, the new kaas coordinator simply reads the same file — the file is the materialized state.
• Transactional offsets are staged in a pending layer keyed by (group ID, producer ID) and only become visible to OffsetFetch when the transaction commits — see Transactions & idempotence.

Implementation notes (for contributors)

• Routing: crates/kaas-broker/src/group_hash.rs — pure functions over (key, brokers, alive), no state, no I/O (gh #92). The same hash gates txn-slot ownership (gh #91).
• The coordinator manager's group-assignment source is hot-swapped from bins/kaas/src/cluster.rs after the broker Coordinator boots; the bootstrap source is an always-true local stub for the brief window before the cluster runtime is up (tests substitute their own). Don't unwire the swap: an earlier attempt (v0.1.52) hit the chicken-and-egg where strict coordinator checks blocked fresh-group bootstrap, and was reverted in v0.1.53 (gh #92).
• Takeover: GroupTakeoverDriver (crates/kaas-broker/src/group_takeover.rs) runs a single sweep over Manager::resident_groups() on every assignment change — the old prev→next diff is subsumed (only gain-side logging still compares prev); the sweep fixed the gh #89 stale---list symptom. Lazy group creation is Manager::get_or_create.
• Read-side ownership filtering: Manager::list_groups() / describe_groups() in crates/kaas-coordinator/src/manager.rs.
• State: membership/generation in crates/kaas-coordinator/src/group.rs; offset persistence in crates/kaas-coordinator/src/offset_store.rs.

Transactions & idempotence

Idempotent-producer dedupe, the transaction coordinator state machine on slot-sharded JSON files, and EOS v2 end to end.

In Apache Kafka, exactly-once rests on two pieces of broker machinery: per-partition producer state (the idempotence dedupe window) and a transaction coordinator whose state lives in the internal __transaction_state topic. kaas keeps the first nearly verbatim and replaces the second: there is no __transaction_state topic. Like __consumer_offsets (Consumer-group coordination), it is an internal topic replaced by plain JSON files on the shared volume — the third substitution from the introduction.

None of this is exotic. The Java producer has enabled idempotence by default since Kafka 3.0, so every kafka-console-producer invocation exercises this machinery — it's hot-path, not an opt-in feature. Four layers of state, all on the shared volume:

Idempotent producer

InitProducerId (key 22) hands a non-transactional producer a fresh PID at epoch 0. On every Produce, classification runs under the partition mutex, before append against a per-PID ring of the last 5 batches — mirroring the Java client's max.in.flight.requests.per.connection=5:

• duplicate → echo the cached baseOffset, no log write;
• out-of-order sequence → error 45 (OUT_OF_ORDER_SEQUENCE_NUMBER);
• stale epoch → error 47 (PRODUCER_FENCED);
• otherwise accept and advance the ring.

The ring survives leadership moves via producer-state.snapshot (written on segment roll + relinquish, restored on take-over — see File-handle ownership).

PIDs are never reused

The dedupe ring is keyed by (PID, epoch), so handing the same PID to two different producers is not a cosmetic collision — the second one inherits the first one's sequence history. Its batches are then either silently dropped (sequence range matches a cached batch → classified duplicate, stale base offset echoed, produce "succeeds", consumers read nothing) or rejected with OUT_OF_ORDER_SEQUENCE_NUMBER. Both failure modes have been observed in practice.

Apache Kafka draws PIDs from a global counter whose next block is persisted (ZooKeeper's /latest_producer_id_block, KRaft's ProducerIdsRecord). kaas has no metadata quorum (a stated non-goal), so it partitions the PID space by broker ordinal instead:

pid = (broker_id + 1) * 2^40 + local

Each broker is the single writer of its own slice and of its own block file /data/__cluster/producer_ids/kaas-<id>.json, so there is no cross-broker read-modify-write on the shared volume. local advances in blocks of 1000, and the block end is persisted (tmp + fsync + rename) before any PID in it is handed out — a crash can only skip PIDs forward, never rewind. The + 1 keeps broker 0 clear of the low PIDs an earlier in-memory allocator handed out, so an upgrade can't collide with producer state already on the volume.

Fencing across partitions and brokers

A transactional producer that reconnects gets the same PID with epoch+1 — fencing is the monotonic epoch, exactly Apache's KIP-360 contract. Two mechanisms make the bump stick everywhere:

• Cross-partition fence: after every epoch > 0 rejoin, the InitProducerId handler walks every local partition, advances the PID's epoch and clears its dedupe window — so a zombie batch from the old session is fenced even on partitions the new session hasn't touched yet.
• Cross-broker fence broadcast: the bump is appended to a per-broker fence log under /data/__cluster/producer_fences/; every peer polls the logs and applies the bumps it hasn't seen. Same shared-volume pattern as the marker queue below — no new RPC surface.

Transaction state machine

Per-transactional.id state is slot-sharded across /data/__cluster/txn_state/slot-N.json (50 slots, fnv1a(transactional.id) % 50 — the same 50 Apache Kafka defaults to for transaction.state.log.num.partitions). The states a transaction actually visits:

stateDiagram-v2
    [*] --> Empty : InitProducerId first allocation<br/>PID assigned, epoch 0
    Empty --> Ongoing : AddPartitionsToTxn /<br/>AddOffsetsToTxn<br/>stamps ongoingSinceMs
    Ongoing --> PrepareCommit : EndTxn(commit)<br/>partitions + groups retained —<br/>the durable dispatch set
    Ongoing --> PrepareAbort : EndTxn(abort)
    Ongoing --> PrepareAbort : timeout reaper, 10 s sweep<br/>ongoingSinceMs + transactionTimeoutMs elapsed<br/>epoch bump, dispatch set retained
    PrepareCommit --> CompleteCommit : every marker durable<br/>clears partitions + groups,<br/>staged offsets committed
    PrepareAbort --> CompleteAbort : every marker durable<br/>staged offsets discarded
    CompleteCommit --> Ongoing : AddPartitionsToTxn /<br/>AddOffsetsToTxn<br/>next transaction begins
    CompleteAbort --> Ongoing : AddPartitionsToTxn /<br/>AddOffsetsToTxn

Facts the diagram compresses:

• The Prepare* states are the durability pivot, exactly as in Apache: a prepared entry keeps its partition and group lists, and that retained list is the durable record of "these markers still owe a write". A marker-dispatch failure, a coordinator crash, or a producer retry all re-derive the identical dispatch set from it; only the transition to Complete* — taken once every marker is durable — clears the lists and releases the staged offsets.
• InitProducerId on a rejoin does not reset the state: the entry keeps the same PID and bumps epoch += 1 — fencing is purely the monotonic epoch. Only epoch overflow (i16::MAX) allocates a fresh PID and resets to Empty.
• A retried EndTxn in the matching Complete* state is answered idempotently (no second transition); a direction mismatch returns INVALID_TXN_STATE, and EndTxn on Empty is INVALID_TXN_STATE too. Epoch mismatches return PRODUCER_FENCED everywhere.

EndTxn: commit flow

Cross-broker marker dispatch goes through a queue on the shared volume — there is no WriteTxnMarkers RPC between brokers. EndTxn is two-phase: prepare, dispatch every marker, then complete. For a peer-led partition, a durably written queue entry counts as a dispatched marker (the peer's watcher retries until it applies); a dispatch failure returns the retriable COORDINATOR_NOT_AVAILABLE and leaves the transaction prepared, so a producer retry — or the background reconcile below — re-derives the same dispatch set and finishes the job.

flowchart TD
    producer["Producer: EndTxn(commit)"] --> handler["EndTxn handler on the txn coordinator broker<br/>ownership gate — otherwise NOT_COORDINATOR"]
    handler --> prepare["state store: prepare_end_txn<br/>Ongoing → PrepareCommit<br/>partitions + groups retained — the dispatch set<br/>persist slot-N.json (tmp + fsync + rename)"]
    prepare --> split{"leader of each<br/>txn partition?"}
    split -- "self-led" --> local["write COMMIT control batch directly<br/>append to the log, acks=-1"]
    split -- "peer-led" --> enqueue["marker queue enqueue<br/>marker_queue/to-&lt;broker&gt;/&lt;pid&gt;-&lt;epoch&gt;.json"]
    local --> complete["state store: complete_end_txn<br/>PrepareCommit → CompleteCommit<br/>clears partitions + groups, ongoingSinceMs = 0"]
    enqueue --> complete
    complete --> hook["offset hook, per recorded group<br/>commit → commit pending offsets<br/>abort → discard pending offsets"]
    hook --> respond["EndTxn response error_code=0"]
    split -- "any dispatch fails" --> retriable["respond COORDINATOR_NOT_AVAILABLE (retriable)<br/>entry stays PrepareCommit — reconcile finishes it"]
    enqueue -.-> watcher["peer broker's marker watcher<br/>polls its own to-&lt;self&gt;/ every 2 s"]
    watcher -.-> apply["applies marker as control-batch append<br/>to partitions it leads, then deletes the file"]

The offset hook fires on the complete transition — not the prepare — so staged offsets only become visible to OffsetFetch once the markers backing them are durable.

Self-led markers are written before the queue entries, so a coordinator crash mid-dispatch never loses the local marker. A retried EndTxn overwrites the same {pid}-{epoch}.json file — the queue is idempotent by naming. Consumers in read_committed only see the transaction's records once these markers land (the fetch path clamps to the last stable offset).

Coordinator routing and staged offsets

Which broker coordinates a transaction is the same deterministic hash story as consumer groups: hash(transactional.id) picks the slot owner, and non-coordinators answer the txn APIs with NOT_COORDINATOR — see Consumer-group coordination. On coordinator failover the new owner simply reads the same slot file off the shared volume: close-to-open consistency means the file is the materialized state, with no log replay — this is the architectural replacement for Apache's __transaction_state topic.

TxnOffsetCommit (key 28) stages consumer offsets in a pending layer keyed by (group ID, PID) in the offset store — invisible to OffsetFetch until EndTxn commits. AddOffsetsToTxn (key 25) records which groups the transaction will touch, so the EndTxn offset hook knows exactly which pending sets to commit or discard. That hook firing atomically with the state transition is the KIP-447 (EOS v2) contract.

The timeout reaper and the marker reconcile

The transaction timeout reaper fires every 10 s — Apache's transaction.abort.timed.out.transaction.cleanup.interval.ms default. Any Ongoing entry past ongoingSinceMs + transactionTimeoutMs transitions to PrepareAbort with an epoch bump — and, crucially, keeps its partition and group lists: a timed-out transaction owes ABORT markers exactly like a client-driven abort does.

A marker reconcile pass shares the same 10 s tick: it walks every prepared transaction this broker coordinates, places the outstanding markers, and runs the complete transition — which is when the staged offsets are discarded (or committed) via the offset hook. The two halves are complementary, not redundant: the EndTxn handler's inline dispatch keeps commit latency off the sweep interval, but only the reconcile can finish a transaction whose producer crashed, was fenced, or (for a reaper abort) never existed to retry at all. A dispatch that still fails is left prepared and retried next pass, deliberately without bound.

Both sweeps are ownership-gated: a transaction slot file has exactly one legal writer — its coordinator — so each broker reaps and reconciles only the transactions it owns (an ungated sweep would have every broker read-modify-writing the same slot files on the shared volume, violating the substrate rules). The gate degrades safely at both edges: with no coordinator installed (dev/single-broker) everything is owned, and in cluster mode nothing is owned until the first assignment load — which delays a sweep by one poll rather than skipping it.

Implementation notes (for contributors)

• Dedupe ring: crates/kaas-storage/src/idempotence.rs (ProducerStates); snapshot persistence: crates/kaas-storage/src/producer_snapshot.rs.
• PID block allocator: crates/kaas-broker/src/producer_id.rs (gh #219 — both the silent-drop and the OUT_OF_ORDER symptoms of PID reuse were seen there; the pre-fix allocator was an in-memory AtomicI64).
• Fence-on-rejoin contract: gh #22. Cross-broker fence broadcast (gh #108 phase 2): fence log in crates/kaas-coordinator/src/fence_log.rs, applied by each peer's FenceWatcher (crates/kaas-broker/src/fence_watcher.rs).
• Txn state store + slot sharding: crates/kaas-coordinator/src/txn_state.rs — the architectural answer to gh #29 (no literal __transaction_state topic).
• Marker queue: gh #175. Txn-slot hash ownership: gh #91. Two-phase EndTxn + the marker reconcile: gh #225, shared dispatch in crates/kaas-broker/src/txn_markers.rs (reconcile_pending_markers over TxnStateStore::pending_marker_dispatches).
• The reaper and reconcile are spawned by the broker's cluster runtime (bins/kaas/src/cluster.rs), both gated on Broker::owns_txn (abort_overdue_owned on the store side; the ungated abort_overdue is tests/dev-mode only).
• The full KIP-447 consume-process-produce-commit round trip runs against an in-process broker in bins/kaas/tests/eos_v2.rs.

Listeners, authentication, authorization

Strimzi-shaped listeners, per-listener authentication engines, and cluster-wide ACL and quota enforcement.

If you have configured Apache Kafka, you know the listener trinity — listeners, advertised.listeners, listener.security.protocol.map — with a cluster-wide authorizer and KIP-13 quotas layered on top. If you have run Strimzi, you know its friendlier shape: an array of listeners, each declaring its own port, type, TLS, and authentication. kaas adopts the Strimzi shape 1:1 and keeps Apache Kafka's split intact: authentication is per-listener; authorization and quotas are cluster-wide.

Where the security metadata lives is the kaas difference. Apache Kafka stores SCRAM credentials and ACLs in the metadata quorum, managed with kafka-configs.sh / kafka-acls.sh; kaas manages users as Kubernetes custom resources (KafkaUser, mirroring Strimzi's), which the operator materializes into JSON files on the shared volume — part of the CRs-as-metadata substitution from the introduction. Brokers hot-reload those files: no broker restart on user or ACL changes, and no Kubernetes API call on the request path.

Three orthogonal listener axes

Listeners are declared in the Helm chart (.Values.listeners[]); each entry combines three independent axes:

• type: internal (in-cluster only) vs external (Gateway + cert-manager + per-broker hostnames).
• tls: false / true. mtls authentication implies tls: true; everything else is independent.
• authentication.type: none / scram-sha-512 / mtls / plain / oauth. Each listener gets its own auth engine, selected by listener name — a free-form string the chart picks.

Running one listener per combination is normal — e.g. keep plain anonymous for in-cluster bench/UI traffic and add an authed SCRAM listener side by side, both governed by the same cluster-wide ACLs:

listeners:
  - name: plain            # anonymous, in-cluster
    port: 9092
    type: internal
    tls: false
    authentication:
      type: none
  - name: authed           # SASL required, same ACL policy
    port: 9095
    type: internal
    tls: false
    authentication:
      type: scram-sha-512

Per-listener Metadata advertisement

Each broker endpoint carries a per-listener port map, and the Metadata handler answers with the port matching the listener the request arrived on: a client that bootstrapped on :9095 gets :9095 back, not :9092. Without this, an authed-listener client was handed the anonymous listener's port in the Metadata response and looped on SCRAM retry against a listener that never asks for SASL.

The pre-auth gate on an authed listener

Anonymous listeners use an allow-all engine (no SASL, no principal); on authenticated listeners the dispatcher blocks every API except the pre-auth allowlist — SaslHandshake (17), ApiVersions (18), SaslAuthenticate (36) — until the SASL exchange completes:

sequenceDiagram
    participant C as Client
    participant D as Dispatcher<br/>(per-listener gate)
    participant S as SCRAM-SHA-512 engine

    C->>D: ApiVersions (18)
    D-->>C: ok — pre-auth allowlist: 17 / 18 / 36
    C->>D: Metadata (3), before SASL
    D-->>C: CLUSTER_AUTHORIZATION_FAILED (31)<br/>in-band error, connection stays open
    C->>D: SaslHandshake (17), mechanism SCRAM-SHA-512
    D-->>C: supported: SCRAM-SHA-512, PLAIN
    C->>D: SaslAuthenticate (36)<br/>client-first: n,,n=user,r=client-nonce
    D->>S: step exchange (state kept per connection)
    S-->>C: server-first: r=combined-nonce,<br/>s=salt, i=iterations
    C->>D: SaslAuthenticate (36)<br/>client-final: c=biws, r, p=proof
    S->>S: recompute signature, constant-time<br/>compare against StoredKey
    S-->>C: server-final: v=server-signature, done
    Note over D: connection state: principal = User:name,<br/>sasl_done = true
    C->>D: Metadata (3)
    D-->>C: dispatched — authorization now via<br/>cluster-wide ACLs + quotas

An mTLS listener satisfies the same gate at the TLS handshake instead: the server extracts the principal from the client certificate (through the KIP-371 principal-mapping rules below) and marks the connection authenticated before any Kafka API arrives.

Anonymous listeners are not necessarily allow-all forever: setting the chart's auth.requireSasl (the KAAS_REQUIRE_SASL env) hands listeners declared authentication.type: none the real SASL engine too, arming the pre-auth gate cluster-wide — every listener then demands a completed SASL exchange before dispatching. Only KAAS_AUTH_DISABLED=true outranks it.

OAuth listeners (SASL/OAUTHBEARER)

If you have pointed a Strimzi listener at an OIDC provider — oauth.valid.issuer.uri, oauth.jwks.endpoint.uri — the kaas shape will look familiar. An oauth listener authenticates clients with the OAUTHBEARER mechanism (KIP-255): the client obtains an OAuth 2 access token (a JWT) from an external issuer — EntraID, Keycloak, Dex — and presents it during the SASL exchange. The broker validates the token locally: signature against the issuer's published JWKS, exp/nbf with 60 s clock-skew allowance, exact iss match, and optionally aud. No introspection round-trip per connection, no client secret on the broker.

listeners:
  - name: oauth
    port: 9096
    type: internal
    tls: true                # required in practice — see below
    authentication:
      type: oauth
      validIssuerUri: "https://login.microsoftonline.com/<tenant>/v2.0"
      jwksEndpointUri: "https://login.microsoftonline.com/<tenant>/discovery/v2.0/keys"
      userNameClaim: sub
      maxSecondsWithoutReauthentication: 3600

The Kafka principal is User:<claim> with the claim configurable (userNameClaim, default sub; a fallbackUserNameClaim is tried when the primary is absent) — for an EntraID service principal, sub is the SP object id, so ACLs written for that GUID match the same identity Strimzi sees. The signing keys are re-fetched every jwksRefreshSeconds (default 300), with an early re-fetch when a token names an unknown key id — an issuer rotating its keys costs one rejected connection attempt, not five minutes of failures. Until the first successful JWKS fetch every token is rejected: an unreachable issuer means clients cannot authenticate, never that validation is skipped.

Four deliberate hard edges:

• OAUTHBEARER requires TLS. A bearer token on the wire is a reusable credential — anyone who reads it can be you until it expires. kaas refuses the mechanism on plaintext connections, same as it refuses SASL PLAIN (SCRAM, which sends proofs rather than secrets, stays allowed on plaintext).
• The alg header is an allowlist (RS256/RS384/RS512/ ES256), never trusted from the token: none and the HMAC family are rejected outright, which closes the classic algorithm-confusion attack where an HS256 token is "signed" with the public JWKS bytes.
• Re-authentication is bounded (KIP-368). With maxSecondsWithoutReauthentication set, a successful authentication advertises session_lifetime_ms = min(configured, token remaining lifetime) and the broker refuses further requests past the deadline until the client re-authenticates on the same connection — a connection cannot outlive its token by more than the configured bound. Re-authentication may not change the principal. Unset, the session is unbounded — the same default as Apache Kafka's connections.max.reauth.ms=0.
• A rejected token fails in two steps, per RFC 7628: the broker answers with a JSON {"status":"invalid_token"} challenge, the client acknowledges, and only then does the exchange fail with SASL_AUTHENTICATION_FAILED (58). The JSON body is deliberately content-free; the reason (expired, wrong issuer, unknown key) goes to the broker log.

Once a principal is on the connection, Produce/Fetch and the admin surfaces consult the single cluster-wide authorizer and quota checker — which is what lets an anonymous plain listener and an authed SCRAM listener share one ACL/quota policy.

Authorization

The cluster-wide authorizer is wired by KAAS_AUTHORIZATION_TYPE: empty (default) means allow-all; simple enables ACL evaluation against /data/__cluster/acls.json. KAAS_SUPER_USERS (comma-separated User:foo,User:bar) wraps whichever authorizer was picked in a super-user early-allow layer.

ACLs and credentials are operator-materialized: KafkaUser CRs become entries in credentials.json + acls.json, which brokers hot-reload. SCRAM credentials can also be rotated over the wire with Kafka's SCRAM admin API (KIP-554, kafka-configs.sh --alter --add-config SCRAM-SHA-512=…): describe answers from the hot-reloaded credential store, and alter patches the KafkaUser CR — the CR stays the source of truth, and the operator materializes the new credential as usual (see the ACL & quota admin APIs). KAAS_AUTH_DISABLED=true switches the whole subsystem off for dev setups.

Authorization-only users (OAuth principals). A KafkaUser's spec.authentication is optional (gh #42). An OAUTHBEARER principal authenticates against the issuer, not against a stored credential, so there is nothing for the operator to materialize — its KafkaUser omits authentication entirely and carries only authorization (and optional quotas), naming the principal through metadata.name (the token's sub claim). The reconciler writes no credentials.json entry for it and only contributes its rules to acls.json. This mirrors Strimzi, whose OAuth users are authorization-only too, and is what lets you author ACLs and quotas for a token-authenticated identity.

mTLS principal mapping (KIP-371)

kaas parses Apache's ssl.principal.mapping.rules syntax — regex over the full subject DN with $1/$2 back-references and /L//U case postfixes; first matching rule wins, DEFAULT returns the CN. The server applies the mapper to the client certificate's subject DN during the TLS handshake. Parse errors fail at startup, so a chart-config typo is a crash-loop with a clear message, not every certificate silently mapping to its CN.

Quotas

The quota checker defaults to no-op and switches to real token buckets when auth is enabled. Two properties matter:

• Quotas are orthogonal to authorization — they fire even with authorization off, and per KIP-13 they are per-broker: with N brokers the effective cluster ceiling is N × the configured rate (the CRD field names say so explicitly — see Kubernetes integration).
• Debt-carry: the token bucket carries negative balances forward as debt rather than clamping at zero. With clamping, N concurrent clients each saw a "full" bucket and burst at N× the configured rate before throttling engaged — the observed 16-vs-10 MiB/s gap under bench load. Removing the clamp matches Apache's behaviour.

Throttle decisions surface as throttle_time_ms in responses. kaas computes and returns it but does not yet mute the connection channel afterwards (KIP-219's enforcement half) — cooperative clients throttle themselves; adversarial ones are a known gap tracked in the KIP index.

Implementation notes (for contributors)

• Listener array → KAAS_LISTENERS JSON env: gh #126. The connection's listener name is carried by crates/kaas-protocol/src/connstate.rs (free-form string, no predefined constants — the chart picks the names).
• Per-listener auth engines live in crates/kaas-auth, selected per listener name; the pre-auth gate is enforced in the protocol dispatcher (gh #124).
• Per-listener Metadata port advertisement and quota debt-carry: gh #125. Debt-carry is pinned by the multi_client_contention_carries_debt unit test next to the token bucket (crates/kaas-auth/src/quota.rs).
• ACL evaluation: crates/kaas-auth/src/acls.rs.
• Principal mapping: crates/kaas-auth/src/principal_mapping.rs (gh #43).
• OAUTHBEARER + JWT/JWKS validation: crates/kaas-auth/src/oauth.rs (gh #42). The validator is pure-sync (SASL hot path); the JWKS HTTP fetch loop lives in bins/kaas/src/main.rs (spawn_jwks_refreshers), mirroring the credentials/ACL hot-reload split. Config field names mirror Strimzi's KafkaListenerAuthenticationOAuth 1:1 and travel chart → KAAS_LISTENERS → crates/kaas-broker/src/cli.rs. End-to-end smoke: bins/kaas/tests/oauth_smoke.rs (TLS listener + wiremock JWKS + ES256 tokens).

Kubernetes integration

The three CRDs, their reconcilers, reconcile-time cleanup (no finalizers), and the broker's RBAC surface.

In Apache Kafka, cluster metadata — topics, users, ACLs, quotas — lives in the cluster's own replicated metadata log, and you manage it through the Admin API or the shell tools. kaas keeps that admin surface but moves the durable home of the metadata into Kubernetes custom resources — the CR half of the Lease-and-CRs substitution for the KRaft quorum (see the overview). If you have run Kafka under Strimzi, the shape is deliberately familiar: KafkaTopic and KafkaUser CRs reconciled by a single-replica operator. The difference is what reconciliation produces — not configuration pushed into running brokers, but files on the shared volume that brokers read directly.

The CRD surface

Three CRDs — KafkaCluster, KafkaTopic, and KafkaUser; the overview table shows what each materializes into, and the CRD reference documents every field. The CRD YAML ships bundled with the Helm chart. KafkaUser mirrors Strimzi 1:1 for spec.authentication / spec.authorization, with two deliberate divergences:

• Quota field naming: spec.quotas uses producerMaxByteRatePerBroker / consumerMaxByteRatePerBroker where Strimzi says producerByteRate / consumerByteRate. The semantics are identical to Strimzi/Apache (KIP-13: quotas are per-broker; N brokers → N× cluster ceiling) — the kaas names just say so honestly at the CR level.
• No group abstraction: there are no separate ACL or user-group CRs. ACLs are authored inline on each KafkaUser's spec.authorization.acls; granting the same rule to N principals means repeating it on N CRs — the standard Strimzi-pattern trade.

KafkaUser.spec.authentication is optional: an authorization-only user — an OAuth principal, named by metadata.name — carries only authorization and optional quotas, and materializes no credential entry (see Listeners, authentication, authorization).

TopicID (KIP-516) — where it stands

The KafkaTopic reconciler mints a v4 UUID into Status.TopicID on first reconcile and never rotates it, so a re-created topic gets a distinct ID — Apache's contract. Honesty note about the other half: the broker's topic watch does deliver that UUID into the topic registry, where it backs the stale-directory identity gate (a broker refuses to open a directory stamped with a previous incarnation's ID) — but it is deliberately kept out of the wire-facing metadata, so Metadata v10+ still serves nil topic IDs for all topics. Clients treat that as "broker doesn't expose topic IDs" and fall back to names. See the KIP index for the tracked gap.

Operator reconcile loops

One reconciler per CRD. None of them use cleanup finalizers — deleting a CR never blocks on the operator being alive; owned Kubernetes resources carry OwnerReferences so garbage collection is Kubernetes-native, and on-disk leftovers are reclaimed by a leader-elected sweep that runs at leader election and every five minutes thereafter.

flowchart LR
    api["Kubernetes API<br/>watch streams"]

    subgraph operator["kaas-operator — single replica, leader-elected"]
        rt["KafkaTopic reconciler<br/>requeue 300 s"]
        ru["KafkaUser reconciler<br/>await_change"]
        rc["KafkaCluster reconciler<br/>requeue 300 s"]
        sweep["orphan sweep — on leadership,<br/>then every 5 min: drop topic dirs +<br/>credential entries with no matching CR"]
    end

    api --> rt
    api --> ru
    api --> rc

    rt --> dirs["partition dirs<br/>/data/&lt;topic&gt;/&lt;0..N&gt;/ + .config.json"]
    rt --> tstat["Status.TopicID — v4 UUID minted on<br/>first reconcile, never rotated (KIP-516)"]
    ru --> creds["__cluster/credentials.json (upsert user)<br/>__cluster/acls.json (rebuilt from all users)"]
    ru --> secret["&lt;user&gt;-kafka-credentials Secret<br/>OwnerReference → K8s GC"]
    rc --> plumbing["cert-manager Certificates ·<br/>per-broker Services · TLSRoutes<br/>OwnerReferences → K8s GC"]
    sweep --> dirs
    sweep --> creds

Reconciler guard rails worth knowing:

• KafkaTopic refuses partition decrease (Ready=False, no filesystem mutation) — partitions only grow, matching Kafka semantics.
• KafkaUser with a missing referenced Secret parks on await_change instead of hot-looping.
• A CR with deletionTimestamp set is left untouched by the reconcilers; cleanup happens via K8s GC (owned resources) and the orphan sweep (on-disk state).

What brokers do with the CRDs

On the broker side, the CRD surface is read-mostly — but not read-only: the Kafka admin APIs CreatePartitions and IncrementalAlterConfigs are served by patching the KafkaTopic CR (spec.partitions / spec.config), CreateTopics mints a fresh one, DeleteTopics deletes it, and AlterReplicaLogDirs records partition moves in the CR's status. The user side works the same way against existing KafkaUser CRs: the ACL admin APIs edit spec.authorization.acls, the SCRAM admin API (KIP-554) patches spec.authentication.scram, and the quota admin API patches spec.quotas. The operator then materializes the change as usual. That is why broker RBAC carries the full verb set on kafkatopics (including create and delete), get,update,patch on kafkatopics/status, and read plus update,patch on kafkausers. Why admin writes route through CRs at all is covered in Broker/operator runtime independence.

ArgoCD and runtime-created topics

A topic created over the Kafka protocol exists as a KafkaTopic CR that no GitOps tool put there: it has no tracking metadata and no owner references, so in an ArgoCD-managed cluster it is invisible in the Application tree — and hand-adding ArgoCD's tracking label would be worse, because a tracked resource absent from git is exactly what sync-with-prune deletes.

Setting admin.argocd.enabled: true on the Helm chart opts broker-minted CRs into ArgoCD coexistence. Each one is created with:

• argocd.argoproj.io/tracking-id naming the Application (defaults to the Helm release name), so the topic renders in the Application tree alongside the git-managed resources;
• argocd.argoproj.io/compare-options: IgnoreExtraneous, so ArgoCD does not diff it against git — no drift, no selfHeal prune;
• argocd.argoproj.io/sync-options: Delete=false, so runtime-created topics survive an Application delete.

The last two are chart values (admin.argocd.compareOptions / syncOptions); setting either to "" skips that annotation — e.g. an empty compareOptions deliberately surfaces "this topic is not in git" as drift in the ArgoCD UI. Off by default: non-ArgoCD installs get plain CRs.

This applies only to CRs the broker creates — today, topics. KafkaUser CRs are never created at runtime: the ACL, SCRAM, and quota admin APIs edit the spec.authorization.acls / spec.authentication.scram / spec.quotas of a user that must already exist, and stamping ArgoCD metadata onto a git-managed resource would cause the drift the annotations exist to avoid. Runtime ACL edits to git-managed users therefore still show as drift until the next sync — the intentional trade described in the ACL API notes.

Why there are no finalizers

Earlier versions used kaas.rs/*-cleanup finalizers that drained on CR delete. ArgoCD's parallel cascade-delete then deadlocked a teardown: the operator pod was deleted before its CRs, and every CR hung forever waiting for a finalizer that nothing would ever clear. The replacement design:

• Owned external resources (Certificates, Services, TLSRoutes, Secrets) carry OwnerReferences — Kubernetes GC handles them with no operator involvement.
• On-disk state (topic dirs, credential entries) is reclaimed by the leader-elected orphan sweep — run at leader election and every five minutes thereafter — which drops anything on the volume with no matching CR.

Deleting the operator, the CRs, or both in any order can no longer wedge — the cost is that on-disk cleanup lands at the next sweep pass, within minutes, rather than synchronously with the delete.

Readiness gate

Broker pods declare the kaas.rs/PartitionsReady readiness gate; the broker patches its own pod condition once the partition directories it needs exist on the volume — keeping a broker out of Service endpoints until the storage it serves from is actually in place.

That gate is the storage-provisioned precondition. The full readiness answer — /readyz returning 200 only once the broker is actually serving its assigned partitions, and the controller's alive set tracking main-runtime liveness rather than pod readiness — is its own topic: see Honest readiness & rollout pacing.

Implementation notes (for contributors)

• CRD types are kube-derive structs in crates/kaas-operator-api/src/; cargo xtask gen-crds regenerates the YAML into deploy/crds/ and the chart copy — CI fails on drift.
• Reconcilers and the startup sweep live in crates/kaas-operator-controllers/.
• The Strimzi-shape KafkaUser auth/authz surface landed in gh #135, which also removed the earlier KafkaACL / KafkaUserGroup CRs.
• Broker RBAC is deploy/helm/kaas/templates/broker-rbac.yaml — check it whenever a new admin write path lands.
• The readiness-gate patcher is crates/kaas-k8s/src/readiness.rs.

Honest readiness & rollout pacing

/readyz means "serving every assigned partition"; a separate liveness signal feeds the controller's alive set — together they pace rolling updates and evict wedged brokers.

In Apache Kafka, a rolling restart is paced by replication: you (or Strimzi) roll one broker, wait for under-replicated partitions to drain back to zero, then roll the next — the ISR is the signal that it is safe to continue. kaas has no replicas, so that signal doesn't exist. What paces a kaas rollout instead is the Kubernetes readiness probe, which means /readyz has to answer a precise question: is this broker serving the partitions it was assigned? Getting that answer right is what lets a StatefulSet rolling update pace itself — and getting it wrong is what once let two brokers go out of service at the same time, and let a wedged broker sit undetected for 25 minutes.

Two signals, not one

A booting broker and a wedged broker look identical from the outside: both are NotReady, both are still heartbeating. They demand opposite treatment — the booting one must be kept (so it can take over), the wedged one evicted (so its partitions move). No single readiness bit separates them, so the broker publishes two:

The crucial subtlety: serving cannot detect a wedge. When the main runtime seizes up — the observed failure was both worker threads pinned on a synchronous NFS scan under a 2-CPU limit — the partitions stay open in the engine, so serving still reads true. The thing that actually dies is the runtime's ability to run tasks, which is exactly what the healthy tick measures: no worker free to bump the tick → it goes stale → the broker stops reporting healthy.

/readyz = listeners-bound
        AND main runtime alive           (not wedged)
        AND (cluster ? serving : true)   (takeover complete)

/readyz (and /healthz) are served from a dedicated thread and runtime, never the main runtime they report on. That is what makes the wedge observable: the handler can still answer while the main runtime is pinned, and it answers unready, because the liveness tick it reads has gone stale.

The circular dependency, and how healthy breaks it

Gating /readyz on serving looks like it should deadlock, and with the old alive set it did:

flowchart TD
    boot["broker boots, listeners bind"] --> ready0["/readyz honest:<br/>NotReady until takeover done"]
    ready0 --> es["EndpointSlice: NotReady<br/>→ dropped from readiness"]
    es --> alive0["alive set filters on readiness<br/>→ broker excluded"]
    alive0 --> noassign["controller assigns it<br/>zero partitions"]
    noassign --> notake["nothing to take over"]
    notake --> ready0
    style ready0 fill:#fee,stroke:#c33
    style alive0 fill:#fee,stroke:#c33

The fix is to make the alive set depend on healthy, not readiness. A booting broker's main runtime is running (it is busy taking over), so it reports healthy = true throughout boot and stays assignable — even while its /readyz is deliberately NotReady. A wedged broker reports healthy = false and drops out. Readiness is freed to be honest.

healthy travels on the heartbeat, which runs on the broker's control-plane runtime — a separate runtime from the main one. That is deliberate: the heartbeat survives a main-runtime wedge (so the broker can still be told things), which is precisely why healthy has to be an explicit bit rather than "is the heartbeat connected". A connected heartbeat proves the control runtime is alive; only the tick proves the main runtime is.

A rolling update, end to end

sequenceDiagram
    participant SS as StatefulSet<br/>controller
    participant N as kaas-2<br/>(restarting)
    participant HB as kaas-2 heartbeat<br/>(control runtime)
    participant CTL as kaas-0<br/>(cluster controller)
    participant HZ as kaas-2 /readyz<br/>(dedicated runtime)

    SS->>N: delete + recreate (image bump)
    N->>N: listeners bind
    N->>N: main-runtime liveness tick starts
    HB->>CTL: BrokerStatus{ healthy = true }
    Note over CTL: alive set = connected ∧ healthy<br/>→ kaas-2 is assignable
    CTL->>N: assignment.json: kaas-2 leads P0..Pk
    N->>N: takeover: open + recover P0..Pk<br/>(off the main runtime)
    SS->>HZ: GET /readyz
    HZ-->>SS: 503 — alive but NOT serving yet
    Note over SS: minReadySeconds timer cannot start<br/>→ kaas-1 is NOT killed
    N->>N: takeover completes → serving
    SS->>HZ: GET /readyz
    HZ-->>SS: 200 — serving
    Note over SS: continuously Ready ≥ minReadySeconds<br/>→ now safe to roll kaas-1
    SS->>SS: proceed to kaas-1

Contrast the wedge case: after the image bump kaas-2's main runtime seizes. The tick stops, healthy flips to false, and the controller drops kaas-2 from the alive set and reassigns its partitions to a live peer — in seconds, off the heartbeat, rather than waiting ~25 minutes for a tcpSocket liveness probe to notice. Meanwhile kaas-2's /readyz, answered from its still-responsive dedicated runtime, reports 503.

Rolling-upgrade note

healthy is trusted unconditionally: a connected broker reporting false is evicted from the alive set. Fast failover comes from the heartbeat connection itself — a genuinely dead broker drops its stream and vanishes from the alive set within a heartbeat — while a wedged one is evicted on its next healthy = false report. One consequence of the pre-v1 no-backwards-compatibility policy: broker images that predate the healthy signal always report false, so a rolling upgrade from one is unsupported — deploy fresh instead.

Belt and braces: minReadySeconds

broker.minReadySeconds (default 60) makes the StatefulSet wait for that many seconds of continuous readiness before rolling the next pod. With honest readiness it is a safety margin rather than the mechanism — a readiness flap during a late-breaking takeover resets the timer and re-paces the rollout for free.

Implementation notes (for contributors)

The incidents behind this design: gh #208 (two brokers out of service at once), gh #211 (25-minute undetected wedge), gh #209/#210 (the main-runtime seize on a synchronous NFS scan; takeover moved off the main runtime).

• crates/kaas-observability/src/health.rs — compute_ready, the record_main_tick/main_alive liveness tick, RuntimeState::serving.
• crates/kaas-broker/src/coordinator.rs — is_serving (assigned ⊆ open).
• crates/kaas-storage/src/engine.rs — open_partition_keys on the trait.
• bins/kaas/src/main.rs — the dedicated health runtime + the main-runtime tick task.
• bins/kaas/src/cluster.rs — decide_alive, the healthy-gated alive-set policy.
• crates/kaas-controller/src/heartbeat_server.rs — per-broker healthy / broker_liveness().
• proto/heartbeat.proto — BrokerStatus.healthy (field 6). The earlier sticky ever_healthy guard tolerated images predating field 6 (proto3 default false); it was dropped under the pre-v1 no-backcompat policy — see docs/RELEASING.md.

Broker fencing

A Kafka cluster has three answers to "what happened to that broker?", and only two of them are visible in a Metadata response. The broker is serving, it is gone, or it is registered but not serving — Kafka calls the third state fenced. Metadata omits fenced brokers entirely, so a client that only asks Metadata sees a three-broker cluster become a two-broker cluster, with nothing to say whether that was a scale-down or a crash.

kaas answers all three. This page is how.

What "registered" means here

In Apache Kafka (KRaft), a broker registers with the controller quorum, gets a broker epoch, and heartbeats. Miss broker.session.timeout.ms and the controller fences it: still registered, no longer leader-eligible, dropped from Metadata. Deregistration is a separate, explicit act — an operator running kafka-cluster.sh unregister for a broker that is never coming back.

kaas has no metadata quorum to register against; controller election runs on a Kubernetes Lease (see Non-goals). So it registers against the thing that already knows which brokers are supposed to exist: the headless Service's EndpointSlices. The broker watches them anyway, to learn where its peers are.

That gives the three states without inventing a registry:

Kubernetes supplies the hard part for free. A scale-down removes the endpoint, which is the deregistration — no unregister verb, no operator ceremony, no tombstone to garbage-collect.

Where each state is produced

BrokerRegistry (crates/kaas-k8s/src/endpoints.rs) keeps every endpoint the slice lists, carrying its readiness rather than filtering on it. Two rules make that safe:

• An ordinal absent from its slice is deregistered. A real EndpointSlice update always carries the slice's whole membership, so "absent" is meaningful. Without this, keeping not-ready entries would leak a scaled-away broker into every response for the life of the process. The removal is scoped to the slice that owns the ordinal, because a Service's endpoints may be sharded across several slices and a naive sweep would have each slice evict the others' brokers.
• Self is pinned. This broker is never inserted, downgraded, or removed by slice data. A readiness blip on its own pod must not make it forget it exists — that failure was observed live, and it is self-sustaining: self eviction → the controller balances over an empty set → every partition unassigned → the resulting takeover storm fails the next probe too.

The controller turns that into cluster-wide state. assignment.json's broker list has always had the shape for it —

#![allow(unused)]
fn main() {
pub enum BrokerHealth { Alive, Draining, Dead }
}

— and now has a producer: every registered broker gets a row, marked Alive if the heartbeat says so, Draining if it announced a shutdown, and Dead otherwise. A fenced broker keeps its last_seen from when it was last alive, rather than being refreshed to "just now" on every recompute.

The coordinator-divisor bug this fixed

Reporting was the motivation; correctness was the surprise.

group_hash picks a group's coordinator with hash(groupID) % num_brokers, and its documentation is emphatic that the divisor must be the full broker set, "including draining / dead" — holding it constant is what keeps group coordinatorship stable across restarts. But the list it divides by came from assignment.json, and that list was the alive set: a dead broker was dropped rather than marked. So losing one broker of three silently changed the divisor from 3 to 2 and rehashed roughly two-thirds of all group and transaction coordinators — precisely when the cluster was already degraded and least able to absorb the churn.

With the tri-state, only the groups that actually lived on the lost broker move. a_dead_broker_moves_only_its_own_groups in crates/kaas-broker/src/assignment.rs pins that: of 200 groups across three brokers, a broker loss moves the ~1/3 that hashed to it and leaves the rest where they were.

Draining: fencing from the other direction

A broker that is shutting down is not unhealthy — it serves normally right up until its listeners close. But it is leaving, and the controller used to find that out the slow way, by timing out a heartbeat from a process that had already exited.

The SIGTERM path now calls mark_draining() before anything is torn down. The next heartbeat (~1 s) carries draining = true; the controller drops the broker from the alive set immediately — moving its partitions and group coordinatorships while it is still healthy enough to hand them over — and marks its row Draining rather than Dead.

Two properties are load-bearing. The broker stays in the registered list, so the divisor above doesn't move. And the self-pin in the alive-set policy outranks draining: a draining controller keeps itself in the set, because an empty alive set would unassign the entire cluster.

This is the proactive half of graceful shutdown — controlled shutdown and fencing are the same feature approached from two sides.

What clients see

Metadata omits fenced brokers, exactly as Apache does: advertising one would send clients to a broker that cannot answer.

DescribeCluster v2 reports them, with IsFenced per row, when the request sets IncludeFencedBrokers. That version is KIP-1073 — Kafka 4.0 surface, and a deliberate exception to kaas's Apache 3.7 parity target, taken because there is now a real fenced state and no other way to report it.

The broker answering a request never reports itself as fenced. It is serving — it just answered.

Notes for contributors

• BrokerHealth::Dead vs absent from the list is the fenced/gone distinction. Don't "clean up" dead rows: they are the divisor.
• Fenced is derived from EndpointSlice readiness, so a booting broker reads as fenced until takeover completes and /readyz flips. That is deliberate and matches Apache, which fences a registered broker until it has caught up. It also means a rolling restart shows a fenced broker at each step — which is the honest report, not a defect.
• Readiness and the alive set are still separate signals, and conflating them is the gh #208 trap: the alive set is driven by the heartbeat's healthy bit, not by readiness, so a booting broker stays assignable even while it reads as fenced for reporting.
• Relevant source: crates/kaas-k8s/src/endpoints.rs (registry), crates/kaas-controller/src/assignment_writer.rs (build_broker_entries), bins/kaas/src/cluster.rs (decide_alive, the registered/draining sources), crates/kaas-broker/src/handlers/describe_cluster.rs (the wire surface).

Observability

OTLP metrics and tracing pushed to Prometheus, and the /healthz endpoint's rich runtime state.

Apache Kafka exposes its metrics as JMX MBeans, and most deployments bolt a Prometheus JMX exporter onto every broker to scrape them. kaas is OpenTelemetry-native instead: brokers push OTLP metrics straight to Prometheus's native OTLP receiver, emit OTel spans, and correlate every log line with the active trace — no sidecar, no exporter agent.

Bootstrap: push-mode OTLP

The OTel SDK is wired from environment variables (all emitted by the Helm chart):

• OTEL_EXPORTER_OTLP_METRICS_ENDPOINT — metrics are pushed via OTLP/HTTP (http/protobuf) to Prometheus's native OTLP receiver (/api/v1/otlp/v1/metrics). Prometheus speaks only http/protobuf on that path — exporting gRPC just gets an h2 GoAway.
• OTEL_EXPORTER_OTLP_ENDPOINT — traces via OTLP/gRPC.
• KAAS_METRIC_EXPORT_INTERVAL — a duration string (30s, 1m); default 30s. The SDK default of 60 s made rate([1m]) panels see a single sample and gap.
• OTEL_SERVICE_VERSION, MY_POD_NAME, KAAS_NAMESPACE — folded into resource attributes so dashboards can filter by broker and namespace.
• OTEL_TRACES_SAMPLER_ARG — trace sampling ratio in [0, 1], default 0.1.

With neither endpoint set the SDK still runs and exports go nowhere — safe for tests and local dev. Before bootstrap runs, the global metrics handle is a no-op registry, so pre-boot code and tests can record without nil checks.

One metrics-pipeline invariant is load-bearing enough to repeat here: gauge callbacks (high watermark, log start offset) read the storage engine through a lock-free snapshot, never the partition mutex. Before that split, a stuck NFS fsync holding the mutex stalled the OTel gauge callback and all broker metrics vanished from Prometheus until the stall cleared — exactly when you needed them most.

/healthz and /readyz

Both are served on port 8080:

• /readyz is the kubelet readiness probe, and it is honest: listeners bound, the main runtime alive, and — in cluster mode — every partition the assignment gives this broker open in the storage engine, i.e. takeover complete. It is served from a dedicated thread and runtime, never the main runtime it reports on, so a wedged broker answers unready instead of hanging the probe. The full story — the two signals, and how readiness paces a rolling update — is Honest readiness & rollout pacing.
• /healthz returns a JSON runtime view: controller identity (is_controller, controller_id, controller_epoch), heartbeat health (heartbeat_rtt_ms, heartbeat_age_ms), assignment state (assignment_version, assignment_age_ms), partition counts (partitions_led, partitions_assigned, partitions_recovering), and storage_stalled — true when any partition's most recent committer fsync tripped the stall watchdog, surfacing "storage backend wedged" before the broker accumulates enough queued appenders to look outwardly idle.

Fields with no measurement yet return -1 internally and render as JSON null, so dashboards show "not measured" rather than a misleading zero. In local-dev mode (no cluster runtime) the whole runtime view is absent and the handler serves zero-valued fields — the right answer when no controller, coordinator, or heartbeat client is running.

partitions_led sources from the same assignment.json-backed view of leadership as the Produce/Fetch ownership check — /healthz never invents its own answer to "who leads what".

Byte-opacity tripwires

The broker's load-bearing invariant is "kaas is a byte mover, not a byte interpreter": no code path decodes individual records or re-encodes a RecordBatch (see Storage engine hot path). Two counters give that invariant teeth — codec_record_decode and codec_batch_reencode — which must stay at zero in steady state. Every increment names the offending call site in a site attribute, and the KaasByteOpacityViolated alert fires on non-zero.

As of today no kaas code path calls these functions; the counters exist so the first violation is an alert, not an archaeology project.

Tracing

One global tracing stack is installed at boot: a filter honouring KAAS_LOG_LEVEL, a log formatter honouring KAAS_LOG_FORMAT (json or text), and an OTel layer that emits every span through the tracer from bootstrap. The correlation contract: every log line carries trace_id + span_id whenever a span is active, so a log grep pivots straight into the trace view. Span sampling defaults to 10% — set OTEL_TRACES_SAMPLER_ARG=1.0 for a debugging session.

Implementation notes (for contributors)

• Everything lives in crates/kaas-observability: bootstrap.rs (OTel SDK bring-up), health.rs (the axum /healthz//readyz surface; the JSON view is the RuntimeState trait), byteopacity.rs (tripwire counters), tracing.rs (install_tracing, the tracing-subscriber stack).
• Codec-side tripwire counterparts: crates/kaas-codec/src/tripwires.rs; CI assertion: bins/kaas/tests/byte_opacity.rs.
• Two standing rules: pick KAAS_METRIC_EXPORT_INTERVAL with the Prometheus rate() window in mind, and gauge callbacks must read the engine's ArcSwap snapshot, never the partition mutex.
• gh #95 — the fsync stall watchdog behind storage_stalled; gh #208/#211 — the honest, serving-gated /readyz (see readiness & rollout).

Wire protocol & framing

Length-prefixed frames, KIP-482 flexible versions with tagged fields, and byte-opaque RecordBatch handling — the codec layer under every parity claim in this part of the book.

Point an unmodified Kafka client at kaas — Java, librdkafka, franz-go, the stock shell tools — and everything it sends and receives passes through the codec described here. The codec's standing claim: every registered API key and version encodes and decodes byte-identically against fixtures captured from Apache Kafka 3.7. The rest of Part II substantiates compatibility surface by surface; this page covers the layer all of those surfaces share.

Framing and headers

Kafka's transport is length-prefixed frames: a 4-byte big-endian size, then the message. Inside a frame:

• Request header — API key, API version, correlation ID, client ID, and (for flexible versions) tagged fields.
• Response header — correlation ID, plus tagged fields when flexible.

One subtlety Kafka clients normally hide from you: the header's own version depends on (api key, api version). kaas resolves it through a per-API header-version function carried on each registry entry, so every key answers with exactly the header shape Apache Kafka 3.7 would use.

KIP-482: flexible versions and tagged fields

From a per-API cutover version, Kafka switches to "flexible" encoding: compact strings and arrays (varint lengths) and tagged fields — an extensible (tag, size, bytes) section that lets brokers and clients attach optional data without a version bump.

Which version each API flips to flexible is data, not code: the "Flexible" column in the generated API matrix comes straight from the codec registry, and the same registry table drives the ApiVersions response — the matrix cannot disagree with what the broker advertises.

The byte-opacity contract

There is no Record struct in kaas — deliberately. RecordBatch payloads flow through the codec as opaque bytes: a zero-copy slice of the request frame on the way in, the same bytes off disk on the way out. The only code that reads past the fixed v2 batch header is CRC verification (CRC32C over opaque input) and a batch-header walker that counts records without decoding them.

Consequences worth internalizing:

• The log file is the wire format — Produce appends the client's bytes verbatim (base offset rewritten in place), Fetch returns them verbatim (see Storage engine hot path).
• Per-record features (timestamps, headers, per-record validation) are honoured byte-opaquely: kaas preserves what producers wrote without interpreting it. Where Apache applies per-record semantics — e.g. tombstone expiry during compaction — kaas will apply the batch-level equivalent (the compactor itself is not yet implemented).
• The contract is enforced, not aspirational: tripwire counters (surfaced as metrics — see Observability) must read zero after every test run, and any future record-decoding code path is required to bump them, making the first violation loud.

Where to dig deeper

• API support matrix — every served key, generated from the registry.
• Per-API reference — semantics and deviations per key.
• KIP index — protocol-evolution KIPs and where kaas stands on each.
• Non-goals — the keys and features that are absent on purpose.
• Verification story — the suites that would catch any of this being wrong.

Implementation notes (for contributors)

Everything wire-level lives in crates/kaas-codec:

• crates/kaas-codec/src/frame.rs — frame reader/writer, plus the streaming FrameReader used by the server accept loop.
• crates/kaas-codec/src/headers.rs — per-API HeaderVersion resolution.
• crates/kaas-codec/src/tagged.rs — the KIP-482 tagged-field envelope; crates/kaas-codec/src/primitives.rs — compact strings, arrays, and the other wire primitives.
• crates/kaas-codec/src/crc.rs — CRC32C over opaque batch bytes; crates/kaas-codec/src/recordbatch_count.rs — the batch-header walker.
• crates/kaas-codec/src/api/ — one request/response module per API; crates/kaas-codec/src/api/registry.rs — the ApiSpec table that drives both the ApiVersions response and the generated matrix.
• crates/kaas-codec/src/tripwires.rs — the byte-opacity counters, asserted zero after real traffic by bins/kaas/tests/byte_opacity.rs.

API support matrix

kaas registers 40 Kafka API keys. This table is generated from the ApiSpec registry (crates/kaas-codec/src/api/registry.rs) — the same table that builds the ApiVersions response — so the version ranges below are the wire truth, not documentation aspiration. "Flexible" is the first version using KIP-482 flexible encoding (see Wire protocol & framing).

Apache 3.7 keys kaas does not serve

Clients discover the served surface via ApiVersions, so an absent key is a clean "unsupported", not an error path. Each absence is either a tracked follow-up or a documented non-goal:

Inter-broker/KRaft keys (LeaderAndIsr, StopReplica, UpdateMetadata, ControlledShutdown, the quorum/Envelope family), delegation-token keys, and tiered-storage-only surfaces are deliberately absent — see Non-goals.

Per-API reference

Grouped-by-domain reference for each implemented API key: versions, semantics, deviations from Apache Kafka 3.7, source paths, test coverage.

The generated matrix tells you which versions of each API kaas serves; these pages tell you how it behaves — the semantics behind each key, and every place behaviour differs from Apache Kafka 3.7, stated plainly. Every registered key is documented across seven domain pages, grouped so related behaviour (and shared deviations) read in one place, with a stable anchor per key that the matrix's "Reference" column links to:

• Produce, Fetch, ListOffsets & Metadata — the data plane
• Consumer-group APIs — find/join/sync/commit and the admin group surface
• Transaction APIs — the EOS surface
• Topic & config admin APIs
• ACL & quota admin APIs
• SASL authentication APIs
• Cluster & log-dir APIs

Every anchor follows the same template:

1. Purpose — what the API does in one or two sentences.
2. Versions — the supported range, matching the registry (the matrix is generated from it, so a mismatch here is a doc bug by definition).
3. Handling — how the request flows through kaas.
4. Deviations — where behaviour differs from Apache Kafka 3.7, stated plainly.
5. Source — handler and codec paths (contributor material, kept out of the narrative).
6. Verified by — unit/integration tests and scripts/kafka-*.sh scenarios.

Produce, Fetch, ListOffsets & Metadata

Per-API reference — see the API support matrix for the generated version table.

Produce

Append RecordBatches to partition logs. This is the write hot path: the storage engine chapter covers the mechanics end to end; this section covers the wire contract.

Versions: v3–v9 (flexible from v9)

Four gates run before any byte reaches the engine, per partition: the topic must exist in the registry (else UNKNOWN_TOPIC_OR_PARTITION, 3), the broker Coordinator must own the partition per the applied assignment.json (else NOT_LEADER_FOR_PARTITION, 6), the controller-connectivity self-fence must see a fresh controller heartbeat — a broker cut off from the controller stops acking within 3 s (also 6) — and the cluster-wide authorizer must grant topic Write (else TOPIC_AUTHORIZATION_FAILED, 29). A batch larger than Apache's message.max.bytes default (1 MiB + 12 bytes batch overhead) is rejected with MESSAGE_TOO_LARGE (18) before it hits storage.

Inside the engine, under the partition mutex: idempotent-producer classification runs first — a duplicate (PID/epoch/sequence already in the 5-batch ring window) echoes the cached base offset with error_code = 0 and never touches the log; out-of-order sequences get OUT_OF_ORDER_SEQUENCE_NUMBER (45); a stale producer epoch gets INVALID_PRODUCER_EPOCH (47). Accepted batches have their base offset rewritten in place to the current high watermark (the v2 CRC covers byte 21 onward, so the 8-byte overwrite is wire-correct) and the client's bytes land on disk verbatim — the broker never parses records, only fixed-size header peeks. For acks = -1, the append that crosses the flush threshold parks on the per-partition committer's group-commit fsync — one sync_all() cycle serves every concurrent appender — with KAAS_FLUSH_INTERVAL_MESSAGES (default 1) as the durability dial, overridable per topic via the flush.messages topic config.

The produce quota is checked once per request over the summed record bytes, after the appends, and feeds throttle_time_ms in the response.

Deviations from Apache 3.7:

• acks = 0 still gets a wire response. Apache sends nothing for acks=0 produce requests; kaas's dispatcher has no response-suppression path, so the handler's response is always written back.
• The message.max.bytes cap is the fixed Apache default — the per-topic max.message.bytes config override is not consulted.
• log_append_time_ms is always -1: the broker never stamps append time, so message.timestamp.type=LogAppendTime semantics are not implemented.
• The v8+ record_errors / error_message fields are never populated — errors are reported per partition, not per record.

Source: crates/kaas-broker/src/handlers/produce.rs, crates/kaas-codec/src/api/produce.rs; engine side in crates/kaas-storage/src/partition.rs and crates/kaas-storage/src/idempotence.rs.

Verified by: unknown_topic_returns_error_code_3 (handler unit test), produce_fetch_metadata_roundtrip in bins/kaas/tests/smoke.rs, storage_round_trip_is_byte_identical in bins/kaas/tests/byte_opacity.rs, and the full transactional round trip in bins/kaas/tests/eos_v2.rs. Shell suite: scripts/kafka-console-producer.sh, scripts/kafka-producer-perf-test.sh, scripts/kafka-verifiable-producer.sh.

Fetch

Read RecordBatches from partition logs. The symmetric half of the byte-opacity contract: batch bytes come back off disk undecoded (see the storage hot path).

Versions: v4–v12 (flexible from v12)

The front door mirrors Produce: topic-exists, Coordinator::owns, and a cluster-wide topic Read ACL check, per partition (there is no self-fence gate on reads). The handler then picks the read cap: the high watermark for read_uncommitted, the last stable offset for read_committed (the LSO is the highest offset with no in-flight transaction extending past it). The engine copies batch bytes from the segment files into memory; for read_committed the result is trimmed to whole batches strictly below the LSO — a batch that straddles the cap is dropped along with everything after it, since batches are atomic — and aborted_transactions is populated from the partition's aborted-txn index over the returned range, so the client can filter aborted records. A reader already at or past the cap gets an empty response. Out-of-range offsets map to OFFSET_OUT_OF_RANGE (1). The fetch quota is checked once over the summed response bytes and feeds throttle_time_ms.

The response is materialized bytes — copied from the segments, not sendfile/spliced. The codec keeps records byte-opaque precisely so a splice path stays possible; today it is a future optimisation, not what ships.

Deviations from Apache 3.7:

• No KIP-227 fetch sessions: session_id = 0 on every response, regardless of what the client sent. That is Apache's documented marker for "broker doesn't support sessions", so clients fall back to full fetch requests per poll — a deliberate contract, not a gap (non-goals). session_epoch and forgotten_topics are decoded and ignored.
• No long-poll: max_wait_ms and min_bytes are decoded but ignored. An empty fetch returns immediately instead of parking until data arrives or the wait expires; clients simply re-poll.
• The leader-epoch fields (current_leader_epoch, last_fetched_epoch) are not validated — kaas never returns FENCED_LEADER_EPOCH or UNKNOWN_LEADER_EPOCH, and KIP-320 truncation detection does not apply (there are no followers to diverge from).
• preferred_read_replica is always -1 and rack_id is ignored — no follower fetching, because there is no replication (non-goals).

Source: crates/kaas-broker/src/handlers/fetch.rs, crates/kaas-codec/src/api/fetch.rs; aborted-txn index in crates/kaas-storage/src/txn_index.rs.

Verified by: the trim_to_offset_* unit tests and unknown_topic_returns_error_3 (asserts session_id == 0) in the handler, produce_fetch_metadata_roundtrip in bins/kaas/tests/smoke.rs, and the read-committed assertions eos_commit_path_records_visible_to_read_committed / eos_abort_path_populates_aborted_transactions in bins/kaas/tests/eos_v2.rs. Shell suite: scripts/kafka-console-consumer.sh, scripts/kafka-consumer-perf-test.sh, scripts/kafka-e2e-latency.sh.

ListOffsets

Resolve a timestamp (or a sentinel) to an offset per partition — what kafka-get-offsets.sh and every consumer's auto.offset.reset ride on (KIP-32 semantics).

Versions: v1–v7 (flexible from v6)

The handler checks the topic exists (else UNKNOWN_TOPIC_OR_PARTITION, 3), then translates the request timestamp: -2 (EARLIEST) returns the partition's log_start_offset, -1 (LATEST) returns the high watermark — both echoed with timestamp = -1, matching Apache's sentinel-response shape. Any other timestamp is passed to the engine's offset_for_timestamp.

Deviations from Apache 3.7:

• Timestamp→offset lookup is not implemented. Segments track their max_timestamp, but the engine-level index that maps a timestamp to an offset is a follow-up: offset_for_timestamp returns the (-1, -1) "no matching offset" sentinel unconditionally, in both the disk and in-memory engines. Every non-sentinel timestamp query — including MAX_TIMESTAMP (-3, nominally in the registered v7 range) — gets offset = -1, timestamp = -1 with error_code = 0. Only the -1/-2 sentinels resolve.
• isolation_level is decoded but ignored: LATEST returns the high watermark even for read_committed clients, where Apache returns the last stable offset.
• The tiered-storage-only sentinels (EARLIEST_LOCAL_TIMESTAMP, EARLIEST_PENDING_UPLOAD_OFFSET) are deliberately unsupported — clients only send them when configured for a remote tier (non-goals); on the wire they fall into the same (-1, -1) path as any other timestamp.
• leader_epoch is always 0 in success responses.
• No authorization or leadership gate: there is no topic Describe ACL check, and no Coordinator::owns check — a non-leader broker answers from its local engine view (typically offset 0) instead of NOT_LEADER_OR_FOLLOWER. Clients that route by Metadata leadership never hit this in steady state.

Source: crates/kaas-broker/src/handlers/list_offsets.rs, crates/kaas-codec/src/api/list_offsets.rs; the sentinel-returning engine lookup in crates/kaas-storage/src/disk.rs.

Verified by: handler unit tests latest_returns_high_watermark, earliest_returns_log_start_offset, unknown_topic_returns_3. Shell suite: scripts/kafka-get-offsets.sh (exercises the -2/-1 sentinels before and after producing — the paths clients actually use).

Metadata

The cluster-discovery API: which brokers exist, who leads each partition, and where to connect — the response every client bootstraps from.

Versions: v1–v10 (flexible from v9)

Per-listener advertisement: the handler precomputes one advertised (host, port) per configured listener from the KAAS_LISTENERS env, and each connection carries its listener name, so a client that bootstrapped on :9095 gets :9095 back — not the anonymous listener's port. Without this, authed-listener clients were routed back to the plain listener and looped on SCRAM retry (Listeners & auth). Peer brokers are advertised at their stable in-cluster FQDN with the port of the listener the client connected on; per-broker external hostname templates for peers are a follow-up.

Leadership comes from assignment.json via the broker Coordinator: each partition's leader_id is the assignment entry's broker ordinal, and controller_id is parsed from the assignment's controller identity (Controller). Dev mode — and any partition missing from the assignment (fresh topic, recompute pending) — falls back to self. An empty request topic list returns every known topic (Apache's "all topics" contract); unknown requested topics get a per-topic UNKNOWN_TOPIC_OR_PARTITION (3). replica_nodes and isr_nodes are [leader] — the truthful shape for a single-writer design with no replication (non-goals).

Deviations from Apache 3.7:

• Topic IDs serve the all-zero sentinel on the wire. The v10 topic_id field is encoded, but the value is always the null UUID: the operator mints a real v4 UUID into each KafkaTopic's Status.TopicID, and the broker-side observation plumbing exists, yet the watcher callback that feeds the topic registry inserts [0; 16]. KIP-516 is partial: minted operator-side, not yet propagated to the Metadata response.
• leader_epoch is always 0 (v7+ field) — clients that cache epochs for truncation detection get a constant.
• allow_auto_topic_creation is decoded and ignored. There is no auto-creation: topics exist when a KafkaTopic CR exists (authored directly or via CreateTopics, which writes the CR).
• topic_authorized_operations / cluster_authorized_operations are always 0, even when the v8+ request flags ask for them — the ACL bitsets are not computed.
• rack is always null and is_internal always false — kaas advertises no internal topics (there is no on-wire __consumer_offsets).
• v11+ is not registered; clients negotiate down to v10 via ApiVersions.

Source: crates/kaas-broker/src/handlers/metadata.rs, crates/kaas-codec/src/api/metadata.rs; the registry-feeding watcher callback in bins/kaas/src/main.rs.

Verified by: handler unit tests per_listener_port_echoed_back, returns_self_as_only_broker_and_leader, empty_topic_list_returns_all_known, unknown_topic_returns_per_topic_error_3; produce_fetch_metadata_roundtrip in bins/kaas/tests/smoke.rs. Shell suite: scripts/kafka-topics.sh (--list / --describe ride Metadata).

Consumer-group APIs

Per-API reference — see the API support matrix for the generated version table.

Every API on this page except FindCoordinator is answered only by the group's coordinator broker; any other broker returns NOT_COORDINATOR (16) and the client re-resolves. Coordinator-of-G is two-tier: an explicit assignment.json.consumerGroups[] entry wins, otherwise a deterministic FNV-1a 32-bit hash of the group ID mod the full broker set (not the alive count), with a deterministic fallback into the alive subset when the preferred slot is down. Apache answers the same question with __consumer_offsets partition leadership; kaas has no __consumer_offsets topic, so it hashes directly into the broker set — see Consumer-group coordination for the full routing story.

Where the state lives: membership, generation, and rebalance state are in-memory on the coordinator — lost on coordinator failover, after which consumers rejoin and rebalance, which is exactly what an Apache coordinator move looks like to clients. Committed offsets are durable per-group JSON files at /data/__cluster/__consumer_offsets/<groupID>.json on the shared volume; the next coordinator reads the same file. The offsets file is lazily loaded into memory when a group first joins on a broker (Manager::get_or_create) — a caveat for manual-assignment consumers is noted under OffsetFetch.

During the boot window before the coordinator Manager is installed, handlers return NOT_COORDINATOR / COORDINATOR_NOT_AVAILABLE so clients retry — no request is silently dropped.

OffsetCommit

Persists a group's consumed positions. Key 8.

Versions: v2–v8 (flexible from v8)

Handling: the handler flattens the nested (topic, partition) request into a topic/partition → offset map plus a parallel metadata map and hands both to Manager::offset_commit, which merges them into the group's in-memory cache and atomically rewrites __consumer_offsets/<groupID>.json (tmp + fsync + rename). The resulting group-level error code is stamped uniformly across every partition in the response. Per-partition committed_metadata strings round-trip; empty strings clear the entry and come back as the wire null sentinel.

Deviations from Apache 3.7:

• The advertised floor is v2, not v0 — the v0/v1 shapes were never decoded correctly and are not offered. Additionally, v2–v4's retention_time_ms field is not decoded, so v2–v4 requests that carry it mis-parse; this is an acknowledged divergence tracked as follow-up in the codec module. In practice every modern client negotiates v8 via ApiVersions and never hits it.
• generation_id, member_id, and group.instance.id are decoded but not validated against the group state machine — no ILLEGAL_GENERATION, UNKNOWN_MEMBER_ID, or FENCED_INSTANCE_ID fencing on the commit path. A zombie commit from an evicted member is accepted.
• Commits are best-effort durable: a disk-write failure is logged but still reported as success (Manager::offset_commit).
• committed_leader_epoch (v6+) is decoded but not persisted; fetches always return -1 for it.

Source: crates/kaas-broker/src/handlers/offset_commit.rs, crates/kaas-codec/src/api/offset_commit.rs, crates/kaas-coordinator/src/offset_store.rs.

Verified by: crates/kaas-broker/tests/group_dispatch.rs (full lifecycle round trip), commit_then_fetch_roundtrips / commit_with_metadata_persists in offset_store.rs, scripts/kafka-consumer-groups.sh (--reset-offsets --execute drives OffsetCommit), scripts/kafka-console-consumer.sh.

OffsetFetch

Reads a group's committed positions back. Key 9.

Versions: v1–v8 (flexible from v6)

Handling: v1–v7 carry a single group; v8+ batches multiple groups in one request (groups[]), and the handler resolves each group independently — per-group NOT_COORDINATOR for groups owned elsewhere. Lookups read the coordinator's in-memory cache; partitions without a committed offset return the -1 sentinel. Offsets staged by transactional producers via TxnOffsetCommit live in a separate pending layer and are invisible here until the transaction commits — an aborted transaction's offsets are never observable.

Deviations from Apache 3.7:

• The "fetch everything" sentinel (null topics list) returns an empty result instead of enumerating all committed offsets. Symptom: kafka-consumer-groups --describe on a group with no active members shows nothing even though offsets are on disk (noted in scripts/kafka-consumer-groups.sh, which dropped its --shift-by scenario because of it). Follow-up tracked in the handler.
• require_stable (v7+, KIP-447) is accepted and ignored. kaas never returns UNSTABLE_OFFSET_COMMIT; it returns the last durably-committed offset. Reads can never see dirty in-flight offsets (the pending layer guarantees that), but a caller asking to wait out an in-flight commit doesn't get the retriable error Apache would send.
• Committed offsets are only loaded from disk when the group first joins on this broker. A standalone manual-assignment consumer (assign() + commitSync, no JoinGroup) that fetches after a coordinator change hits a cold cache and gets -1 until its next commit.
• committed_leader_epoch is always -1.
• Advertised floor is v1, not v0 (v0 was the ZooKeeper-era offsets path).

Source: crates/kaas-broker/src/handlers/offset_fetch.rs, crates/kaas-codec/src/api/offset_fetch.rs, crates/kaas-coordinator/src/offset_store.rs.

Verified by: crates/kaas-broker/tests/group_dispatch.rs, pending_invisible_to_fetch_until_commit_pending / discard_pending_drops_unmaterialised_offsets in offset_store.rs, bins/kaas/tests/eos_v2.rs (transactional offsets become visible only after EndTxn), scripts/kafka-consumer-groups.sh.

FindCoordinator

Resolves which broker coordinates a group (or a transactional ID). Key 10. The one group API any broker answers.

Versions: v0–v4 (flexible from v3)

Handling: key_type 0 routes through the group-assignment source (explicit assignment.json entry, else the FNV-1a hash — see the preamble), key_type 1 through the transaction-assignment source (same hash machinery); any other value gets INVALID_REQUEST (42). The resolved broker ID is mapped to (node_id, host, port) via the EndpointSlice-backed broker registry; in dev mode the lookup resolves self only. No manager installed, no txn source yet, or no alive broker for the slot → COORDINATOR_NOT_AVAILABLE (15) and the client retries. v4+ batches multiple coordinator_keys[] into one request; v0–v3 use the legacy single-key shape (at v3 the legacy shape merely gains flexible encoding — the array form is strictly v4+).

Deviations from Apache 3.7:

• The response is not listener-aware: the advertised port is the broker's primary client port (the first configured listener / the headless Service's Kafka port), regardless of which listener the request arrived on. Metadata got per-listener advertisement; FindCoordinator has not, so on a multi-listener cluster a client that bootstrapped on a secondary listener is handed the first listener's port here.

Source: crates/kaas-broker/src/handlers/find_coordinator.rs, crates/kaas-codec/src/api/find_coordinator.rs, crates/kaas-coordinator/src/manager.rs, crates/kaas-broker/src/group_hash.rs; the broker registry in bins/kaas/src/cluster.rs and crates/kaas-k8s/src/endpoints.rs.

Verified by: find_coordinator_resolves_self / find_coordinator_txn_with_no_source_is_unavailable / find_coordinator_unknown_key_type_is_invalid_request in manager.rs, hash-routing tests in group_hash.rs, crates/kaas-broker/tests/group_dispatch.rs, scripts/kafka-consumer-groups.sh.

JoinGroup

Enters a member into the group and drives the rebalance state machine (Empty → PreparingRebalance → CompletingRebalance → Stable). Key 11.

Versions: v2–v9 (flexible from v6)

Handling: the handler translates the codec request into the coordinator's JoinRequest and parks the connection on a oneshot until the rebalance round completes. Joining an Empty group starts the initial rebalance with Apache's 3 s group.initial.rebalance.delay.ms (extended per new arrival, capped at the max member rebalance_timeout_ms); joining a Stable group bounces it back to PreparingRebalance and cancels any in-flight sync round. The leader is the first joiner of the round; protocol selection is leader-first-mutual (first protocol the leader declares that every member also lists). Dynamic members that fail to rejoin within the rebalance timeout are evicted at round completion; static members (KIP-345 group.instance.id) survive a missed rejoin.

Deviations from Apache 3.7:

• KIP-394's v4+ two-step handshake is not implemented: kaas never returns MEMBER_ID_REQUIRED (79). An empty member_id is assigned inline and the join proceeds in one round trip — the legacy pre-v4 path, explicitly marked as a follow-up in the group coordinator's join path. Clients work fine (they simply skip a retry), but Apache's ghost-member protection is absent.
• KIP-345 is partial: group.instance.id is plumbed through join/sync/heartbeat/leave and static members survive the non-rejoin eviction, but there is no FENCED_INSTANCE_ID fencing — two members presenting the same instance ID are treated as two members — and a static rejoin triggers a full rebalance rather than Apache's return-cached-assignment fast path (skip_assignment is always false).
• KIP-800 reason strings (v8+) are decoded and discarded, not logged.
• The advertised floor is v2, not v0 — v0/v1 lack rebalance_timeout_ms and were never decoded correctly.
• Member IDs are generated as <principal>-<counter-hex> from the connection's authenticated principal (empty for anonymous listeners), not Apache's <client.id>-<UUID> — cosmetic, but visible in --describe.

Source: crates/kaas-broker/src/handlers/join_group.rs, crates/kaas-codec/src/api/join_group.rs, crates/kaas-coordinator/src/group.rs.

Verified by: single_member_rebalance_completes_via_initial_delay / shutdown_fires_pending_joiners_with_unknown_member in group.rs, crates/kaas-broker/tests/group_dispatch.rs, scripts/kafka-consumer-groups.sh, scripts/kafka-verifiable-consumer.sh.

Heartbeat

Keeps a member's session alive and signals rebalances in progress. Key 12.

Versions: v0–v4 (flexible from v4)

Handling: each heartbeat re-arms the member's session-timeout task (a tokio timer against the member's session_timeout_ms); expiry evicts the member and bounces a Stable group into PreparingRebalance. Checks run in Apache's order: unknown member (or Empty/Dead group) → UNKNOWN_MEMBER_ID (25); generation mismatch → ILLEGAL_GENERATION (22); group mid-rebalance → the timer is still reset, then REBALANCE_IN_PROGRESS (27) tells the member to rejoin; otherwise success.

Deviations from Apache 3.7:

• group.instance.id (v3+, KIP-345) is decoded but not used for fencing — no FENCED_INSTANCE_ID here either.

Source: crates/kaas-broker/src/handlers/heartbeat.rs, crates/kaas-codec/src/api/heartbeat.rs, crates/kaas-coordinator/src/group.rs.

Verified by: heartbeat_unknown_member_for_empty_group in group.rs, not_coordinator_on_offset_commit_when_source_says_no in manager.rs (also covers heartbeat's NOT_COORDINATOR path), crates/kaas-broker/tests/group_dispatch.rs, any live consumer in scripts/kafka-console-consumer.sh / scripts/kafka-verifiable-consumer.sh.

LeaveGroup

Removes one or more members from a group. Key 13.

Versions: v0–v5 (flexible from v4)

Handling: the handler collects member IDs from both the legacy single member_id field (v0–v2) and the v3+ members[] batch (KIP-345 admin removal shape), then removes them in one pass: each known member is dropped (its session timer aborted, any parked JoinGroup waiter woken), unknown members get per-member UNKNOWN_MEMBER_ID (25). The last member leaving returns the group to Empty; a leave from Stable triggers PreparingRebalance.

Deviations from Apache 3.7:

• On the legacy v0–v2 single-member shape the per-member result is dropped: the top-level error_code is 0 whenever this broker coordinates the group, so an unknown-member leave at v0–v2 reports success. The v3+ batch shape carries per-member codes correctly.
• Leaving a group the broker coordinates but holds no state for returns top-level success with an empty member list rather than per-member UNKNOWN_MEMBER_ID.
• KIP-800 per-member reason strings (v5) are decoded and discarded.

Source: crates/kaas-broker/src/handlers/leave_group.rs, crates/kaas-codec/src/api/leave_group.rs, crates/kaas-coordinator/src/group.rs.

Verified by: leave_drops_state_back_to_empty in group.rs, v5_java_client_fixture_with_reason in the codec module (byte-level fixture captured from a real Java client), crates/kaas-broker/tests/group_dispatch.rs, scripts/kafka-consumer-groups.sh.

SyncGroup

Distributes the leader-computed assignment to every member of a rebalance round. Key 14.

Versions: v0–v5 (flexible from v4)

Handling: the leader's SyncGroup stores per-member assignments on the current sync round, flips the group to Stable, and wakes every parked follower; followers park until the leader delivers (or a fresh JoinGroup cancels the round, in which case they wake with REBALANCE_IN_PROGRESS (27) instead of a bogus empty assignment). Checks: unknown member → UNKNOWN_MEMBER_ID (25); generation mismatch → ILLEGAL_GENERATION (22). A follower's sync is valid in both CompletingRebalance and Stable (the leader may finish first). Members the leader omitted receive a zero-byte assignment.

Deviations from Apache 3.7:

• v5's protocol_type / protocol_name request fields are accepted but not cross-checked against the group — kaas never returns INCONSISTENT_GROUP_PROTOCOL from SyncGroup.
• Omitted members get raw empty bytes; encoding a valid empty ConsumerProtocolAssignment struct instead is a follow-up (tracked as gh #111).
• No FENCED_INSTANCE_ID for static members (KIP-345), same as the rest of the group surface.

Source: crates/kaas-broker/src/handlers/sync_group.rs, crates/kaas-codec/src/api/sync_group.rs, crates/kaas-coordinator/src/group.rs.

Verified by: sync_returns_leader_supplied_assignment in group.rs, join_then_sync_then_offset_commit_then_fetch_roundtrip in manager.rs, crates/kaas-broker/tests/group_dispatch.rs, scripts/kafka-verifiable-consumer.sh.

DescribeGroups

Snapshots named groups: state, protocol, members, assignments. Key 15.

Versions: v0–v5 (flexible from v5)

Handling: Manager::describe_groups answers per group and filters by ownership — a group coordinated elsewhere gets a per-group NOT_COORDINATOR (16) entry, which matters because the Java AdminClient unions results across brokers (without the filter, one broker's stale in-memory entry reappeared cluster-wide). Owned groups return their live snapshot; group-state strings match Apache's exactly (Empty, PreparingRebalance, CompletingRebalance, Stable, Dead).

Deviations from Apache 3.7:

• A group this broker coordinates but has no state for is described as Empty with no members; Apache describes an unknown group as Dead.
• Per-member client_host and member_metadata are returned empty (the coordinator tracks the host but the snapshot doesn't carry it yet); member_assignment is populated.
• authorized_operations (v3+) is always 0 — kaas neither computes the operations bitmap nor returns Apache's INT32_MIN "not requested" sentinel.

Source: crates/kaas-broker/src/handlers/describe_groups.rs, crates/kaas-codec/src/api/describe_groups.rs, crates/kaas-coordinator/src/manager.rs.

Verified by: scripts/kafka-consumer-groups.sh (--describe scenario). There is no dedicated unit test for the DescribeGroups handler; the underlying Group::describe snapshot is exercised by the lifecycle tests in crates/kaas-coordinator/src/group.rs.

ListGroups

Enumerates the groups a broker coordinates. Key 16.

Versions: v0–v4 (flexible from v3)

Handling: snapshots every in-memory group and filters by coordinator ownership — the same ownership filter as DescribeGroups, so the AdminClient's cross-broker union never shows a group twice or shows stale orphans. The v4+ states_filter is applied broker-side by exact state-string match. With no coordinator manager installed (boot window) the response is an empty list with error_code = 0, mirroring an idle Apache broker.

Deviations from Apache 3.7:

• Only in-memory groups are listed. A group that exists solely as a committed-offsets file — no live members since the coordinator restarted or the group moved — doesn't appear in --list until a member joins again. Apache materialises such groups from __consumer_offsets and lists them as Empty.

Source: crates/kaas-broker/src/handlers/list_groups.rs, crates/kaas-codec/src/api/list_groups.rs, crates/kaas-coordinator/src/manager.rs, crates/kaas-broker/src/group_takeover.rs (the orphan sweep that keeps the list honest).

Verified by: crates/kaas-broker/tests/group_dispatch.rs, scripts/kafka-consumer-groups.sh (--list before and after --delete — the group must appear and then actually vanish).

DeleteGroups

Drops a group's coordinator state and committed offsets. Key 42.

Versions: v0–v2 (flexible from v2)

Handling: per group, in order: not this broker's group → NOT_COORDINATOR (16); no in-memory state and no offsets file → GROUP_ID_NOT_FOUND (69); live state that isn't Empty/Dead → NON_EMPTY_GROUP (67); otherwise the in-memory group is shut down and the __consumer_offsets/<groupID>.json file deleted. A disk-delete failure after the in-memory wipe is swallowed — the stale file is harmless and the operator's startup sweep re-cleans it. A group whose only trace is the offsets file (all members long gone) counts as existing and is deletable, matching Apache.

Deviations from Apache 3.7: None known.

Source: crates/kaas-broker/src/handlers/delete_groups.rs, crates/kaas-codec/src/api/delete_groups.rs, crates/kaas-coordinator/src/manager.rs.

Verified by: delete_group_non_empty_when_state_is_stable / delete_group_unknown_returns_group_id_not_found in manager.rs, scripts/kafka-consumer-groups.sh scenario 5 (--delete must succeed and the group must vanish from a subsequent --list).

OffsetDelete

Drops specific (topic, partition) committed offsets without deleting the group. Key 47. Drives kafka-consumer-groups.sh --delete-offsets and AdminClient.deleteConsumerGroupOffsets().

Versions: v0 only (not flexible)

Handling: the handler builds the canonical topic/partition key list and calls Manager::delete_offsets, which removes the entries from the group's cache and rewrites the offsets file. Per-partition results: removed → 0; no committed entry under that key → UNKNOWN_TOPIC_OR_PARTITION (3). Per-partition errors are suppressed (0) whenever the group-level error is non-zero. Wire-shape quirk faithfully reproduced: the group-level error_code precedes throttle_time_ms — the opposite field order from DeleteGroups.

Deviations from Apache 3.7:

• The only group-level errors kaas produces are NOT_COORDINATOR (16) and success. An unknown group returns group-level 0 with every partition marked UNKNOWN_TOPIC_OR_PARTITION, where Apache returns GROUP_ID_NOT_FOUND (69).
• No subscription guard: Apache refuses to delete offsets for topics a Stable consumer-protocol group is actively subscribed to (GROUP_SUBSCRIBED_TO_TOPIC, 86). kaas deletes them regardless of group state.

Source: crates/kaas-broker/src/handlers/offset_delete.rs, crates/kaas-codec/src/api/offset_delete.rs, crates/kaas-coordinator/src/offset_store.rs.

Verified by: delete_partitions_removes_only_requested_keys in offset_store.rs; the shell-tool path rides scripts/kafka-consumer-groups.sh.

Transaction APIs

Per-API reference — see the API support matrix for the generated version table.

These six keys are the KIP-98 transactional-producer surface plus the KIP-447 (EOS v2) offset-commit path. The machinery behind them — the slot-file state store that replaces Apache's __transaction_state topic, the shared-volume marker queue that replaces coordinator-to-leader RPCs, and the timeout reaper — is described in Transactions & idempotence (including the state diagram); this page sticks to the wire contracts.

Four facts are shared by everything below:

• Routing. The transaction coordinator for a transactional.id is a pure function: hash(transactional.id) into the sorted full broker set, with a deterministic fallback into the alive subset when the preferred broker is down (pick_txn_coordinator, crates/kaas-broker/src/group_hash.rs). Clients resolve it via FindCoordinator with key_type = 1 (crates/kaas-coordinator/src/manager.rs); the coordinator-side handlers (keys 22, 24, 25, 26) re-check ownership and answer NOT_COORDINATOR (16) from any other broker. Dev mode owns every id.
• State. Per-transactional.id state lives in /data/__cluster/txn_state/slot-N.json, sharded fnv1a32(transactional.id) % 50 — Apache's transaction.state.log.num.partitions=50 default (crates/kaas-coordinator/src/txn_state.rs). Every mutation re-reads the slot file and writes back atomically (tmp + fsync + rename), so coordinator failover is just the new owner reading the same file — no log replay. EndTxn is two-phase, as in Apache: prepare_end_txn moves Ongoing → Prepare{Commit,Abort} retaining the partition and group lists (the durable record of which markers still owe a write), and complete_end_txn clears them only once every marker is durable.
• Errors. The store's failures map identically in every coordinator-side handler: unknown id or PID mismatch → INVALID_PRODUCER_ID_MAPPING (49); epoch mismatch → PRODUCER_FENCED (90, the txn-coordinator convention — the Produce path keeps INVALID_PRODUCER_EPOCH 47); transition already in flight → CONCURRENT_TRANSACTIONS (51); store not yet wired at boot → COORDINATOR_NOT_AVAILABLE (15, retryable); empty transactional.id → INVALID_REQUEST (42). One honest wart: invalid transitions are answered with wire code 50, but Apache's INVALID_TXN_STATE is 48 (50 is INVALID_TRANSACTION_TIMEOUT), so a Java client raises InvalidTxnTimeoutException where Apache raises InvalidTxnStateException. Off by label, not by behaviour.
• Reaper. A per-broker task fires every 10 s (Apache's transaction.abort.timed.out.transaction.cleanup.interval.ms default) and transitions Ongoing entries past ongoingSinceMs + transactionTimeoutMs to PrepareAbort with an epoch bump, keeping the dispatch set — a timed-out transaction owes ABORT markers like any other. A marker-reconcile pass on the same tick places the outstanding markers and completes the transaction, which is when staged offsets are discarded (bins/kaas/src/cluster.rs, crates/kaas-broker/src/txn_markers.rs). Both sweeps are ownership-gated on the txn-coordinator hash — each broker walks only its own slots; dev mode (no coordinator) owns everything.

One cross-cutting note up front: this surface is ACL-gated the way Apache 3.7 gates it. Every transactional handler checks WRITE on the TransactionalId resource (denial → TRANSACTIONAL_ID_AUTHORIZATION_FAILED, 53) before coordinator routing is revealed; the offset-adjacent handlers additionally check READ on the group (30) and — for TxnOffsetCommit — READ per topic (29); WriteTxnMarkers requires CLUSTER_ACTION on the Cluster resource (31). The one deliberate exception is the idempotent InitProducerId path (empty transactional.id), which is not gated — the Java client enables idempotence by default and Apache relaxed the same gate in KIP-679; Produce's per-topic WRITE check is the enforcement point.

InitProducerId

Allocates the (producer id, producer epoch) pair — the entry point for both idempotent and transactional producers. The Java client enables idempotence by default since Kafka 3.0, so every producer sends this at startup.

Versions: v0–v4 (flexible from v2).

Handling — a null or empty transactional.id is the idempotent path: any broker answers locally with a fresh PID from its persisted block allocator (below) and epoch 0, no coordinator gate. A non-empty transactional.id hits the coordinator gate, then the state store: the first call for an id allocates a fresh PID at epoch 0; every reconnect returns the same PID with epoch + 1 — fencing is the monotonic epoch, the KIP-98 contract as amended by KIP-360. Epoch overflow at i16::MAX rotates to a fresh PID at epoch 0. The request's transaction_timeout_ms is recorded on the entry as the reaper's deadline input.

After every epoch > 0 bump the handler fences the old session twice over: an in-process walk advances the PID's epoch and clears its dedupe window on every partition this broker leads, and the bump is appended to this broker's outbound fence file (/data/__cluster/producer_fences/from-<broker>.json) so peer brokers' FenceWatcher applies it within its 2 s poll. During the boot window before the store is wired, the handler degrades gracefully: fresh PID, epoch 0, and a one-shot warning that the rejoin fence is disabled.

Deviations from Apache 3.7:

• The v3+ request fields producer_id / producer_epoch (KIP-360) are decoded but ignored. Apache validates the caller's current epoch and fences stale producers with PRODUCER_FENCED; kaas bumps the epoch for any caller — a zombie that re-calls InitProducerId is handed a new valid epoch instead of an error (each rejoin fences the other session, so mutual fencing still converges, but not Apache's answer).
• A rejoin during an Ongoing transaction does not abort it. Apache aborts the in-flight transaction first (answering CONCURRENT_TRANSACTIONS until done); kaas bumps the epoch and leaves the entry Ongoing for the timeout reaper to sweep.
• transaction_timeout_ms is not validated against a transaction.max.timeout.ms ceiling (Apache rejects oversized values with INVALID_TRANSACTION_TIMEOUT); kaas records whatever is sent.
• PID allocation is cluster-unique but by a different mechanism than Apache's. Apache allocates PID blocks through the quorum; kaas has no quorum, so it partitions the PID space per broker (pid = (broker_id + 1) × 2⁴⁰ + local) and persists each broker's block high-water to the shared volume before any PID in the block is handed out — a crash skips forward to a fresh block, never rewinds. No two brokers, and no two incarnations of one broker, can issue the same PID.

Source: crates/kaas-broker/src/handlers/init_producer_id.rs (handler), crates/kaas-codec/src/api/init_producer_id.rs (codec), crates/kaas-coordinator/src/txn_state.rs (store), crates/kaas-coordinator/src/fence_log.rs + crates/kaas-broker/src/fence_watcher.rs (cross-broker fence), crates/kaas-storage/src/idempotence.rs (dedupe window it seeds).

Verified by: handler unit tests (same-PID/epoch-bump rejoin, fence-log broadcast, empty-string id treated as non-transactional); first_call_allocates_epoch_zero_rejoin_bumps and epoch_overflow_rotates_to_fresh_pid in crates/kaas-coordinator/src/txn_state.rs; bins/kaas/tests/eos_v2.rs; scripts/kafka-txn-coordinator.sh and scripts/kafka-txn-timeout.sh (wire-surface probes — the Kafka 4.x CLI dropped --transactional-id from the verifiable producer, so shell coverage is ApiVersions plus on-PVC state checks).

AddPartitionsToTxn

Declares the partitions a transaction will produce to, before the first transactional batch lands there. The first successful Add* call is what actually starts the transaction.

Versions: v0–v3 (flexible from v3).

Handling — after the shared gates, the store unions the requested (topic, partition) tuples into the entry's partition list. Validation order matches Apache: missing entry → 49, PID mismatch → 49, epoch mismatch → 90, Prepare* in flight → 51. From Empty or a Complete* state the entry transitions to Ongoing and stamps ongoingSinceMs — the timeout reaper's deadline clock. Re-adding already-recorded partitions with no state change is an idempotent no-op (no slot-file rewrite). The v0–v3 response has no top-level error field, so a top-level rejection (wrong coordinator, empty id, boot window) is repeated on every requested partition; the Java client picks any one.

Deviations from Apache 3.7:

• Apache 3.7 additionally serves v4 — the KIP-890 phase-1 batched shape its brokers use for server-side verification. kaas stops at v3, which is the version client producers negotiate; nothing client-visible is missing.
• None known otherwise, beyond the shared warts in the preamble (wire code 50 for invalid transitions is unreachable here — bad states map to 49/90/51).

Source: crates/kaas-broker/src/handlers/add_partitions_to_txn.rs (handler), crates/kaas-codec/src/api/add_partitions_to_txn.rs (codec), crates/kaas-coordinator/src/txn_state.rs (add_partitions).

Verified by: handler unit tests (per-partition error fan-out, happy path); add_partitions_happy_path_then_idempotent, add_partitions_unions_across_calls, epoch_mismatch_fences, and add_partitions_concurrent_transition_rejected in crates/kaas-coordinator/src/txn_state.rs; bins/kaas/tests/eos_v2.rs; scripts/kafka-txn-coordinator.sh (ApiVersions advertisement).

AddOffsetsToTxn

Declares the consumer group whose offsets the transaction will commit — what the Java client sends when sendOffsetsToTransaction() is called, before the TxnOffsetCommit itself.

Versions: v0–v3 (flexible from v3).

Handling — after the shared gates, the store appends group_id to the entry's group list (deduplicated; re-adding is a no-op). Exactly like AddPartitionsToTxn, a call from Empty or Complete* transitions the entry to Ongoing and stamps ongoingSinceMs — either Add* API can open the transaction. The recorded group list is what EndTxn's offset hook later walks to commit or discard the pending offsets staged by TxnOffsetCommit. The response is a single top-level error code. An empty group_id is rejected through the invalid-transition mapping (wire 50 — see the preamble wart).

Deviations from Apache 3.7:

• None known beyond the shared warts in the preamble (wire 50 where Apache uses 48).

Source: crates/kaas-broker/src/handlers/add_offsets_to_txn.rs (handler), crates/kaas-codec/src/api/add_offsets_to_txn.rs (codec), crates/kaas-coordinator/src/txn_state.rs (add_offsets_to_txn).

Verified by: handler unit tests (group recorded on happy path, unknown producer → 49); end_txn_happy_commit_clears_partitions_and_fires_hook in crates/kaas-coordinator/src/txn_state.rs (group list consumed by the hook); bins/kaas/tests/eos_v2.rs; scripts/kafka-txn-coordinator.sh (ApiVersions advertisement).

EndTxn

Commits or aborts the transaction (committed boolean in the request). This is where kaas diverges most visibly from Apache's internals while keeping the client-visible contract.

Versions: v0–v3 (flexible from v3).

Handling — after the shared gates, the flow is prepare → dispatch → complete, Apache's own shape. prepare_end_txn moves Ongoing → Prepare{Commit,Abort}, deliberately retaining the partition and group lists — the durable record of which markers still owe a write. Marker dispatch then splits by partition leader (from assignment.json via the broker Coordinator): self-led partitions get the COMMIT/ABORT control batch built and appended directly with acks = -1, before any queue writes, so a coordinator crash mid-dispatch never loses the local marker; peer-led partitions get one queue file per target broker under /data/__cluster/marker_queue/to-<broker>/<pid>-<epoch>.json — a durably written queue entry is the durability boundary for a peer partition, since the peer's MarkerWatcher (2 s poll) retries until the marker applies. Only once every marker is durable does complete_end_txn run: Prepare* → Complete*, lists cleared, ongoingSinceMs zeroed, and the offset hook fired per recorded group (commit materialises the offsets TxnOffsetCommit staged, abort discards them). Any dispatch failure answers the retriable COORDINATOR_NOT_AVAILABLE (15) and leaves the entry prepared — a producer retry or the 10 s marker reconcile re-derives the identical dispatch set and finishes the job.

A retried EndTxn in the matching Complete* state is answered idempotently (error 0, no second marker); a direction mismatch or EndTxn against Empty returns wire 50 (intended INVALID_TXN_STATE); an epoch mismatch returns PRODUCER_FENCED. The queue file name makes producer retries overwrite rather than pile up.

Deviations from Apache 3.7:

• Peer markers are applied asynchronously. Apache's coordinator drives WriteTxnMarkers RPCs and completes the transaction when every marker is written to the partition log; kaas completes once the queue entries land, so read_committed visibility (LSO advance) on peer-led partitions trails the commitTransaction() return by up to the 2 s poll.
• CoordinatorEpoch in emitted markers is always 0 (kaas tracks no txn coordinator epoch distinct from the assignment epoch); consumers do not act on the field.
• Wire 50 where Apache answers INVALID_TXN_STATE (48) — see preamble.

Source: crates/kaas-broker/src/handlers/end_txn.rs (handler), crates/kaas-codec/src/api/end_txn.rs (codec), crates/kaas-broker/src/control_batch.rs (marker batch), crates/kaas-broker/src/txn_markers.rs (shared dispatch + reconcile), crates/kaas-coordinator/src/txn_state.rs (prepare_end_txn / complete_end_txn), crates/kaas-coordinator/src/marker_queue.rs + crates/kaas-broker/src/marker_watcher.rs (cross-broker dispatch).

Verified by: handler unit tests (commit appends a marker and advances the HWM, idempotent retry writes no second marker, epoch mismatch → 90, Empty → invalid); end_txn_idempotent_retry_returns_ok and end_txn_against_empty_is_invalid in crates/kaas-coordinator/src/txn_state.rs; queue round-trip and overwrite-on-retry tests in crates/kaas-coordinator/src/marker_queue.rs; bins/kaas/tests/eos_v2.rs (commit path visible to read_committed, abort path populates AbortedTransactions[]).

WriteTxnMarkers

Apache's inter-broker API: the transaction coordinator tells each partition leader to write COMMIT/ABORT control batches. kaas serves the receiver side for wire compatibility — but no kaas broker ever sends it; cross-broker markers travel the shared-volume queue instead (see EndTxn above).

Versions: v0–v1 (flexible from v1).

Handling — for each marker in the request the handler builds a control batch from (producer_id, producer_epoch, transaction_result, coordinator_epoch) and, per partition: checks leadership against the assignment (NOT_LEADER_OR_FOLLOWER, 6, if this broker doesn't lead it), runs an idempotent create_partition safety net, and appends with acks = -1. Append failures map to UNKNOWN_SERVER_ERROR (-1) per partition. Dev mode (no Coordinator) treats every partition as self-led. An external coordinator or test harness driving this API gets exactly Apache's receiver behaviour.

Deviations from Apache 3.7:

• The sender side does not exist: kaas coordinators dispatch markers via /data/__cluster/marker_queue/, never via this RPC. Invisible to clients (the API is broker-internal in Apache), but relevant when tracing a cluster on the wire.
• CLUSTER_ACTION on the Cluster resource is required, as in Apache (denial → CLUSTER_AUTHORIZATION_FAILED (31) on every partition). No kaas component holds that ACL — coordinators dispatch via the marker queue — so under simple authorization this API is effectively broker-only unless you grant ClusterAction to a harness principal deliberately.

Source: crates/kaas-broker/src/handlers/write_txn_markers.rs (handler), crates/kaas-codec/src/api/write_txn_markers.rs (codec), crates/kaas-broker/src/control_batch.rs (marker batch).

Verified by: handler unit tests (per-partition append advances the HWM, empty marker list → empty response). No shell script drives it — the Apache CLI tools never send this API, and kaas brokers don't either; the queue path it replaces is exercised end to end by bins/kaas/tests/eos_v2.rs and the marker-queue tests.

TxnOffsetCommit

The transactional counterpart of OffsetCommit: stages the consume-side offsets of a consume-process-produce cycle so they become visible atomically with the transaction — the KIP-447 (EOS v2) contract.

Versions: v0–v3 (flexible from v3).

Handling — this handler runs on the group coordinator (hash(group.id)), not the transaction coordinator; a broker that doesn't coordinate the group answers NOT_COORDINATOR on every partition, as does the boot window before the manager is wired. Offsets are flattened to the same key shape OffsetCommit uses and staged in the offset store's pending layer keyed by (group_id, producer_id) — invisible to OffsetFetch until EndTxn(commit) fires commit_pending; abort (or a reaper sweep) fires discard_pending. The pending layer is memory-only by design: staged offsets of an unfinished transaction dying with the broker is abort-equivalent, which is the correct outcome.

Deviations from Apache 3.7:

• producer_epoch, generation_id, and member_id are decoded but not validated. Apache fences zombies at this API with INVALID_PRODUCER_EPOCH / ILLEGAL_GENERATION / UNKNOWN_MEMBER_ID; kaas keys staging purely on (group_id, producer_id).
• Cross-broker gap: the EndTxn offset hook fires on the txn coordinator's local offset store. When hash(transactional.id) and hash(group.id) resolve to different brokers, the pending entry staged here is never materialised — the group replays from its last committed offset, which breaks exactly-once (duplicates, not loss) for that group. Single-broker deployments and hash-coinciding cases are complete; cross-broker completion is an open follow-up (tracked as gh #114).

Source: crates/kaas-broker/src/handlers/txn_offset_commit.rs (handler), crates/kaas-codec/src/api/txn_offset_commit.rs (codec), crates/kaas-coordinator/src/offset_store.rs (pending layer), crates/kaas-coordinator/src/txn_state.rs (TxnOffsetHook seam).

Verified by: handler unit tests (pending staged and invisible until commit, NOT_COORDINATOR without a manager); pending_invisible_to_fetch_until_commit_pending and discard_pending_drops_unmaterialised_offsets in crates/kaas-coordinator/src/offset_store.rs; bins/kaas/tests/eos_v2.rs (staged offsets across commit and abort); scripts/kafka-txn-coordinator.sh (ApiVersions advertisement).

Topic & config admin APIs

Per-API reference — see the API support matrix for the generated version table.

The whole admin surface on this page is CR-mediated: kaas never mutates topic state directly off a wire request. Writes become creates/patches/deletes of KafkaTopic custom resources, the operator reconciles the CR into on-disk state, and the broker observes the result through its topic watcher — see Kubernetes integration. In dev mode (MY_POD_NAME unset, no kube client) the CR writer is a stub that refuses every write, so the mutating APIs answer CLUSTER_AUTHORIZATION_FAILED (31) with the message broker is not running in cluster mode.

CreateTopics

Creates topics — the broker side of kafka-topics.sh --create and AdminClient.createTopics().

Versions: v0–v7 (flexible from v5).

Handling: per requested topic, the handler authorizes Create on the topic resource, then POSTs a fresh KafkaTopic CR. The operator reconciles it into partition directories on the shared volume; the broker picks the topic up via its topic watcher and serves it on subsequent requests — creation is therefore asynchronous (a success response means the CR was accepted, not that partition dirs exist yet). A non-positive num_partitions (the AdminClient's "server default" convention) maps to 1, mirroring Apache's num.partitions=1 default; the same rule applies to replication_factor. Kafka topic names that aren't valid RFC 1123 subdomains (Kafka Streams internals, dotted names) get a deterministic synthetic CR name kaas-topic-<16 hex> with the literal name stashed in spec.topicName. Config overrides on the request (--config retention.ms=600000) are validated against the supported key set and land in the minted CR's spec.config, so the operator materialises them on first reconcile exactly as if they had been authored on the CR; an unknown key or an unparseable value fails that topic's creation with INVALID_CONFIG (40), as in Apache. validate_only (v1+) runs the authorization, writer, and config-validation checks, then returns the would-be response without minting the CR. Error mapping: authorization denial → TOPIC_AUTHORIZATION_FAILED (29), bad config → INVALID_CONFIG (40), existing CR → TOPIC_ALREADY_EXISTS (36), missing writer or Kubernetes RBAC denial → CLUSTER_AUTHORIZATION_FAILED (31), other kube errors → UNKNOWN_SERVER_ERROR (-1). On ArgoCD-managed clusters, admin.argocd.enabled on the Helm chart makes the minted CR carry ArgoCD tracking/coexistence annotations so runtime-created topics appear in the Application tree without being prune targets — see Kubernetes integration.

Deviations from Apache 3.7:

• The supported config-key set is the eight tunable keys DescribeConfigs reports — an override outside it is rejected with INVALID_CONFIG where Apache would accept any of its several dozen topic keys. One accept-only exception: message.timestamp.type=CreateTime validates and is dropped (it names the only behaviour kaas has; Kafka Streams stamps it on every internal topic it creates), while LogAppendTime is rejected.
• The v7+ response topic_id (KIP-516) is always the all-zero UUID: the real TopicID is minted by the operator on first reconcile, after the response has gone out.
• replication_factor is accepted and echoed but has no effect — kaas is single-writer-per-partition by design (see Non-goals).
• validate_only does not check for an existing topic; it reports success even when a real create would return TOPIC_ALREADY_EXISTS.

Source: crates/kaas-broker/src/handlers/create_topics.rs, crates/kaas-broker/src/topic_cr_writer.rs, crates/kaas-codec/src/api/create_topics.rs.

Verified by: scripts/kafka-topics.sh (create/list/describe scenarios); codec round-trip tests in crates/kaas-codec/src/api/create_topics.rs (including v7_carries_topic_id); CR-name mapping and config-conversion tests in crates/kaas-broker/src/topic_cr_writer.rs; config-threading and rejection handler tests in crates/kaas-broker/src/handlers/create_topics.rs.

DeleteTopics

Deletes topics by name — kafka-topics.sh --delete, AdminClient.deleteTopics().

Versions: v0–v5 (flexible from v4).

Handling: per topic, the handler deletes the KafkaTopic CR, then drops the topic from the in-memory registry. The operator's reconcile tears down the partition directories; before that lands, every broker's topic watch sees the Kubernetes delete event, drops the topic from its registry, abandons the open partitions (closing log/index file handles without persisting state — the topic is gone, not handed over), and purges the topic's committed consumer-group offsets, as Apache tombstones them out of __consumer_offsets. Closing the handles first matters because NFS silly-renames open files and the operator's directory delete wedges (see File-handle ownership). A missing CR answers UNKNOWN_TOPIC_OR_PARTITION (3); other writer errors are reported as INVALID_REQUEST (42) with a message. In dev mode only the registry removal runs — on-disk (in-memory-engine) data is left alone.

Deviations from Apache 3.7:

• Authorization: Delete on the topic per entry, as in Apache — denial answers TOPIC_AUTHORIZATION_FAILED (29) and skips the CR delete.
• Deletion is asynchronous: the wire response confirms the CR delete, while directory teardown follows on the operator's reconcile.

Source: crates/kaas-broker/src/handlers/delete_topics.rs, crates/kaas-broker/src/topic_cr_writer.rs.

Verified by: scripts/kafka-topics.sh (scenario 5, delete-and-confirm).

DeleteRecords

Advances a partition's log start offset (KIP-107) — kafka-delete-records.sh, Kafbat-UI's "purge messages".

Versions: v0–v2 (flexible from v2).

Handling: this is a storage-path API, not a CR write. Per partition the handler applies the same ownership gate Produce uses — with a cluster coordinator wired, partitions this broker doesn't lead answer NOT_LEADER_OR_FOLLOWER (6). The storage engine then advances logStart to the target offset (-1 = purge to the high watermark; a target past the HWM is OFFSET_OUT_OF_RANGE (1)) and returns the new low watermark. Records below logStart become invisible to Fetch immediately, and closed segments that fall entirely below it are unlinked from disk on the spot — safe on NFS because only the leader holds open handles.

Deviations from Apache 3.7:

• The active segment is not rolled or reclaimed by DeleteRecords, and a closed segment only partially covered by the purge is kept whole. Visibility moves immediately; the covering bytes are reclaimed later by segment roll and retention. Apache behaves similarly for partial segments but kaas holds the active segment even when the purge covers the entire log.
• Authorization: Delete on the topic, as in Apache — denial answers TOPIC_AUTHORIZATION_FAILED (29) per partition and nothing is purged.

Source: crates/kaas-broker/src/handlers/delete_records.rs, crates/kaas-storage/src/partition.rs (delete_records), crates/kaas-storage/src/disk.rs.

Verified by: scripts/kafka-delete-records.sh (produce 10, purge to 7, assert earliest = 7); delete_records_* unit tests in crates/kaas-storage/src/partition.rs and crates/kaas-storage/src/memory.rs.

DescribeConfigs

Reads topic and broker configuration — kafka-configs.sh --describe and every admin UI's config pane.

Versions: v0–v4 (flexible from v4).

Handling: two resource types are served. TOPIC: authorize DescribeConfigs on the topic (denial → 29), require the topic in the registry (miss → UNKNOWN_TOPIC_OR_PARTITION (3)), then answer an Apache-3.7-compatible defaults table of the nine config keys kaas actually honours — retention.ms, retention.bytes, segment.bytes, segment.ms, cleanup.policy, min.compaction.lag.ms, delete.retention.ms, flush.messages, and the fixed-value message.timestamp.type (always CreateTime) — with the topic's stored overrides layered on top. (flush.messages advertises a null default: its effective default is the broker-wide flush interval, which the static table can't know — a fixed number here would be the advertised-vs-enforced drift this page keeps warning about.) An overridden key reports the override as its value with source DYNAMIC_TOPIC_CONFIG, so kafka-configs.sh --describe (which shows only non-default entries) and admin UIs distinguish "someone set this" from "this is the default", as in Apache. Overrides are re-read from the operator-materialised per-topic config file on every request, so a change is visible as soon as the operator has reconciled it — no broker restart. v1+ attaches the synonym chain per entry (the dynamic override first when present, then the DEFAULT_CONFIG it shadows), v3+ adds one-line documentation strings, and the request's configuration_keys filter is honoured. BROKER: answers a small fixed read-only table (broker.id plus static defaults) so kafka-configs.sh --entity-type brokers and Kafbat-UI's broker page work. Everything else (BROKER_LOGGER included) gets a per-resource UNSUPPORTED_VERSION (35).

Deviations from Apache 3.7:

• Only nine topic keys are reported, versus Apache's several dozen; tools that iterate the full key set see a short list.
• In dev mode (in-memory storage engine) there is no per-topic config file, so every key reports its default.
• The broker table reports static kafka.version = 3.6.0 / inter.broker.protocol.version = 3.6 strings (predating the 3.7 parity target).
• BROKER_LOGGER is unsupported and answers UNSUPPORTED_VERSION (35), where Apache serves log4j levels.

Source: crates/kaas-broker/src/handlers/describe_configs.rs, crates/kaas-broker/src/topic_config_defaults.rs.

Verified by: scripts/kafka-configs.sh (broker describe, topic describe, --describe --all, per-broker-id describe); override-layering handler tests in crates/kaas-broker/src/handlers/describe_configs.rs.

CreatePartitions

Grows a topic's partition count (KIP-195) — kafka-topics.sh --alter --partitions N.

Versions: v0–v3 (flexible from v2).

Handling: authorize Alter on the topic (denial → 29), then merge-patch KafkaTopic.spec.partitions to the new count. The writer reads the CR first and refuses a decrease client-side with INVALID_PARTITIONS (37) — the operator's reconciler enforces the same guard as backstop. A missing CR is UNKNOWN_TOPIC_OR_PARTITION (3); dev mode / RBAC denial is CLUSTER_AUTHORIZATION_FAILED (31). The operator creates the new partition directories on reconcile and the broker serves them after its watcher fires — expansion is asynchronous, same as topic creation. validate_only (v1+) short-circuits before the patch.

Deviations from Apache 3.7:

• A request for the same partition count succeeds as a no-op; Apache returns INVALID_PARTITIONS when the requested count doesn't exceed the current one. Only a strict decrease is refused.
• The request's manual assignments (replica placement per new partition) are ignored — there are no replicas to place (see Non-goals); partition-to-broker placement is the controller's job.

Source: crates/kaas-broker/src/handlers/create_partitions.rs, crates/kaas-broker/src/topic_cr_writer.rs (expand_topic).

Verified by: scripts/kafka-topics.sh (scenario 4, alter-and-describe); writer unit tests in crates/kaas-broker/src/topic_cr_writer.rs.

IncrementalAlterConfigs

Per-key topic config mutation (KIP-339) — kafka-configs.sh --alter --add-config / --delete-config.

Versions: v0–v1 (flexible from v1).

Handling: TOPIC resources only. The handler authorizes AlterConfigs on the topic, translates the op list, and issues a single JSON-merge patch on KafkaTopic.spec.config: SET writes the parsed value (integer keys become JSON numbers), DELETE — and SET with a null value — write JSON null. The patchable key set is the eight tunable keys DescribeConfigs reports, accepted in dotted or camelCase form; a key outside it, or a value that doesn't parse for its key, is rejected with INVALID_CONFIG (40) before anything reaches the Kubernetes API server. The operator materialises the change on reconcile, the storage engine's cleaner picks it up, and a subsequent DescribeConfigs reports the override as DYNAMIC_TOPIC_CONFIG. validate_only runs the same validation and skips the patch. BROKER and BROKER_LOGGER resource types answer a per-resource UNSUPPORTED_VERSION (35) — there is no dynamic broker-config surface.

Deviations from Apache 3.7:

• APPEND and SUBTRACT are unsupported and answer UNSUPPORTED_VERSION (35): every kaas topic-config key is scalar, so the list-valued ops have nothing to apply to.
• Config keys outside the allow-list answer INVALID_CONFIG (40) as Apache does for unknown names — but the allow-list itself is far smaller than Apache's key set, so keys Apache would accept (max.message.bytes, ...) are rejected here.
• BROKER / BROKER_LOGGER alteration is unsupported (Apache 3.7 supports dynamic broker configs, KIP-226).
• One bad op fails the whole resource — the ops for a resource are applied as a single all-or-nothing merge patch.
• The change is asynchronous: it is visible to DescribeConfigs once the operator has reconciled the CR (typically well under a second), not atomically with the alter response.

Source: crates/kaas-broker/src/handlers/incremental_alter_configs.rs, crates/kaas-broker/src/topic_cr_writer.rs (update_topic_config, config_key_to_json_field, config_value_to_json).

Verified by: scripts/kafka-configs.sh (scenario 3); key/value-mapping unit tests in crates/kaas-broker/src/topic_cr_writer.rs; rejection handler tests in crates/kaas-broker/src/handlers/incremental_alter_configs.rs.

ACL & quota admin APIs

Per-API reference — see the API support matrix for the generated version table.

kaas has no ACL store of its own: ACLs live inline on each principal's KafkaUser CR (spec.authorization.acls — see Kubernetes integration). The three ACL admin APIs translate the AdminClient's int8-enum wire shape into that CR shape and delegate to the ACL CR writer; the operator's reconcile then rebuilds /data/__cluster/acls.json and every broker's ACL engine hot-reloads it. Runtime edits to git-managed KafkaUser CRs will show up as ArgoCD drift until the next sync — the intentional trade for letting the admin protocol reach the canonical store. Without a writer wired (dev mode), DescribeAcls returns an empty set and CreateAcls/DeleteAcls report per-entry success without persisting anything.

One cross-cutting note: the trio is gated the way Apache gates it — DescribeAcls requires Describe on the Cluster resource, and CreateAcls/DeleteAcls require Alter on it; denial answers CLUSTER_AUTHORIZATION_FAILED (31) before the CR writer is consulted, so the answer is the same with or without an apiserver wired. The host field of a binding is stored and round-tripped verbatim but ignored by ACL evaluation.

DescribeAcls

Lists ACL bindings matching a filter — kafka-acls.sh --list.

Versions: v0–v3 (flexible from v2).

Handling: the wire filter's ANY/UNKNOWN codes and null strings collapse to wildcards; a MATCH pattern filter expands to literal + prefixed per KIP-290; v0 (pre-KIP-290) pins the pattern filter to literal so prefixed entries are never returned to a v0 client. The writer lists every KafkaUser CR (skipping ones mid-deletion), expands each inline ACL entry into one binding per operation, applies the filter, and the handler folds the flat list back into Apache's per-resource shape — one resource row per (type, name, pattern) with the matching ACLs inside. Filter errors (resource types kaas can't express) answer INVALID_REQUEST (42); apiserver failures answer UNKNOWN_SERVER_ERROR (-1).

Deviations from Apache 3.7:

• Resource types are limited to topic, group, cluster, and transactional-ID — DELEGATION_TOKEN and USER filters answer INVALID_REQUEST (42) (delegation tokens are a non-goal).
• Dev mode answers an empty list rather than an error.

Source: crates/kaas-broker/src/handlers/acls.rs, crates/kaas-broker/src/acl_cr_writer.rs, crates/kaas-codec/src/api/acl_types.rs, crates/kaas-codec/src/api/describe_acls.rs.

Verified by: scripts/kafka-acls.sh (list/add/list/remove round trip against a temporary KafkaUser); enum-translation and grouping unit tests in crates/kaas-broker/src/handlers/acls.rs; filter-matching unit tests in crates/kaas-broker/src/acl_cr_writer.rs.

CreateAcls

Adds ACL bindings — kafka-acls.sh --add.

Versions: v0–v3 (flexible from v2).

Handling: per binding, the wire enums are validated (ANY/UNKNOWN codes, and resource types kaas can't express, answer INVALID_REQUEST (42)); v0 bindings get literal pattern semantics. The principal must be of the form User:<name> and a KafkaUser CR with that name must already exist — kaas never auto-creates CRs from a runtime ACL write; both failures answer INVALID_REQUEST (42). Creation is idempotent and coalescing: an existing entry with the same resource, pattern, permission, and host absorbs the new operation into its operations list (or no-ops when already present). The write is a single Update with the read resourceVersion; a concurrent-edit conflict surfaces as UNKNOWN_SERVER_ERROR (-1) and the AdminClient retries.

Deviations from Apache 3.7:

• Principals other than User: (e.g. Group:) are rejected — kaas maps principals 1:1 onto KafkaUser CRs.
• An ACL for a principal with no KafkaUser CR is refused (INVALID_REQUEST with no KafkaUser CR for principal ...); Apache accepts ACLs for arbitrary principal strings. Create the KafkaUser first.
• Dev mode reports success without persisting.

Source: crates/kaas-broker/src/handlers/acls.rs, crates/kaas-broker/src/acl_cr_writer.rs (create_acl), crates/kaas-codec/src/api/create_acls.rs.

Verified by: scripts/kafka-acls.sh; unit tests in crates/kaas-broker/src/handlers/acls.rs and crates/kaas-broker/src/acl_cr_writer.rs (principal parsing, enum mapping); end-to-end ACL enforcement in bins/kaas/tests/auth_smoke.rs (acl_denies_unconfigured_topic).

DeleteAcls

Removes ACL bindings matching filters — kafka-acls.sh --remove.

Versions: v0–v3 (flexible from v2).

Handling: same filter translation as DescribeAcls (KIP-290 MATCH expansion, v0 literal pinning). The writer walks every KafkaUser CR, partitions each inline entry's operations into matched vs kept, rewrites the CR when anything matched, and returns the flat list of removed bindings — one per (entry, operation) pair — which the handler echoes as the per-filter matching_acls. Entries whose operations are only partially matched are kept with the remaining operations; entries emptied out are dropped. CRs mid-deletion are skipped.

Deviations from Apache 3.7:

• Dev mode reports success with zero matches, without touching anything.

Source: crates/kaas-broker/src/handlers/acls.rs, crates/kaas-broker/src/acl_cr_writer.rs (delete_acls), crates/kaas-codec/src/api/delete_acls.rs.

Verified by: scripts/kafka-acls.sh (remove-and-verify scenario); filter-partition unit tests in crates/kaas-broker/src/acl_cr_writer.rs.

DescribeClientQuotas

Reads client quota entries (KIP-546) — kafka-configs.sh --entity-type users --describe.

Versions: v0–v1 (flexible from v1).

Handling: authorizes DescribeConfigs on the Cluster resource (Apache's mapping for quota describe; denial → CLUSTER_AUTHORIZATION_FAILED (31)). kaas supports a single entity axis: user. An exact-match component describes that user; ANY (or an empty component list) lists every user with a quota. Values resolve runtime override first, CR-backed store second: overrides installed by AlterClientQuotas shadow the quotas the operator materialised into /data/__cluster/credentials.json from KafkaUser.spec.quotas. Reported keys: producer_byte_rate, consumer_byte_rate, request_percentage. With no quota enforcer wired (auth disabled), the response is an empty success — indistinguishable on the wire from "no quotas configured", mirroring Apache.

Deviations from Apache 3.7:

• Only the user entity axis exists. client-id / ip components, and the DEFAULT match type (<default> user entity), return an empty result rather than an error — kaas users are CR-instantiated, so there is no default entity.
• Quota values are per-broker (KIP-13) — same semantics as Apache, but worth restating: with N brokers the cluster-wide ceiling is N × the reported value. The CR field names (producerMaxByteRatePerBroker) say so explicitly; the wire keys keep Apache's names.

Source: crates/kaas-broker/src/handlers/describe_client_quotas.rs, crates/kaas-auth/src/quota.rs (describe_user_quota, list_user_quotas).

Verified by: scripts/kafka-configs.sh (quota scenarios 6–9); resolution- order unit tests in crates/kaas-auth/src/quota.rs (describe_user_quota_resolution_order).

AlterClientQuotas

Sets or removes client quota values (KIP-546) — kafka-configs.sh --entity-type users --alter.

Versions: v0–v1 (flexible from v1).

Handling: authorizes AlterConfigs on the Cluster resource once for the whole request (denial → per-entry CLUSTER_AUTHORIZATION_FAILED (31)). Each entry must name exactly one user entity with an explicit name — anything else answers INVALID_REQUEST (42). Ops merge onto the user's current effective quotas with Apache semantics: a set replaces just the named key, a remove drops just that key, unspecified keys are preserved. Supported keys are producer_byte_rate, consumer_byte_rate, and request_percentage; an unknown key answers INVALID_CONFIG (40). The merged result is installed as a runtime override on the quota enforcer, live-updating any active token bucket; a merge that empties every field clears the override, reverting the user to the CR-backed value. With no enforcer wired (auth disabled) each entry answers UNSUPPORTED_VERSION (35). validate_only skips the install.

Deviations from Apache 3.7:

• Alterations are not persisted. The override lives in the enforcer's memory: it does not write back to the KafkaUser CR, it is lost on broker restart, and it applies only on the broker that served the request — peers keep the store-backed value. Durable, cluster-wide quotas belong on KafkaUser.spec.quotas (see Kubernetes integration). Treat this API as a live-tuning knob, not a store.
• request_percentage is accepted, stored, and reported, but nothing enforces it — kaas throttles produce/fetch byte rates only, with no request-time CPU quota.
• Entity axes other than a single named user are rejected (INVALID_REQUEST), including the <default> entity.

Source: crates/kaas-broker/src/handlers/alter_client_quotas.rs, crates/kaas-auth/src/quota.rs (set_user_quota).

Verified by: scripts/kafka-configs.sh (alter/describe/clear round trip); set_user_quota_live_updates_existing_bucket and the debt-carry contention test in crates/kaas-auth/src/quota.rs; enforcement end-to-end in bins/kaas/tests/auth_smoke.rs (produce_exceeds_quota_returns_throttle).

DescribeUserScramCredentials

Lists which SCRAM mechanisms a user has credentials for — kafka-configs.sh --describe --entity-type users and the AdminClient's describeUserScramCredentials() (KIP-554).

Versions: v0 (flexible from v0).

Handling: authorize Describe on the cluster (denial → top-level CLUSTER_AUTHORIZATION_FAILED (31)), then answer from the live credential store — the operator-materialised credentials.json, hot-reloaded, so the response reflects the store the SCRAM authenticator actually verifies against. A null users array describes every user with SCRAM credentials; a named user without any answers a per-user RESOURCE_NOT_FOUND (83); a user named twice answers DUPLICATE_RESOURCE (81) — all Apache's shapes. Only mechanism + iteration count are reported, never salts or keys.

Deviations from Apache 3.7:

• Every credential reports mechanism SCRAM-SHA-512 — kaas serves no SCRAM-SHA-256, so a user never has more than one credential entry.

Source: crates/kaas-broker/src/handlers/describe_user_scram_credentials.rs, crates/kaas-codec/src/api/describe_user_scram_credentials.rs, crates/kaas-auth/src/credentials.rs (list_all_scram_users).

Verified by: codec round-trip tests (null-vs-empty users pinned) in crates/kaas-codec/src/api/describe_user_scram_credentials.rs; scripts/kafka-configs.sh (user describe scenarios).

AlterUserScramCredentials

Rotates a user's SCRAM credential over the wire — kafka-configs.sh --alter --entity-type users --add-config 'SCRAM-SHA-512=...' (KIP-554).

Versions: v0 (flexible from v0).

Handling: authorize Alter on the cluster (denial → per-user CLUSTER_AUTHORIZATION_FAILED (31)). The wire carries pre-salted material — (salt, saltedPassword, iterations), never the password — and the broker derives the RFC 5802 stored/server keys and patches them into KafkaUser.spec.authentication.scram. The operator materialises the change into credentials.json on reconcile and every broker hot-reloads it, so the rotation is asynchronous (typically a few seconds) and cluster-wide. The CR must already exist (RESOURCE_NOT_FOUND (83) otherwise) and its authentication.type must be scram-sha-512 or unset — rotating a tls user's SCRAM credential would silently flip its auth mechanism, so that answers INVALID_REQUEST (42). Iterations below 4096 or empty salt/saltedPassword answer UNACCEPTABLE_CREDENTIAL (93).

Deviations from Apache 3.7:

• SCRAM-SHA-512 only — SHA-256 upsertions answer UNSUPPORTED_SASL_MECHANISM (33).
• Deletions are refused with UNSUPPORTED_VERSION (35): the credential lifecycle belongs to the KafkaUser CR (delete the CR or change its authentication.type). The operator would re-materialise anything the broker removed, and a deletion that silently comes back is worse than a refusal.
• The rotation is visible to SCRAM handshakes only after the operator reconcile plus the brokers' credential reload (~10 s worst case), not atomically with the response.

Source: crates/kaas-broker/src/handlers/alter_user_scram_credentials.rs, crates/kaas-broker/src/user_cr_writer.rs, crates/kaas-codec/src/api/alter_user_scram_credentials.rs, crates/kaas-auth/src/scram.rs (keys_from_salted_password).

Verified by: handler tests (key derivation, mechanism/iteration rejection, deletion refusal) in crates/kaas-broker/src/handlers/alter_user_scram_credentials.rs; codec round-trips in crates/kaas-codec/src/api/alter_user_scram_credentials.rs.

SASL authentication APIs

Per-API reference — see the API support matrix for the generated version table.

Authentication in kaas is per-listener: each listener gets its own auth engine, and the dispatcher's pre-auth gate rejects every API except SaslHandshake (17), ApiVersions (18), and SaslAuthenticate (36) with CLUSTER_AUTHORIZATION_FAILED (31) until the connection's SASL exchange completes — see Listeners, authentication, authorization. mTLS listeners satisfy the same gate at TLS-handshake time instead (the server stamps the connection authenticated from the client certificate, with KIP-371 principal mapping applied), so they never touch these two APIs.

SaslHandshake

Negotiates the SASL mechanism before authentication — the first call every SASL client makes after ApiVersions.

Versions: v0–v1 (not flexible).

Handling: the handler advertises the listener's own mechanism list (each listener's auth engine answers mechanisms()): a SCRAM/PLAIN listener advertises SCRAM-SHA-512, PLAIN in preference order; an oauth listener advertises OAUTHBEARER only, and answers UNSUPPORTED_SASL_MECHANISM (33) to a SCRAM attempt. A supported mechanism is stamped on the connection state so SaslAuthenticate instantiates the right exchange; an unsupported one answers 33 with the list, and nothing is stamped — the client must retry the handshake.

Deviations from Apache 3.7:

• The mechanism menu is SCRAM-SHA-512, PLAIN, and (on oauth listeners) OAUTHBEARER (KIP-255). SCRAM-SHA-256 is not implemented (the credentials pipeline materialises scram-sha-512 entries only), and GSSAPI / delegation-token authentication are absent (see Non-goals).
• v0 is accepted on the wire, but the pre-KIP-152 flow it implies — bare SASL tokens sent without Kafka framing after the handshake — is not implemented. Clients must use SaslAuthenticate; every client from the KIP-152 era (Kafka 1.0+) does.

Source: crates/kaas-broker/src/handlers/sasl.rs, crates/kaas-codec/src/api/sasl_handshake.rs.

Verified by: handler unit tests in crates/kaas-broker/src/handlers/sasl.rs (known/unknown mechanism); bins/kaas/tests/auth_smoke.rs; scripts/kafka-acls.sh and any script run with an authenticated client properties file exercise it against a live broker.

SaslAuthenticate

Carries the SASL exchange itself (KIP-152 framing).

Versions: v0–v2 (flexible from v2).

Handling: on the first call the handler instantiates the per-listener engine's exchange for the handshake-negotiated mechanism (defaulting to SCRAM-SHA-512 if the client skipped the handshake), then steps the state machine with each request's auth_bytes. SCRAM-SHA-512 is a full RFC 5802 server-side implementation (two round trips); PLAIN completes in one; OAUTHBEARER validates the bearer JWT locally against the issuer's JWKS (a failed token gets OAUTHBEARER's two-step failure round trip — error JSON, then the client's terminating response — per RFC 7628). On completion the handler stamps the resolved principal and sasl_done on the connection, which opens the dispatcher's pre-auth gate; the principal then feeds ACL checks and quota buckets. A failed step answers SASL_AUTHENTICATION_FAILED (58) and drops the exchange state, so the client must restart from the handshake. SCRAM/PLAIN credentials come from /data/__cluster/credentials.json, materialised by the operator from KafkaUser CRs and hot-reloaded; OAUTHBEARER principals carry no stored credential — validation is against the JWKS keys fetched from the issuer (fail-closed before the first fetch) — see Kubernetes integration and Listeners, authentication, authorization.

Deviations from Apache 3.7:

• PLAIN and OAUTHBEARER are refused on non-TLS connections with NETWORK_EXCEPTION (13), before the credential bytes are read. Apache allows SASL_PLAINTEXT; kaas deliberately does not ship a path that sends reusable credentials in cleartext.
• Session re-authentication (KIP-368) is partial: session_lifetime_ms is 0 (never expires) for SCRAM/PLAIN, but an oauth listener with maxSecondsWithoutReauthentication set advertises min(configured, token expiry), the dispatcher answers SASL_AUTHENTICATION_FAILED (58) to requests past the deadline, and a re-authentication must resolve the same principal as the session it replaces.
• A client that skips the handshake gets SCRAM-SHA-512 assumed, rather than Apache's handshake-required strictness.

Source: crates/kaas-broker/src/handlers/sasl.rs, crates/kaas-auth/src/scram.rs, crates/kaas-auth/src/plain.rs, crates/kaas-auth/src/oauth.rs, crates/kaas-auth/src/engine.rs, crates/kaas-protocol/src/dispatch.rs (pre-auth gate + re-auth deadline).

Verified by: bins/kaas/tests/auth_smoke.rs (scram_handshake_then_authenticate_unblocks_produce drives the full SCRAM exchange over a real socket and proves the gate opens); bins/kaas/tests/oauth_smoke.rs (TLS pre-auth → handshake → OAUTHBEARER against a live JWKS fixture); PLAIN/TLS/OAUTHBEARER unit tests in crates/kaas-broker/src/handlers/sasl.rs; SCRAM vectors in crates/kaas-auth/src/scram.rs.

Cluster & log-dir APIs

Per-API reference — see the API support matrix for the generated version table.

ApiVersions

The bootstrap call: tells the client which API keys and version ranges this broker serves, so everything else on these pages is discoverable rather than guessed.

Versions: v0–v4 (flexible from v3, KIP-482).

Handling: the response is built directly from the codec's ApiSpec registry — the same table that generates the API support matrix — so the advertised surface is the wire truth by construction: 40 keys, sorted, deduplicated (unit tests assert the count and both invariants). The API is on the pre-auth allowlist, so it works before SASL completes. Two protocol subtleties are implemented faithfully:

• The v0-response-header quirk: the ApiVersions response header is always encoded as header v0 — no tagged-field block — even on flexible request versions. This is Apache's documented exception, kept so a client that misjudged the broker's capabilities can still parse the error code.
• Unknown-version fallback: when a client requests a version outside the supported range, the dispatcher does not return UNSUPPORTED_VERSION (as it does for every other key) — ApiVersions is special-cased so version negotiation can always complete.

Deviations from Apache 3.7:

• The unknown-version fallback clamps to the broker's max version and answers success, where Apache answers error 35 in a v0-encoded body listing its supported range. Only clients newer than v4 can observe the difference.
• The v3+ request fields client_software_name / client_software_version are ignored — the handler doesn't decode the request body at all, so they never reach logs or metrics.
• No SupportedFeatures / FinalizedFeatures tagged fields (KIP-584) in the response. They're optional tagged fields, so clients treat their absence as "no feature versioning" — consistent with kaas having no KRaft feature levels (see Non-goals).

Source: crates/kaas-broker/src/handlers/api_versions.rs, crates/kaas-codec/src/api/api_versions.rs (response_from_registry, the always-v0 response_hdr), crates/kaas-codec/src/api/registry.rs, crates/kaas-protocol/src/dispatch.rs (clamp + pre-auth allowlist).

Verified by: scripts/kafka-broker-api-versions.sh (asserts the required API set is advertised and that every advertised range overlaps the Java client's — any [usable: -1] line fails the run); handler and codec round-trip tests in the source files above; dispatcher clamp tests in crates/kaas-protocol/src/dispatch.rs.

DescribeCluster

What AdminClient.describeCluster() calls: cluster id, controller, and the live broker set — the same three facts Metadata carries, without the per-topic payload. kafka-cluster.sh cluster-id and most UIs' cluster panes land here.

Versions: v0–v2 (flexible from v0 — this API postdates KIP-482, so there is no legacy encoding). v2 is a deliberate exception to the Apache 3.7 parity target: IncludeFencedBrokers / IsFenced are KIP-1073, which is Kafka 4.0 surface. kaas serves it because it has a real fenced state to report — see broker fencing — and because clients that ask for fenced brokers unconditionally cannot otherwise negotiate a version where the field is legal.

Handling: broker rows come from the same catalog Metadata advertises, so both APIs answer with the port of the listener the request arrived on, peers at their stable per-broker DNS name, and self at its own advertised host. A client that reached one API on the authed listener is never handed the anonymous port by the other. The controller is the broker holding the kaas-controller Lease, read from the applied assignment.json.

Fenced brokers: a fenced broker is registered but not serving — its pod exists and its EndpointSlice entry is still there, but it is not Ready. Those rows are omitted unless a v2 request sets IncludeFencedBrokers, matching Apache and matching Metadata, which never advertises them. Asking for them is the only way to tell a degraded cluster from a smaller one. The broker answering the request never reports itself as fenced.

The API itself is not authorization-gated — Apache answers the broker list to any authenticated principal. The optional ClusterAuthorizedOperations bitfield is: it needs Describe on the cluster resource, and then each supported operation is evaluated in turn. Three distinct answers, which clients read differently: -2147483648 when the client didn't ask, 0 when it asked but lacks Describe, and the computed bitfield otherwise.

kaas serves broker endpoints only — controller election runs on a Kubernetes Lease rather than a KRaft quorum (see Non-goals) — so a v1 request for the controller endpoint type gets MISMATCHED_ENDPOINT_TYPE (114) and any other value UNSUPPORTED_ENDPOINT_TYPE (115), mirroring Apache. v0 carries no endpoint-type field, so neither error can reach a v0 client.

Deviations from Apache 3.7:

• The ClusterAuthorizedOperations bitfield never sets the ClusterAction or IdempotentWrite bits. Apache lists both as cluster-scoped operations; kaas models neither — there is no Kafka-RPC inter-broker surface for ClusterAction to guard (peers talk gRPC heartbeats), and idempotent produce is gated by Write on the topic rather than by a cluster-level grant. Reporting a bit the authorizer cannot evaluate would be a guess.
• Rack is always null — kaas has no rack awareness (it has no replicas to place).
• IsFenced is derived from EndpointSlice readiness, not from a KRaft-style broker registration + session timeout. The practical difference: a broker that is booting (registered, not yet finished takeover) reads as fenced, which is the same answer Apache gives for a broker that has registered but not yet caught up.
• ControllerId falls back to this broker when no assignment has been applied yet, where Apache would answer -1. Every kaas broker serves the same admin surface, so a client that dials the answer always reaches something that can serve it.

Source: crates/kaas-broker/src/handlers/describe_cluster.rs, crates/kaas-broker/src/listener_advert.rs (the catalog shared with Metadata), crates/kaas-codec/src/api/describe_cluster.rs.

Verified by: scripts/kafka-cluster.sh (cluster-id round trip plus an AdminClient.describeCluster() call through kafka-broker-api-versions); handler tests in the source file above, covering fenced-row filtering at v1 and v2 in both directions; the DescribeCluster leg of bins/kaas/tests/smoke.rs, which drives the flexible-from-v0 header path through the real dispatcher.

DescribeLogDirs

Reports log directories and per-partition sizes — kafka-log-dirs.sh --describe and Kafbat-UI's storage pane.

Versions: v0–v4 (flexible from v2; matches Apache 3.7's range).

Handling: kaas reports one log directory per pool member — the default data dir plus every storage.pool[] volume (the volume pool; single-volume deployments report exactly one dir, as before). Partitions group under the dir the placement record assigns them to. A null topics filter expands to every topic in the broker's registry with all partitions; a named topic with an empty partition list expands to all its partitions; unknown topics are silently dropped (matching Apache, which omits partitions it doesn't host). Each partition row carries partition_size from the storage engine, offset_lag = 0, and is_future_key = false; the dir-level error_code is always 0. v4 responses carry per-dir TotalBytes / UsableBytes (KIP-827) from a statvfs of the dir's filesystem — -1 when the probe fails (the dev-mode memory:// sentinel).

Deviations from Apache 3.7:

• partition_size is only real for partitions this broker currently leads — the engine sums segment sizes of open partitions and reports 0 for everything else. Apache reports sizes for every replica a broker hosts; kaas has no replicas, so per-partition sizes on a multi-broker cluster are scattered across the brokers that lead them (kafka-log-dirs.sh queries all brokers by default, so the union is complete — with zero-rows for the non-leaders).
• offset_lag is hardwired to 0 and is_future_key to false — coherent for a broker with no followers and no intra-broker reassignment, but a client should not read fetch-lag meaning into it. Source: crates/kaas-broker/src/handlers/describe_log_dirs.rs, crates/kaas-storage/src/disk.rs (partition_size, log_dirs), crates/kaas-codec/src/api/describe_log_dirs.rs.

Verified by: scripts/kafka-log-dirs.sh (all-dirs describe plus topic-filtered describe); partition_size_sums_segment_sizes and placement_resolver_routes_partition_dirs in crates/kaas-storage/src/disk.rs; version roundtrips in the codec module.

AlterReplicaLogDirs

Moves a partition between log dirs on the same broker (KIP-113) — in kaas, the volume-pool migration verb: the drain path for cordoned pool members.

Versions: v0–v2 (flexible from v2; matches Apache 3.7's range).

Handling: the destination is a log dir path as reported by DescribeLogDirs. Per partition, the current leader closes the partition, fresh-copies its directory to the destination volume, flips the placement record (KafkaTopic.status.volumeAssignments), updates its local registry, and reclaims the source directory. Produce/fetch during the copy window fail with the retriable LEADER_NOT_AVAILABLE. Error codes: unknown path → LOG_DIR_NOT_FOUND (57); cordoned member (KIP-1066: no new placements) → INVALID_REQUEST (42); not this broker's partition → REPLICA_NOT_AVAILABLE (9); failed copy or record flip → KAFKA_STORAGE_ERROR (56), with the copy rolled back so data location and placement record never diverge.

Deviations from Apache 3.7:

• Apache moves a replica live (future replica + catch-up, then swap); kaas pauses the partition for the copy — a brief unavailability window instead of a background reassignment, coherent with single-writer-per-partition and no followers.
• Only the partition's current leader accepts the move (it is the only broker holding the files); Apache accepts on any broker hosting a replica.

Source: crates/kaas-broker/src/handlers/alter_replica_log_dirs.rs, crates/kaas-storage/src/disk.rs (move_partition_to_log_dir), crates/kaas-codec/src/api/alter_replica_log_dirs.rs.

Verified by: move_partition_between_log_dirs in crates/kaas-storage/src/disk.rs; all_versions_roundtrip in the codec module; scripts/kafka-log-dirs.sh.

KIP index

All KIPs the codebase references, split honestly: implemented, partial, or deliberate non-goal.

KIP numbers are how Kafka names its own evolution, which makes them the natural vocabulary for the question every evaluator asks: which Kafka do I actually get? This index answers in three buckets, and the split is source-verified, not aspirational: implemented means the behaviour ships and is exercised by tests or the shell-tool suite; partial pages lead with what's missing; non-goals get a rationale in Non-goals, not silence. Wire-facing KIPs also appear per key in the generated API matrix's KIPs column. The scope is Apache Kafka 3.7, with one deliberate post-3.7 exception: DescribeCluster v2 (KIP-1073) is served, because kaas has a real fenced broker state and no other way to report it — see KIP-700.

Three corrections relative to earlier planning documents, all found by source-verifying during the book build: KIP-516, KIP-32, KIP-58, and KIP-354 are partial, not implemented. Topic IDs are minted but not propagated to the wire (KIP-516); LogAppendTime and timestamp lookup don't exist (KIP-32); and the two compaction knobs (KIP-58, KIP-354) remain config plumbing without an enforcing compactor — though the delete-policy retention cleaner now runs (gh #250).

Implemented (14)

Partial (10)

Each page leads with the "what's missing" block — these are the book's credibility test.

Deliberate non-goals (8)

Rationale for each in Non-goals.

Implementation notes (for contributors)

The missing compactor behind the KIP-58 and KIP-354 rows is tracked as gh #158 — which now covers only the compactor: size- and time-based retention are wired and enforced by the RetentionCleaner (gh #250). The corrections in the note above come from the 2026-07-19 source sweep, recorded on the affected KIP pages.

KIP-13 — per-broker client quotas

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-13 (Kafka 0.9) introduced byte-rate quotas for clients: per-principal producer and consumer caps, enforced per broker rather than cluster-wide. A broker that sees a client exceed its rate computes a throttle delay and returns it as throttle_time_ms in the response, so a well-behaved client backs off. With N brokers, the effective cluster-wide ceiling is N × the configured rate — that per-broker semantic is part of the KIP, not an accident.

How kaas implements it

The enforcement engine is a per-principal token bucket in crates/kaas-auth/src/quota.rs (QuotaEnforcer). Rates come from the KafkaUser CR, whose fields are deliberately named producerMaxByteRatePerBroker / consumerMaxByteRatePerBroker (crates/kaas-operator-api/src/kafkauser.rs) to make the KIP-13 per-broker semantics legible at the CR level — same behaviour as Strimzi/Apache, named honestly.

Mechanics, all in quota.rs:

• Refill is continuous at the configured rate, capped at one second's worth of tokens; a rate of 0 means unlimited; a brand-new principal's bucket is seeded full so first contact isn't throttled.
• Debt-carry: the deduction is unconditional and the bucket is allowed to go negative; throttle_time_ms is the time to refill back to zero. The earlier clamp-at-zero version let N concurrent clients sharing a principal each see a "full" bucket and burst at N×rate — the 16-vs-10 MiB/s gap observed under bench-perf.
• Runtime overrides land via AlterClientQuotas (quota admin APIs, KIP-546): set_user_quota live-updates an existing bucket; resolution order is override > credentials store.

The Produce and Fetch handlers call the checker on every request (crates/kaas-broker/src/handlers/produce.rs, fetch.rs) and put the result straight into the response's throttle_time_ms. Quotas fire regardless of whether authorization is enabled — they are orthogonal axes (Listeners, auth, quotas). On brokers with only anonymous listeners a NoQuotaChecker stands in (ANONYMOUS has no quota config to enforce against).

Where kaas does less than Apache: the broker computes and returns throttle_time_ms but never mutes the connection after responding. That response-then-mute ordering belongs to KIP-219, which is partial — enforcement currently relies on the client honouring the throttle hint, which official clients do.

How it's verified

Unit tests in crates/kaas-auth/src/quota.rs: multi_client_contention_carries_debt (pins the debt-carry — back-to-back drains must yield strictly increasing throttle), over_limit_throttles, zero_rate_means_unlimited, per_principal_isolation, refill_clears_debt_over_time, and set_user_quota_live_updates_existing_bucket.

Against a live cluster, scripts/kafka-configs.sh drives the quota CRUD surface (--alter / --describe user quotas, all four Apache quota keys) and ends with a live throttle probe that pushes an unbounded offered rate through the Apache kafka-producer-perf-test tool against a 10 MB/s cap.

KIP-32 — Record timestamps

Status: partial — see the KIP index. CreateTime timestamps round-trip byte-identically; LogAppendTime and timestamp→offset lookup are not implemented (details below).

What the KIP changes in Apache Kafka

Kafka 0.10 added a timestamp to every message, plus the message.timestamp.type topic config choosing between CreateTime (producer-supplied, the default) and LogAppendTime (broker overwrites the timestamp at append). Timestamps feed time-based retention, compaction lag, and offset-for-timestamp lookup.

How kaas implements it

Timestamps are honoured byte-opaquely. RecordBatches flow through the broker as uninterpreted bytes (the byte-opacity contract), so whatever timestamps and timestamp-type attribute bits the producer wrote are stored and served back byte-identical. Concretely:

• The only timestamp the broker reads is the batch-level MaxTimestamp: parse_batch_offsets in crates/kaas-storage/src/segment.rs pulls it from bytes [35..43] of the v2 batch header — the records payload is never touched. Each segment tracks the highest value seen (ActiveSegment::max_timestamp).
• LogAppendTime is not implemented. kaas never rewrites timestamps. message.timestamp.type is advertised by DescribeConfigs as CreateTime and the config-admin path accepts that value as an accept-and-drop key — validated, never persisted (crates/kaas-broker/src/topic_cr_writer.rs) — while LogAppendTime is rejected with INVALID_CONFIG; the per-topic config file itself (crates/kaas-storage/src/topicconfig.rs, which carries retention, segment, flush, and compaction knobs) has no timestamp-type field. The Produce response always returns log_append_time_ms: -1 — the CreateTime sentinel (crates/kaas-broker/src/handlers/produce.rs). A topic cannot ask the broker to stamp append time; every topic behaves as CreateTime.
• Timestamp-based ListOffsets lookup is a gap. EARLIEST (-2) and LATEST (-1) resolve via log-start / high-watermark (crates/kaas-broker/src/handlers/list_offsets.rs), but a concrete timestamp returns the (-1, -1) "no matching offset" sentinel on both engines (crates/kaas-storage/src/disk.rs, crates/kaas-storage/src/memory.rs) — the segment-level max_timestamp tracking is in place, the timestamp→offset index that would answer the query is a follow-up.
• max.message.time.difference.ms validation does not exist — it would require reading record payloads, which the opacity contract forbids.

How it's verified

• crates/kaas-storage/src/segment.rs: create_then_append_one_batch_updates_state asserts MaxTimestamp is parsed out of the 43-byte header and tracked per segment.
• crates/kaas-storage/src/memory.rs: offset_for_timestamp_sentinel pins the honest "no match" answer for timestamp queries.
• bins/kaas/tests/byte_opacity.rs — the tripwire integration test that proves records (timestamps included) round-trip byte-identical and no code path decoded them.
• scripts/kafka-get-offsets.sh exercises --time -2 / --time -1 against a live broker; scripts/kafka-console-producer.sh / scripts/kafka-console-consumer.sh round-trip producer-stamped records.

KIP-58 — min compaction lag

Status: partial — config surface only; the gate itself is not yet enforced. See the KIP index.

What the KIP changes in Apache Kafka

KIP-58 (Kafka 0.10.1) added min.compaction.lag.ms: a per-topic guarantee that a record stays uncompacted for at least the configured period. The log cleaner treats the head of the log inside the lag window as off-limits, so consumers of a compacted topic get a bounded window in which they are guaranteed to see every update, not just the latest per key.

How kaas implements it

What exists today is the configuration plumbing, end to end:

• min.compaction.lag.ms is a field on the per-topic config file the operator materialises from KafkaTopic.spec.config (min_compaction_lag_ms in crates/kaas-storage/src/topicconfig.rs, written to /data/<topic>/.config.json; None is kept distinct from 0 so "unset" falls through to the engine default).
• IncrementalAlterConfigs accepts the key and patches it onto the CR (crates/kaas-broker/src/topic_cr_writer.rs).
• DescribeConfigs advertises it with default 0 — "compact immediately", matching Apache — from the defaults table in crates/kaas-broker/src/topic_config_defaults.rs.

What does not exist yet is the compactor that would honour the gate. crates/kaas-storage/src/cleaner.rs implements size- and time-based delete-policy retention (whichever target reaps more), and its module doc says so plainly: the compactor honouring these knobs (min.compaction.lag.ms, delete.retention.ms) is follow-up work (tracked as gh #158). The compaction metrics in crates/kaas-observability/src/metrics.rs (kaas.compaction.*) are declared ahead of that work and nothing records them today. So the knob round-trips through every admin surface but gates nothing — and "never compacted" is no longer the safe reading of an infinite lag: the retention cleaner does not consult cleanup.policy, and the enforced retention.ms default of 7 days means a cleanup.policy=compact topic's closed segments age out like any other's unless retention.ms is set to -1. Until the compactor lands, data on compacted topics is subject to delete-style retention.

The intended enforcement semantics (per-segment maxTimestamp inside the lag window ⇒ segment skipped) are described with the storage engine in Storage hot path; treat that section as design intent until the compactor lands.

How it's verified

Verification currently covers the config plumbing, not the gate: roundtrip_preserves_unset_vs_zero and only_set_fields_are_emitted in crates/kaas-storage/src/topicconfig.rs pin the unset-vs-zero distinction, and scripts/kafka-configs.sh exercises the --describe / --alter round trip against a live broker. There are no compaction-behaviour tests because there is no compaction behaviour to test — their absence is the honest signal of this page's status.

KIP-98 — exactly-once foundation

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-98 (Kafka 0.11) is the exactly-once foundation, in two layers. The idempotent producer gets a producer ID and epoch from InitProducerId and stamps every batch with per-partition sequence numbers, letting the broker deduplicate retries. Transactions add a transaction coordinator (state in the __transaction_state internal topic), atomic multi-partition writes terminated by COMMIT/ABORT control batches, and a read_committed isolation level in which consumers only see records below the last stable offset (LSO) and filter aborted transactions. Every txn API — AddPartitionsToTxn, AddOffsetsToTxn, TxnOffsetCommit, EndTxn, WriteTxnMarkers — originates here.

How kaas implements it

The full state machine is documented in Transactions & idempotence; this page maps the KIP's pieces to source and names the substitutions.

Idempotent producer. InitProducerId (key 22, v0–v4, crates/kaas-broker/src/handlers/init_producer_id.rs) hands non-transactional producers a fresh PID with epoch 0 from a per-broker partitioned block allocator (crates/kaas-broker/src/producer_id.rs): pid = (broker_id + 1) · 2^40 + local, with local advancing in 1000-PID blocks whose end is persisted before any PID in the block is handed out — so a restart skips forward instead of reissuing, and brokers can never collide (transactional_id of "" counts as non-transactional, the KIP-98 client convention). Per-partition dedupe lives in crates/kaas-storage/src/idempotence.rs: a five-batch ring per PID — sized to Java's max.in.flight.requests.per.connection=5 — classified under the partition mutex as duplicate (echo the cached baseOffset, no log write), out-of-order (wire error 45), invalid epoch (wire 47), or accept. Only the 57-byte v2 batch header is parsed; record payloads stay opaque. The window survives restart via producer-state.snapshot (crates/kaas-storage/src/producer_snapshot.rs), written on take-over and close/relinquish beside manifest.json.

Transactions — with three honest substitutions:

• No __transaction_state topic. Coordinator state is slot-sharded JSON — /data/__cluster/txn_state/slot-N.json, 50 slots matching Apache's transaction.state.log.num.partitions (crates/kaas-coordinator/src/txn_state.rs). Failover is "open the file", no log replay. Rationale in Non-goals.
• Markers via the shared volume, not an RPC. EndTxn (crates/kaas-broker/src/handlers/end_txn.rs) writes control batches directly to partitions this broker leads; for peers it enqueues one JSON file per (pid, epoch, target) under /data/__cluster/marker_queue/to-<broker>/ (crates/kaas-coordinator/src/marker_queue.rs), which each broker's marker_watcher polls and applies (crates/kaas-broker/src/marker_watcher.rs). EndTxn returns success once the queue entry is written (EndTxn).
• read_committed via LSO clamp. The Fetch handler (crates/kaas-broker/src/handlers/fetch.rs) caps reads at the partition's last stable offset and returns AbortedTransactions[] from the aborted-txn index (crates/kaas-storage/src/txn_index.rs).

The remaining txn handlers (add_partitions_to_txn.rs, add_offsets_to_txn.rs, txn_offset_commit.rs, write_txn_markers.rs under crates/kaas-broker/src/handlers/) drive the Empty → Ongoing → PrepareCommit/PrepareAbort → CompleteCommit/CompleteAbort transitions. EndTxn is two-phase: the prepare transition retains the partition and group lists as the durable dispatch set, and the complete transition runs only once every marker is durable — a dispatch failure leaves the entry prepared and retriable. A 10 s timeout reaper lands overdue transactions in PrepareAbort with an epoch bump, and a marker-reconcile pass on the same tick dispatches their ABORT markers and completes them. Epoch fencing on producer rejoin is KIP-360; transactional consume-process-produce offsets are KIP-447.

How it's verified

bins/kaas/tests/eos_v2.rs runs the whole loop over real TCP with hand-rolled wire bytes: eos_commit_path_records_visible_to_read_committed and eos_abort_path_populates_aborted_transactions. Unit coverage: duplicate_in_window_returns_cached_offset, fresh_pid_first_seq_must_be_zero, older_epoch_is_invalid, and ring_caps_at_five_entries in idempotence.rs; roundtrip_through_atomic_write and future_version_is_dropped_not_misinterpreted in producer_snapshot.rs; first_call_allocates_epoch_zero_rejoin_bumps, end_txn_happy_commit_clears_partitions_and_fires_hook, persistence_round_trip_across_open, and reaper_aborts_overdue_bumps_epoch_fires_hook in txn_state.rs. Shell suite: scripts/kafka-verifiable-producer.sh (explicit enable.idempotence=true run asserting no OutOfOrderSequence / InvalidProducerEpoch), scripts/kafka-txn-coordinator.sh, scripts/kafka-txn-timeout.sh, and scripts/kafka-transactions.sh.

KIP-101 — leader-epoch-based log truncation

Status: partial — see the KIP index.

What's missing

• The leader-epoch cache and lookup. DiskStorageEngine::offset_for_leader_epoch (crates/kaas-storage/src/disk.rs) returns the (-1, -1) sentinel — the segment epochs are recorded on disk but no epoch→offset cache is materialized.
• Wire key 23 (OffsetForLeaderEpoch) is not registered — clients see it as unsupported via ApiVersions (it appears in the matrix's gap table).

What the KIP changes in Apache Kafka

KIP-101 replaced high-watermark-based follower truncation with leader-epoch-based truncation: each replica tracks which offset ranges were written under which leader epoch, and a rejoining follower asks the leader "what's the last offset for epoch E?" (OffsetForLeaderEpoch) to truncate divergent history precisely instead of over- or under-truncating.

What kaas has

Half of the KIP's substance — the epoch bookkeeping — exists structurally: every segment filename carries the leader epoch it was written under ({epoch:08x}-{base_offset:020d}.log), so the divergent-history problem the KIP solves is prevented by construction rather than repaired after the fact. A deposed leader's late writes land in files named with a dead epoch and are never part of the new leader's log (storage engine).

Because kaas has no replication, there are no followers to truncate — the KIP's driving scenario doesn't arise inside the broker. What does still want the lookup is the client-side surface: KIP-320-aware consumers send OffsetForLeaderEpoch to detect log truncation after unclean leadership changes. Registering key 23 backed by a real epoch cache is the tracked follow-up.

How the partial state is verified

The segment-epoch construction is exercised by the storage engine's unit tests and the takeover paths in bins/kaas/tests/cluster_smoke.rs; the stubbed lookup is explicit in crates/kaas-storage/src/disk.rs (grep for offset_for_leader_epoch).

KIP-107 — DeleteRecords

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 0.11 added AdminClient.deleteRecords() and the DeleteRecords API (key 21): per partition, advance the log start offset to a caller-chosen target (-1 = purge everything up to the high watermark). Records below the new log start become invisible to Fetch and eligible for physical deletion; the response returns the resulting low watermark.

How kaas implements it

• The handler (crates/kaas-broker/src/handlers/delete_records.rs, key 21, v0–v2, flexible from v2) gates on partition ownership the same way Produce does: with a cluster coordinator wired, partitions this broker doesn't lead answer NOT_LEADER_OR_FOLLOWER (6). A target past the high watermark maps to OFFSET_OUT_OF_RANGE (1).
• The storage side (Partition::delete_records in crates/kaas-storage/src/partition.rs) runs under the partition mutex: resolve -1 to the HWM, advance log_start, then physically unlink every closed segment (log + index) that falls entirely below the new log start, and publish a fresh read snapshot. Because only the leader holds open file handles (file-handle ownership), the unlink actually frees space on NFS instead of silly-renaming.
• The advanced log_start persists via manifest.json, which is written on take-over and close/relinquish rather than inline — the reopen path recovers it. It also surfaces everywhere Apache surfaces it: ListOffsets EARLIEST, the Produce response's log_start_offset, and Fetch visibility.
• The same primitive drives retention: the cleaner (crates/kaas-storage/src/cleaner.rs) computes size- and time-based targets (cleanup_target_for_size_bytes / cleanup_target_for_time), takes whichever reaps more, and calls delete_records with it, so there is exactly one log-start-advance code path.
• One asymmetry against upstream: delete_records drops closed segments only, and the cleaner never touches the active segment either. A purge-to-HWM makes the active segment's records invisible immediately, but its bytes are reclaimed only after the segment rolls and a later cleanup pass drops it.

How it's verified

• crates/kaas-storage/src/partition.rs unit tests: delete_records_advances_log_start, delete_records_purge_to_hwm, delete_records_past_hwm_is_offset_out_of_range.
• crates/kaas-storage/src/disk.rs: reopen_recovers_state proves the advanced log start survives an engine restart (HWM 6, log start 2 after reopen).
• scripts/kafka-delete-records.sh runs the real kafka-delete-records.sh tool end-to-end: produce 10 records, delete to offset 7 via --offset-json-file, assert earliest = 7.

KIP-195 — CreatePartitions

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 1.0 added AdminClient.createPartitions() and the CreatePartitions API (key 37): increase an existing topic's partition count (the request carries the new total, not a delta), optionally with manual replica assignments for the new partitions, plus a validateOnly mode. Shrinking is rejected.

How kaas implements it

The handler (crates/kaas-broker/src/handlers/create_partitions.rs, key 37, v0–v3, flexible from v2) doesn't touch storage at all. After authorizing Alter on the topic, it translates the request into a merge PATCH on KafkaTopic.spec.partitions via the installed TopicCRWriter (crates/kaas-broker/src/topic_cr_writer.rs). The operator reconciles the CR as usual — creating the new partition directories on the shared volume — and the broker observes them through its topic watcher. This is the same pattern as IncrementalAlterConfigs's config writes: admin writes go through the CR so there is one materialization path, and the broker's RBAC carries update,patch on kafkatopics to allow it.

The writer runs a client-side shrink guard (read current spec.partitions, refuse a decrease with INVALID_PARTITIONS (37)) before patching; a missing CR maps to UNKNOWN_TOPIC_OR_PARTITION (3), RBAC denial to CLUSTER_AUTHORIZATION_FAILED (31). In dev mode (no kube client) every write answers 31 with an explanatory message.

Honest corners, all visible in the source:

• assignments is decoded and ignored. kaas has no replicas (non-goals); partition placement belongs to the controller's balancer, not the caller.
• timeout_ms is ignored. The handler returns success once the CR patch is accepted — before the operator has created the directories. Apache waits up to the timeout for the partitions to exist; with kaas a Metadata refresh moments later shows the new count.
• validate_only validates less than upstream. It short-circuits after the authorization and writer checks but before the CR read, so it reports success even for a nonexistent topic or a would-be shrink.

How it's verified

• crates/kaas-codec/src/api/create_partitions.rs round-trip tests: v0_roundtrip, v1_carries_error_message, v2_is_flexible, v3_roundtrip, validate_only_round_trips.
• crates/kaas-broker/src/topic_cr_writer.rs: noop_writer_returns_forbidden pins the dev-mode error mapping.
• scripts/kafka-topics.sh scenario 4 drives kafka-topics.sh --alter --partitions N against a live cluster and asserts the described partition count actually changes.

See also the CreatePartitions entry in the per-API reference.

KIP-219 — improved quota throttle communication

Status: partial — see the KIP index.

What's missing

The broker never mutes the connection after a throttled response. kaas computes throttle_time_ms and returns it, but continues reading the socket immediately — so throttle enforcement relies on the client honouring the advertised delay. A well-behaved Java/librdkafka/franz-go client backs off exactly as it would against Apache; a client that ignores the field can exceed its quota.

What the KIP changes in Apache Kafka

Before KIP-219, a quota-violating request was held on the broker for the throttle duration before the response was sent — which interacted badly with client-side request timeouts. KIP-219 changed the contract: the broker sends the response immediately with throttle_time_ms set, then mutes the connection's channel for the throttle window, so enforcement no longer depends on client cooperation while responses stay prompt.

What kaas has

The signalling half, on accurate foundations:

• Quota accounting is a token bucket with debt-carry (crates/kaas-auth/src/quota.rs): overshoot goes negative and is carried forward rather than clamped at zero, so N concurrent clients can't each see a "full" bucket and burst at N× the configured rate. The multi_client_contention_carries_debt unit test pins this.
• Handlers surface the resulting delay as throttle_time_ms in responses (checked once per request on the Produce path — see Produce).

Channel muting is the tracked follow-up; the quota architecture is covered in Listeners, authentication, authorization and per-broker quota semantics in KIP-13.

How the partial state is verified

crates/kaas-auth/src/quota.rs unit tests (debt-carry, contention); scripts/kafka-configs.sh exercises quota configuration end-to-end. The absence of muting is verifiable by inspection — there is no mute path in crates/kaas-protocol/src/.

KIP-255 — SASL/OAUTHBEARER

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-255 (Kafka 2.0) added the OAUTHBEARER SASL mechanism (RFC 7628): clients authenticate by presenting an OAuth 2 bearer token instead of a password, with the token's claims carrying the principal. Apache ships an unsecured JWT implementation by default and leaves production validation to pluggable callback handlers — in practice everyone runs Strimzi's kafka-oauth handlers, which validate real JWTs against an OIDC issuer's JWKS.

How kaas implements it

kaas skips the unsecured default and implements the production shape directly: a listener with authentication.type: oauth validates real JWTs against the configured issuer — the equivalent of Strimzi's oauth.valid.issuer.uri + oauth.jwks.endpoint.uri fast-local-JWKS path, with the config field names mirroring Strimzi's KafkaListenerAuthenticationOAuth 1:1.

• crates/kaas-auth/src/oauth.rs — RFC 7628 message parsing (the %x01-framed initial response, gs2 authzid check, the two-step JSON-challenge failure path), JWT validation (signature via the issuer's JWKS, exp/nbf with 60 s skew, exact iss, optional aud pinning), and principal extraction (userNameClaim, default sub, with fallbackUserNameClaim).
• Signature verification uses ring with an alg allowlist (RS256/RS384/RS512/ES256); none and the HMAC family are rejected outright (algorithm-confusion defence).
• The JWKS is hot-swapped: a fetch loop in bins/kaas/src/main.rs (spawn_jwks_refreshers) re-pulls every jwksRefreshSeconds (default 300, Strimzi's default) and early on unknown-kid. Until the first successful fetch every token is rejected — fail closed.
• OAUTHBEARER is refused over plaintext connections, like SASL PLAIN: a bearer token is a reusable credential.
• Mechanism advertisement is per listener: an oauth listener's SaslHandshake response lists OAUTHBEARER alone, never mechanisms the engine would reject at authenticate time.

Deliberate deviation: Apache's default unsecured OAUTHBEARER (base64-JSON "tokens" with no signature) is not implemented at all — there is no mode in which kaas accepts an unsigned token.

How it's verified

Unit tests in crates/kaas-auth/src/oauth.rs cover the RFC 7628 parser edges, a static RS256 vector, ES256 round-trips, expiry/nbf skew, issuer and audience mismatches, alg-confusion rejections (alg_none_and_hs256_rejected), unknown-kid refresh hints, and the two-step failure exchange. bins/kaas/tests/oauth_smoke.rs drives the full stack over TLS: pre-auth gate → handshake → OAUTHBEARER with an ES256 JWT validated against a wiremock-served JWKS → Produce unblocked, plus the challenge-then-58 failure path. Live validation: the canary producer/consumer pair authenticating via EntraID workload identity (the same pair that runs against Strimzi's OAuth listener).

KIP-290 — Prefixed ACL resource patterns

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 2.0 gave every ACL binding a pattern type: LITERAL (exact name, with * as a special literal wildcard) or PREFIXED (the binding covers every resource whose name starts with the stored prefix). Describe/Delete filters additionally gained a MATCH type that finds all bindings — literal, wildcard, or prefixed — affecting a given resource. The ACL wire APIs bumped to v1 to carry the new field.

How kaas implements it

• Evaluation lives in matches_resource in crates/kaas-auth/src/acls.rs: literal matches on equality (with the * wildcard special-case), prefix uses starts_with, and a missing patternType defaults to literal. Deny short-circuits over allow, default is deny, and decisions sit in a 5-second cache that is wiped on every hot reload so a fresh deny rule bites immediately.
• Authoring is CR-first: rules live on KafkaUser.spec.authorization.acls, whose CRD schema admits exactly literal and prefix (crates/kaas-operator-api/src/kafkauser.rs). The operator materialises them to /data/__cluster/acls.json (crates/kaas-operator-controllers/src/acls.rs); every broker hot-reloads the file.
• The wire ACL trio (DescribeAcls 29 / CreateAcls 30 / DeleteAcls 31, all v0–v3, crates/kaas-broker/src/handlers/acls.rs) carries pattern types from v1 per the KIP; the enum mapping (LITERAL=3, PREFIXED=4) is in crates/kaas-codec/src/api/acl_types.rs. v0 requests are pinned to literal semantics on both create and filter, matching pre-KIP-290 behaviour. Writes go through the AclCRWriter onto the matching KafkaUser CR, so wire-created ACLs and CR-authored ACLs are the same rules.

Honest corners:

• MATCH filters are approximated. match_pattern in crates/kaas-broker/src/acl_cr_writer.rs expands a MATCH filter to "literal or prefix bindings", but the resource-name axis still compares exactly — Apache's MATCH also finds prefixed bindings whose prefix merely covers the queried name, which kaas does not.
• ACL host fields are stored but not enforced — only "any host" is evaluated today (noted in the CRD itself).
• CreateAcls requires the principal's KafkaUser CR to exist; kaas has nowhere else to persist a rule.

How it's verified

• crates/kaas-auth/src/acls.rs unit tests: prefix_pattern, deny_overrides_allow, reload_swaps_atomically, anonymous_default_denies.
• crates/kaas-broker/src/handlers/acls.rs: wire_binding_translation_maps_enums (PREFIXED → prefix), v0_filter_defaults_to_literal_pattern.
• scripts/kafka-acls.sh runs kafka-acls.sh CRUD end-to-end via a temporary KafkaUser CR (literal patterns only — it does not exercise prefixed bindings).

See also Listeners & auth for where the authorizer sits in the request path.

KIP-339 — IncrementalAlterConfigs

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 2.3 added IncrementalAlterConfigs (key 44): per-key SET / DELETE / APPEND / SUBTRACT operations on a resource's configuration, replacing the original AlterConfigs' read-modify-write-the-whole-config semantics that raced concurrent admins.

How kaas implements it

The handler (crates/kaas-broker/src/handlers/incremental_alter_configs.rs, key 44, v0–v1, flexible from v1) follows the same CR-write pattern as KIP-195: after authorizing AlterConfigs on the topic, each op becomes part of a merge PATCH on KafkaTopic.spec.config (crates/kaas-broker/src/topic_cr_writer.rs). The operator materialises the change to /data/<topic>/.config.json (crates/kaas-storage/src/topicconfig.rs), which brokers hot-reload.

The surface is deliberately narrower than the KIP, and the code says so rather than pretending:

• TOPIC resources only. BROKER and BROKER_LOGGER return UNSUPPORTED_VERSION (35) — kaas has no dynamic broker-config surface.
• SET and DELETE only. APPEND / SUBTRACT are list-valued ops; every config key kaas supports is scalar, so the writer returns UnsupportedOp and the wire sees 35. SET with a null value is treated as DELETE (patched to JSON null).
• Eight persisted config keys. config_key_to_json_field maps retention.ms, retention.bytes, segment.ms, segment.bytes, cleanup.policy, min.compaction.lag.ms, delete.retention.ms, and flush.messages (dotted or camelCase spellings) onto the CR's fields. A ninth key, message.timestamp.type, is accept-and-drop: CreateTime is validated and deliberately not persisted (kaas is CreateTime-only by construction — KIP-32), while LogAppendTime is rejected. Unknown keys and unparseable values are refused with INVALID_CONFIG (40) — the API server silently prunes unknown merge-patch fields, so anything not rejected client-side would become a success that does nothing.
• validate_only validates for real. Validation runs before the validate-only short-circuit, so a validate-only call answers the same error codes the real call would — unknown keys and bad values answer 40, APPEND/SUBTRACT ops answer 35.
• Per-resource error codes ride in the response body; a missing CR maps to UNKNOWN_TOPIC_OR_PARTITION (3), RBAC denial and dev mode to CLUSTER_AUTHORIZATION_FAILED (31).

How it's verified

• crates/kaas-broker/src/topic_cr_writer.rs unit tests: config_key_to_json_field_matches_known_keys, config_value_parses_integer_fields (integers become JSON numbers, unparseable values fall back to strings for operator-side schema rejection), noop_writer_returns_forbidden.
• crates/kaas-codec/src/api/incremental_alter_configs.rs round-trip tests cover both versions.
• scripts/kafka-configs.sh scenario 3 drives kafka-configs.sh --alter --add-config retention.ms=... end-to-end. (The script's comments still label it XFAIL from before the handler landed — a stale label, not a stale implementation.)

See also the IncrementalAlterConfigs entry in the per-API reference.

KIP-345 — static consumer-group membership

Status: partial — see the KIP index.

What's missing

FENCED_INSTANCE_ID fencing of duplicate static members. In Apache, a second consumer joining with an already-active group.instance.id (but a different member ID) is fenced with error 82 so a misconfigured duplicate can't hijack the instance's partitions. kaas defines no such check — the error code appears nowhere in the tree — so a duplicate static member is treated as a rejoin rather than fenced.

What the KIP changes in Apache Kafka

KIP-345 lets a consumer declare a stable identity (group.instance.id). The coordinator then treats restarts of that process as the same member: no rebalance on a bounce within the session timeout, and no partition churn for rolling restarts of stable fleets.

What kaas has

The behaviour that matters for rolling restarts:

• group.instance.id is decoded and carried through join/sync/leave state (crates/kaas-coordinator/src/group.rs threads group_instance_id through members, join waiters, and describe output).
• Static members survive the rebalance-eviction sweep: evict_non_rejoining_members only evicts members whose group_instance_id is None — a static member that hasn't re-joined within rebalance_timeout_ms keeps its slot instead of being dropped (crates/kaas-coordinator/src/group.rs).

See the per-API details on JoinGroup and LeaveGroup.

How the partial state is verified

Group-coordinator unit tests in crates/kaas-coordinator/src/group.rs cover static-member retention across rebalances; scripts/kafka-consumer-groups.sh exercises group lifecycle end-to-end. The missing fencing is verifiable by inspection: FENCED_INSTANCE_ID has no definition or use anywhere in the workspace.

KIP-354 — tombstone retention

Status: partial — config surface only; tombstone expiry is not yet enforced. See the KIP index.

What the KIP changes in Apache Kafka

Upstream, KIP-354 (Kafka 2.3) is titled "Add a Maximum Log Compaction Lag" and adds max.compaction.lag.ms: an upper bound on how long a record — including a tombstone — can sit uncompacted, so deletion is guaranteed to happen within a bounded time (the GDPR-shaped use case). The tombstone-lifetime knob itself, delete.retention.ms, predates the KIP by years: it bounds how long a delete marker survives after its segment is cleaned, so lagging consumers replaying the compacted log still observe the deletion.

What kaas actually has

kaas's source (crates/kaas-storage/src/topicconfig.rs) attributes its delete.retention.ms field to KIP-354 — the shared goal is bounded tombstone lifetime — but to be precise: kaas implements the delete.retention.ms config surface, and max.compaction.lag.ms does not exist anywhere in kaas (no config field, no CRD entry, no defaults row). The declared kaas semantics also differ from Apache in two ways:

• Expiry granularity is per batch, keyed on the batch baseTimestamp, where Apache decides per record — a consequence of the storage engine never opening batches (Storage hot path).
• 0 is documented as "tombstones live forever" (crates/kaas-broker/src/topic_config_defaults.rs), where Apache's 0 means tombstones are removable as soon as the segment is cleaned.

Like KIP-58, what ships today is the plumbing, not the enforcement. The field flows from KafkaTopic.spec.config through the operator into /data/<topic>/.config.json (delete_retention_ms, topicconfig.rs), IncrementalAlterConfigs accepts the key (crates/kaas-broker/src/topic_cr_writer.rs), and DescribeConfigs advertises a default of 86400000 (24 h, matching Apache) from topic_config_defaults.rs. But the compactor that would drop expired tombstones does not exist yet: crates/kaas-storage/src/cleaner.rs implements size- and time-based delete-policy retention and marks the compactor honouring both knobs as follow-up work (tracked as gh #158). No compaction runs, so no tombstone is ever compacted away — but the outcome is no longer safely conservative. The retention cleaner does not consult cleanup.policy, and the enforced retention.ms default of 7 days means a compacted topic's closed segments — tombstones and live keys alike — age out unless retention.ms is set to -1. With respect to the KIP's guaranteed-removal goal: nothing is guaranteed to disappear within delete.retention.ms, and until the compactor lands everything old enough disappears wholesale under delete-style retention.

How it's verified

Config-plumbing level only: roundtrip_preserves_unset_vs_zero in crates/kaas-storage/src/topicconfig.rs round-trips a set delete_retention_ms next to unset siblings, and scripts/kafka-configs.sh covers the --describe / --alter surface against a live broker. Tombstone-expiry behaviour has no tests because the behaviour is not implemented.

KIP-360 — producer epoch bump

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-360 (Kafka 2.5) lets a producer recover from state loss without losing its identity. InitProducerId v3+ can carry the producer's current PID and epoch; the coordinator responds by bumping the epoch on the same PID rather than minting a new producer, so retried batches from the old session are fenced (PRODUCER_FENCED) instead of poisoning the log, and UNKNOWN_PRODUCER_ID becomes retriable. When the 16-bit epoch is exhausted, the coordinator rotates to a fresh PID.

How kaas implements it

The rejoin contract lives in crates/kaas-coordinator/src/txn_state.rs::get_or_allocate_with_timeout: the first InitProducerId for a transactional.id allocates a fresh PID at epoch 0; every subsequent call returns the same PID with epoch + 1. At epoch == i16::MAX the entry rotates to a fresh PID at epoch 0, matching the KIP's exhaustion rule. The timeout reaper's abort path also bumps the epoch, so a producer returning after its transaction was reaped is fenced rather than resumed.

The bump is then propagated in two rings (fencing details):

• Cross-partition, in-process: after any bump to epoch > 0, the handler (crates/kaas-broker/src/handlers/init_producer_id.rs) calls the engine's fence_producer_epoch, which advances the PID's epoch and clears the dedupe window on every partition this broker leads — a zombie batch is rejected even on partitions the new session hasn't touched.
• Cross-broker, via the shared volume: the bump is appended to this broker's outbound fence file, /data/__cluster/producer_fences/from-<broker>.json (crates/kaas-coordinator/src/fence_log.rs); peers' FenceWatcher (crates/kaas-broker/src/fence_watcher.rs) polls the directory every 2 s and applies newer (pid, epoch) pairs. Polling is deliberate: inotify does not fire for another NFS client's writes.

Where kaas differs from Apache: the v3+ request's producer_id / producer_epoch fields are decoded but not validated — every InitProducerId for a known transactional.id bumps, unconditionally. The single-writer guarantee is preserved (the highest epoch wins and everything older is fenced), but Apache's claimed-identity check — which lets the coordinator reject a stale caller with PRODUCER_FENCED rather than hand it a newer epoch — has no counterpart here. In the boot window or dev mode, before a TxnStateStore is wired, the handler falls back to a fresh PID and logs that rejoin fencing is disabled — graceful degradation, not silent.

How it's verified

Handler tests in init_producer_id.rs: transactional_rejoin_bumps_epoch (same PID, epochs 0 → 1 → 2) and rejoin_appends_to_fence_log_for_broadcast (epoch 0 does not broadcast; each bump overwrites the outbound entry). Store tests in txn_state.rs: epoch_overflow_rotates_to_fresh_pid, epoch_mismatch_fences, reaper_aborts_overdue_bumps_epoch_fires_hook. Fence propagation: append_lower_or_equal_epoch_is_noop and append_higher_epoch_overwrites in fence_log.rs; applies_peer_fences, skips_self_file, dedupe_across_ticks, and higher_epoch_after_first_apply_fires_once_more in fence_watcher.rs; fence_bumps_epoch_and_clears_window in crates/kaas-storage/src/idempotence.rs. Shell suite: scripts/kafka-txn-coordinator.sh probes the rejoin-epoch wire contract against a live broker.

KIP-368 — SASL re-authentication (partial)

Status: partial — see the KIP index.

What's missing

• A broker-wide connections.max.reauth.ms equivalent. kaas bounds sessions only where the credential itself expires: oauth listeners with maxSecondsWithoutReauthentication set. SCRAM/PLAIN sessions are never asked to re-authenticate (their session_lifetime_ms is always 0), which matches Apache's default but cannot be tightened per broker the way Apache allows.
• Apache closes an over-deadline connection after failing the next request; kaas keeps answering SASL_AUTHENTICATION_FAILED (58) to every non-SASL request instead of disconnecting. Cooperative clients re-authenticate or reconnect either way.

What landed

KIP-368 (Kafka 2.2) lets a client re-run the SASL exchange on a live connection before its session expires, with the broker advertising the deadline as session_lifetime_ms in the SaslAuthenticate response.

On an oauth listener with maxSecondsWithoutReauthentication set, a successful authentication advertises min(configured bound, token remaining lifetime) in milliseconds, and the dispatcher (crates/kaas-protocol/src/dispatch.rs) refuses every non-SASL API past the deadline until a fresh exchange completes — a connection cannot outlive its bearer token by more than the configured bound. Re-authentication is accepted on any listener at any time, with one guard (crates/kaas-broker/src/handlers/sasl.rs): the new exchange must resolve to the same principal — a re-auth that would swap identities fails with SASL_AUTHENTICATION_FAILED rather than laundering one principal's connection into another's.

How it's verified

exchange_session_lifetime_capped in crates/kaas-auth/src/oauth.rs pins the min(bound, token lifetime) arithmetic; oauthbearer_end_to_end_over_tls in bins/kaas/tests/oauth_smoke.rs asserts the advertised lifetime on the wire. The client half is exercised live by the canary pair, whose Kafka library re-authenticates on the broker's advertised timer.

KIP-371 — mTLS principal mapping

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-371 (Kafka 2.2) added ssl.principal.mapping.rules: a broker config that turns an mTLS client certificate's X.500 subject DN into a short principal name without writing a custom PrincipalBuilder class. Rules are a comma-separated list of RULE:<regex>/<replacement>/[L|U] entries plus DEFAULT; the first matching rule wins. It mirrors what sasl.kerberos.principal.to.local.rules already did for Kerberos names.

How kaas implements it

crates/kaas-auth/src/principal_mapping.rs parses Apache's rule syntax:

• RULE:<regex>/<replacement>/ with $1, $2, … back-references into the regex's capture groups, matched against the full subject DN.
• Optional /L (lowercase) or /U (uppercase) postfix on the result.
• First matching rule wins; commas inside a regex body are handled — the spec is split only at commas immediately followed by RULE: or DEFAULT, because subject DNs use commas as their own separator.

The rules arrive via the KAAS_SSL_PRINCIPAL_MAPPING_RULES env (crates/kaas-broker/src/cli.rs) and are compiled once at startup in bins/kaas/src/main.rs — a parse error (bad syntax, invalid regex) fails boot rather than silently mapping every certificate to its CN, so a chart-config typo is loud. The compiled mapper is wired into the accept path in crates/kaas-protocol/src/server.rs; after the TLS handshake, crates/kaas-auth/src/mtls.rs extracts the leaf certificate's subject DN, applies the mapper, and authenticates the mapped name against the credentials store. The resulting principal is what the cluster-wide authorizer and quota checker see.

One deliberate deviation: in Apache, DEFAULT (and the no-match fall-through) yields the full DN string as the principal. In kaas both return the certificate's CN (principal_mapping.rs, Rule::Default and the fall-through in apply), and an empty rule spec behaves the same. That matches how kaas keys credentials.json entries — operator-managed KafkaUser names, not DN strings — and preserves the pre-mapping behaviour for clusters that never set the env. If you need Apache's DN-shaped principals, an explicit RULE:^(.*)$/$1/ reproduces them.

How it's verified

Unit tests in crates/kaas-auth/src/principal_mapping.rs: first_rule_wins_with_back_reference, lower_case_flag_applied, upper_case_flag_applied, split_rules_keeps_dn_commas_inside_regex (the comma-in-DN parser edge), multiple_rules_first_match_wins, unmatched_rule_falls_through_to_cn, empty_spec_returns_cn, and the fail-fast pair invalid_rule_syntax_errors / invalid_regex_errors. The DN-extraction + mapper + engine composition is covered in crates/kaas-auth/src/mtls.rs's tests. There is no shell-suite mTLS scenario — the Apache CLI tools in scripts/ bootstrap over the anonymous or SCRAM listeners.

KIP-394 — require member ID for initial join

Status: partial — see the KIP index.

What's missing

The v4+ two-step join handshake. kaas defines the MEMBER_ID_REQUIRED error code (79, crates/kaas-coordinator/src/group.rs) but never returns it: join() takes the legacy assign-inline path for all versions — a first-time joiner is admitted directly instead of being bounced with a broker-assigned member ID and re-joining with it. The follow-up marker sits in the join path (crates/kaas-coordinator/src/group.rs, the pending-member registry comment).

What the KIP changes in Apache Kafka

Before KIP-394, a JoinGroup that timed out client-side could leave a ghost member in the group: the broker had created a member ID the client never learned. From JoinGroup v4, an initial join with an empty member ID is answered MEMBER_ID_REQUIRED plus a broker-assigned member ID; the client immediately re-joins with that ID. Ghosts stop accumulating because only joins that echo a known ID create members.

What kaas has, and why the gap is tolerable so far

kaas serves JoinGroup v2–v9 (matrix) with the pre-KIP-394 admission flow at every version. The ghost-member problem the KIP fixes is bounded by two kaas behaviours: session-timeout eviction reaps members whose clients vanished, and the rebalance sweep evicts non-rejoining dynamic members (see KIP-345 for the static-member exception). Clients negotiate and operate correctly — the Java client only sends the empty member ID on first join and handles either reply shape — but the tracked follow-up is to implement the real two-step handshake rather than rely on timeouts.

See JoinGroup for the full join flow.

How the partial state is verified

Join/rebalance behaviour is covered by group-coordinator unit tests and scripts/kafka-consumer-groups.sh / scripts/kafka-verifiable-consumer.sh end-to-end runs. The gap is explicit in source — grep MEMBER_ID_REQUIRED in crates/kaas-coordinator/src/group.rs.

KIP-447 — EOS v2

Status: implemented — see the KIP index.

What the KIP changes in Apache Kafka

KIP-447 (Kafka 2.5) made exactly-once consume-process-produce scale. EOS v1 required one producer per input partition, because zombie fencing hung entirely off the transactional.id. v2 lets one producer serve a whole process: sendOffsetsToTransaction takes the consumer group's metadata, TxnOffsetCommit v3 carries member.id / generation.id / group.instance.id, and the group coordinator fences offset commits from producers whose consumer generation is stale. Offsets committed inside a transaction stay invisible until the transaction commits.

How kaas implements it

The offsets-in-transactions flow rides the KIP-98 machinery; the state walkthrough is in Transactions & idempotence.

• AddOffsetsToTxn (key 25, crates/kaas-broker/src/handlers/add_offsets_to_txn.rs) records the consumer group on the transaction's entry in the txn state store, so EndTxn later knows which groups to settle.
• TxnOffsetCommit (key 28, v0–v3, crates/kaas-broker/src/handlers/txn_offset_commit.rs) runs on the group coordinator (non-coordinators return NOT_COORDINATOR) and stages the offsets in the pending layer of crates/kaas-coordinator/src/offset_store.rs, keyed on (group_id, producer_id). Staged offsets are not visible to OffsetFetch (TxnOffsetCommit).
• EndTxn (key 26) drives the offset hook for every recorded group — it fires from the complete transition, only once every COMMIT/ABORT marker is durable: commit materialises the pending offsets into the group's committed set (commit_pending); abort discards them (discard_pending). The timeout reaper stops at PrepareAbort, and the marker-reconcile pass fires the same hook when it completes the reaped transaction, so it can't leak half-committed offsets.

The pending layer is memory-only by design — an unfinished transaction's staged offsets reset on broker restart, which the source frames as Apache's "in-flight offsets aren't recovered" contract. Note the shape of that trade honestly: Apache stages pending commits in the replicated __consumer_offsets log, so its window is narrower; in kaas a group coordinator restart between TxnOffsetCommit and EndTxn(commit) drops the staged offsets, and the application re-processes from the last committed offset (at-least-once across that crash, never offset-without-data).

The KIP's headline fencing is not enforced: TxnOffsetCommit v3's generation_id / member_id / group_instance_id are decoded (crates/kaas-codec/src/api/txn_offset_commit.rs) but the handler never checks them against the group's live generation, so a zombie producer is fenced by its producer epoch (KIP-360) rather than by consumer-group generation. The wire surface a v2 client needs is complete; the extra fencing layer is not.

How it's verified

bins/kaas/tests/eos_v2.rs drives the full v2 round trip (InitProducerId → AddPartitionsToTxn → transactional Produce → AddOffsetsToTxn → TxnOffsetCommit → EndTxn) over real TCP: eos_commit_path_records_visible_to_read_committed and eos_abort_path_populates_aborted_transactions. The staging contract is pinned by pending_invisible_to_fetch_until_commit_pending and discard_pending_drops_unmaterialised_offsets in offset_store.rs, and by happy_path_stages_pending_offsets / no_manager_returns_not_coordinator in the handler. Shell suite: scripts/kafka-txn-coordinator.sh probes the coordinator wire surface; scripts/kafka-transactions.sh covers the admin tool (the Kafka 4.x CLI dropped --transactional-id from kafka-verifiable-producer, so the producer-side flow lives in the integration tests above).

KIP-482 — Flexible versions and tagged fields

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

From Kafka 2.4, each API declares a cutover version at which its encoding becomes "flexible": strings, bytes, and arrays switch to compact (unsigned-varint) length prefixes, request headers gain a v2 shape and response headers a v1 shape, and every structure ends in a tagged-field block — an extensible (tag, size, value) section that lets either side attach optional data without bumping the API version. Unknown tags must be skipped, which is what makes the scheme forward-compatible.

How kaas implements it

The whole mechanism lives in crates/kaas-codec:

• crates/kaas-codec/src/tagged.rs — the envelope: uvarint(num_fields) || (uvarint(tag), uvarint(len), bytes)*. read consumes and discards every tag (Apache's documented forward-compat contract); read_into surfaces them for the handlers that inspect specific tags; write_empty emits the single-byte uvarint(0) that is by far the most common case on the wire.
• crates/kaas-codec/src/primitives.rs — compact strings, bytes, and arrays with the KIP's count + 1 length encoding (read_compact_string, read_compact_array_len, and friends).
• crates/kaas-codec/src/headers.rs — the HeaderVersion V0/V1/V2 split: flexible requests use v2 headers (client id + tagged block), flexible responses use v1 (correlation id + tagged block).
• crates/kaas-codec/src/api/registry.rs — each ApiSpec carries min_flexible: Option<i16>; is_flexible(version) and the per-API header-version functions mirror Apache's ApiKeys.requestHeaderVersion(apiKey, apiVersion) table. The "Flexible" column of the API matrix is generated from this table, so the docs cannot disagree with what ApiVersions advertises.

One honest asymmetry: kaas reads tagged fields everywhere but never writes a non-empty block — every response ends in write_empty. That is also what Apache 3.7 does on most paths; kaas simply has no optional server-side tags to attach yet. See Wire protocol & framing for how this sits alongside the byte-opacity contract.

How it's verified

• crates/kaas-codec/src/tagged.rs unit tests: empty_block_roundtrips, multi_field_roundtrip, unknown_tags_discarded_by_read.
• crates/kaas-codec/src/headers.rs: request_v2_flexible_tagged_block, response_v1_flexible_with_tag_block (asserts the exact 5-byte shape).
• crates/kaas-codec/src/api/registry.rs: flex_predicate pins the flexible-from version for ApiVersions itself.
• Every per-API codec module round-trips its flexible versions; the leave_group.rs v5 Java-client fixture is a real regression pin — see KIP-800 for the bug it caught.
• scripts/kafka-broker-api-versions.sh exercises the ApiVersions response (itself a flexible API) against a live broker.

KIP-516 — topic identifiers

Status: partial — see the KIP index. Earlier planning documents listed this KIP as implemented; the 2026-07-19 source sweep corrected it.

What's missing

Wire propagation. The plumbing now reaches the broker: the production topic watch (run_topic_watch in crates/kaas-k8s/src/kube_watchers.rs) delivers status.topicId into the broker's topic registry (TopicRegistry::set_topic_id), where it backs the stale-directory identity gate (gh #241). But the ID is deliberately kept in a side map rather than filled into TopicMeta.topic_id — flipping Metadata from nil to real topic IDs is this KIP's remaining call — so Metadata v10+ still serves nil topic IDs for every topic, and clients fall back to name-based lookups. CreateTopics v7+ returning the UUID in its response is likewise open.

What the KIP changes in Apache Kafka

KIP-516 gives every topic an immutable UUID alongside its name, closing the delete-and-recreate ambiguity: a fetcher or metadata cache holding the old topic's ID can tell it apart from the re-created namesake. Requests and responses grew topic-ID fields from Metadata v10 / Fetch v13 onward.

What kaas has

The identity half, operator-side and durable:

• The KafkaTopic controller (crates/kaas-operator-controllers/src/kafkatopic_controller.rs) mints a cryptographically random v4 UUID into Status.TopicID on first reconcile and never rotates it — a re-created topic gets a distinct ID, exactly Apache's contract. kubectl get kafkatopic -o yaml shows it.
• The broker's topic registry (crates/kaas-broker/src/topic_registry.rs) carries a 16-byte topic_id per topic and the Metadata handler encodes it — today always the all-zero value, which the protocol defines as the "no topic ID" sentinel, so clients degrade gracefully rather than misbehave.

Because kaas serves Fetch v4–v12 (matrix) — below the v13 topic-ID cutover — the Fetch path never needs an ID, which is why the gap is invisible to normal produce/consume traffic.

How the partial state is verified

Operator-side minting and non-rotation are covered by the kafkatopic controller's reconcile tests; the wire-side sentinel is pinned by the Metadata handler's unit tests (crates/kaas-broker/src/handlers/metadata.rs). The gap itself is explicit: every TopicMeta insertion in bins/kaas/src/main.rs writes topic_id: [0u8; 16].

KIP-546 — Client-quota admin APIs

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 2.6 added DescribeClientQuotas (key 48) and AlterClientQuotas (key 49): a unified admin surface over quota entities — user, client-id, ip, and their <default> variants — replacing direct ZooKeeper writes from kafka-configs.sh. Filters support exact-match, any, and default-entity components.

How kaas implements it

Both handlers (crates/kaas-broker/src/handlers/describe_client_quotas.rs and alter_client_quotas.rs, keys 48/49, v0–v1, flexible from v1) wrap the QuotaEnforcer in crates/kaas-auth/src/quota.rs — the same debt-carrying token bucket that throttles Produce/Fetch (KIP-13).

• Entity model: the user axis only. client-id / ip components return an empty result on describe and INVALID_REQUEST (42) on alter. match_type=DEFAULT (--entity-default) is unsupported — kaas users are CR-instantiated, there is no <default> entity.
• Describe resolves override > store > none: a runtime override wins, otherwise the store-backed value from credentials.json (materialised by the operator from KafkaUser.spec.quotas). Exact-match and list-all-users both work.
• Alter merges ops per key onto the user's current quotas (Apache semantics: SET replaces one key, remove drops one key, unspecified keys survive) and installs the result via set_user_quota, live-updating any existing token bucket. Known keys: producer_byte_rate, consumer_byte_rate, request_percentage; anything else answers INVALID_CONFIG (40). validate_only skips the install.
• Alterations are runtime-only and per-broker. The override lives in the enforcer's in-memory map: it does not survive a broker restart and is not broadcast to peer brokers. The durable, cluster-wide path is KafkaUser.spec.quotas (producerMaxByteRatePerBroker / consumerMaxByteRatePerBroker — named for the per-broker semantics, see Listeners & auth).
• request_percentage round-trips but is not enforced — only the byte-rate keys are wired into the bucket (apply_quotas reads just producer/consumer rates). Authorization follows Apache's mapping: describe → DescribeConfigs, alter → AlterConfigs, both on the cluster resource.

How it's verified

• crates/kaas-auth/src/quota.rs unit tests: set_user_quota_live_updates_existing_bucket, describe_user_quota_resolution_order (override > store precedence).
• crates/kaas-codec/src/api/alter_client_quotas.rs and describe_client_quotas.rs round-trip tests cover both versions.
• scripts/kafka-configs.sh scenarios 6–10 run quota CRUD plus a live 10 MB/s throttle probe through the real kafka-configs.sh tool. (Their labels still say XFAIL from before the quota engine landed — stale labels, the scenarios exercise the implemented path.)

See also the DescribeClientQuotas and AlterClientQuotas entries.

KIP-554 — broker-side SCRAM credential admin API

Status: partial — see the KIP index.

What's missing

• SCRAM-SHA-256 — kaas authenticates with SCRAM-SHA-512 only, so SHA-256 upsertions answer UNSUPPORTED_SASL_MECHANISM rather than storing a credential no listener could verify.
• Wire-side credential deletion — deletions answer UNSUPPORTED_VERSION with a message pointing at the KafkaUser CR. The credential's lifecycle belongs to the CR: the operator would re-materialise anything the broker removed on its next reconcile, and a deletion that silently comes back is worse than a refusal. Delete the CR (or change its authentication.type) to remove a credential.

What the KIP changes in Apache Kafka

KIP-554 moved SCRAM credential management from ZooKeeper writes to a proper broker API: admins upsert and inspect salted SCRAM credentials via the AdminClient, with the broker storing the derived keys (never the password).

What kaas has

Credential management is operator-side, via the KafkaUser CR — the Kubernetes-native equivalent of what the KIP provides over the wire:

• The KafkaUser reconciler (crates/kaas-operator-controllers/src/kafkauser_controller.rs) derives salted SCRAM entries into /data/__cluster/credentials.json, which brokers hot-reload.
• The rotation path: a KafkaUser can reference pre-derived SCRAM credentials, which pass through to credentials.json verbatim — enabling zero-downtime rotation without the operator ever seeing the plaintext password.
• Auto-generated passwords land in a <user>-kafka-credentials Secret owned by the CR (Kubernetes GC cleans it up with the user).

And since v0.2.45 the wire path is served too: DescribeUserScramCredentials (key 50) answers mechanism + iteration counts from the same hot-reloaded credential store the SCRAM authenticator verifies against, and AlterUserScramCredentials (key 51) derives the RFC 5802 stored/server keys from the wire's salted material and patches them into KafkaUser.spec.authentication.scram — the same CR-write pattern CreateAcls uses, so the operator remains the only writer of credentials.json and kafka-configs.sh --alter --add-config 'SCRAM-SHA-512=…' rotates a credential cluster-wide within one reconcile + reload cycle.

See Listeners, authentication, authorization for how credentials are consumed.

How the partial state is verified

KafkaUser reconcile tests cover derivation, passthrough rotation, and Secret ownership (crates/kaas-operator-controllers/src/kafkauser_controller.rs); bins/kaas/tests/auth_smoke.rs proves the derived credentials authenticate over the wire. The wire keys are pinned by the registry test in crates/kaas-codec/src/api/registry.rs; the alter handler's key-derivation and rejection semantics by crates/kaas-broker/src/handlers/alter_user_scram_credentials.rs.

KIP-700 — DescribeCluster

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 2.8 added the DescribeCluster API (key 60) to give AdminClient.describeCluster() a request of its own. Until then it asked Metadata for the cluster id, the controller, and the broker list, and paid for every topic in the cluster to find out — on a large cluster that response is megabytes of per-partition rows the caller throws away. The new key answers exactly the three facts, plus an optional ClusterAuthorizedOperations bitfield saying what the calling principal may do at cluster scope.

Kafka 3.7 serves v0–v1 (4.0 adds v2). v1 is KIP-919: the request names the endpoint type it wants (1 = brokers, 2 = controllers) so a client bootstrapped against a KRaft controller quorum can tell it apart from a broker, and two error codes become valid at the top level — MISMATCHED_ENDPOINT_TYPE (114) and UNSUPPORTED_ENDPOINT_TYPE (115).

How kaas implements it

• The codec (crates/kaas-codec/src/api/describe_cluster.rs) serves v0–v2. This is the one key that is flexible from v0 — it postdates KIP-482 entirely, so there is no legacy encoding branch and no version below which the tagged-field blocks disappear.
• v2 (KIP-1073, Kafka 4.0) is served as a deliberate parity-target exception: it carries IncludeFencedBrokers / IsFenced, and kaas grew a real fenced state to report — see broker fencing. Clients that request fenced brokers unconditionally also need a version where the field is legal, or they fail to encode their own request.
• The handler (crates/kaas-broker/src/handlers/describe_cluster.rs) shares its broker catalog with Metadata (crates/kaas-broker/src/listener_advert.rs): same per-listener port, same peer FQDNs, same controller derived from the applied assignment.json. Two discovery APIs that disagreed about which port to hand back would loop an authed client on SASL retry, which is why the rule lives in one place rather than two.
• ClusterAuthorizedOperations is computed only when the client asks for it, and only after Describe on the cluster resource passes. The three answers are distinct on the wire: -2147483648 ("didn't ask"), 0 ("asked, denied"), and the bitfield otherwise.
• kaas serves broker endpoints only. Controller election runs on a Kubernetes Lease, not a KRaft quorum (Non-goals), so there is no controller endpoint to describe: a v1 request for one answers MISMATCHED_ENDPOINT_TYPE, anything unrecognised UNSUPPORTED_ENDPOINT_TYPE.
• Two fields kaas cannot answer the way Apache does: Rack is always null (no rack awareness — there are no replicas to place), and the bitfield never sets ClusterAction or IdempotentWrite, because kaas's authorizer models neither operation. Both are detailed on the DescribeCluster reference.
• IsFenced comes from EndpointSlice readiness rather than a KRaft registration + session timeout, so a booting broker reads as fenced until takeover completes — the same answer Apache gives for a registered broker that hasn't caught up.

How it's verified

• Codec round trips per version in crates/kaas-codec/src/api/describe_cluster.rs, including the v1-gated endpoint-type field and the i32::MIN "not requested" sentinel.
• Handler tests in crates/kaas-broker/src/handlers/describe_cluster.rs cover the per-listener broker catalog, both endpoint-type errors, and all three authorized-operations answers.
• The DescribeCluster leg of bins/kaas/tests/smoke.rs drives a real v1 request through the dispatcher over TCP — the path where a wrong header entry in the registry would misparse every request, since there is no non-flexible version to fall back to.
• scripts/kafka-cluster.sh runs the real kafka-cluster.sh cluster-id against a live broker, which is an AdminClient.describeCluster() call.

KIP-800 — Join/leave reason strings

Status: implemented — back to the KIP index.

What the KIP changes in Apache Kafka

Kafka 3.1 added a nullable reason string to JoinGroup requests (v8) and to each member entry in LeaveGroup requests (v5), so clients can tell the broker why they are rejoining or leaving ("the consumer is being closed", rebalance triggered by metadata change, …). It is a request-side-only, diagnostics-oriented field: Apache logs it to make rebalance storms explainable.

How kaas implements it

• crates/kaas-codec/src/api/join_group.rs decodes the nullable reason on v8+ requests (kaas serves JoinGroup v2–v9, flexible from v6); crates/kaas-codec/src/api/leave_group.rs decodes the per-member reason on v5+ (kaas serves v0–v5, flexible from v4). Both are version-gated exactly per the Apache 3.7 schema — v7/v4 wire bytes must not contain the field, and the encoders enforce that.
• This field earned its regression fixture the hard way: the pre-fix v5 LeaveGroup decoder read the reason's length byte as the tagged-field count, which broke every modern Java client's clean shutdown (the consumer sends "the consumer is being closed" on close). The fixture test in leave_group.rs pins the literal bytes a Java 3.7 client sends.
• The broker decodes the reason and then discards it. Neither handler (crates/kaas-broker/src/handlers/join_group.rs, leave_group.rs) nor the group coordinator (crates/kaas-coordinator/src/group.rs) logs, stores, or acts on it — there is no occurrence of the field beyond the codec layer. Wire parity is what "implemented" means here; surfacing the reason in broker logs/metrics, which is the KIP's operational payoff, is honest follow-up work.

Since clients enable this unconditionally at v8/v5, the load-bearing part is accepting the field without corrupting the rest of the request — which is exactly where the old bug lived.

How it's verified

• crates/kaas-codec/src/api/leave_group.rs: v5_java_client_fixture_with_reason (captured Java-client bytes, reason "bye", null group_instance_id) and v4_omits_reason_on_the_wire (the field must not leak into v4 encodings).
• crates/kaas-codec/src/api/join_group.rs: v8_adds_reason round-trips the field at v8+.
• scripts/kafka-consumer-groups.sh and scripts/kafka-console-consumer.sh run modern Java clients through full join → consume → clean-close cycles, which send reason strings on every leave.

See also the JoinGroup and LeaveGroup entries in the per-API reference.

Non-goals

No KRaft, no replication/ISR, no internal topics, and no tiered storage (yet) — every omission with its rationale, not silence.

kaas keeps Kafka's wire contract but replaces Kafka's distributed-systems machinery with Kubernetes primitives and a shared filesystem. That trade shows up as three deliberate substitutions, plus a short list of deferred features. This page is the ledger. Every entry follows the same shape: what Apache does → what kaas does instead → why → what would change our mind (where there's an honest answer). If a parity task ever implicitly requires one of these, the right move is to flag it, not to quietly grow the machinery.

KRaft / metadata quorum

Apache: a Raft-based controller quorum (KRaft) replaced ZooKeeper as the metadata store and controller-election mechanism.

kaas: a Kubernetes Lease (kaas-controller) elects the controller; leaseTransitions is the monotonic epoch; the Kubernetes API server is the metadata store (details).

Why: (a) the API server already is a replicated, consistent metadata store — reimplementing one in-process duplicates that role for no operational gain; (b) holderIdentity + leaseTransitions encode "current controller + monotonic epoch" exactly as needed; (c) Raft brings a peer gossip protocol and a large code surface the rest of the broker has no use for.

What would change our mind: running kaas outside Kubernetes. That's not on the roadmap — Kubernetes-native is the premise of the project.

Replication / ISR

Apache: each partition is replicated across N brokers with an in-sync replica set, leader election, and fencing RPCs.

kaas: single-writer-per-partition on shared ReadWriteMany storage; the substrate provides durability and the epoch-prefixed segment filenames provide split-brain safety by construction (details).

Why: (a) ISR replication is most of what makes multi-broker Kafka operationally hard — preferred-leader election, under-replicated alerts, controlled-shutdown choreography; kaas trades that for the NFS server's (already-solved) redundancy; (b) modern NFS/SAN substrates replicate at the storage layer — replicating again in-broker doubles the write cost for nothing; (c) a stale ex-leader physically cannot corrupt a new leader's log, because it writes to segment files named with a dead epoch.

Consequence to be honest about: broker loss makes its partitions unavailable until the controller reassigns them (seconds), and storage loss is data loss — durability is exactly as good as the substrate. That's the contract; see Storage substrate requirements.

Internal topics

Apache: coordinator state lives in compacted internal topics — consumer offsets in __consumer_offsets, transaction-coordinator state in __transaction_state — replayed on coordinator failover, with the partition leadership of those topics deciding which broker coordinates which group or transaction.

kaas: plain JSON files on the shared volume. Transaction state is slot-sharded (txn_state/slot-N.json, 50 slots — matching Apache's default of 50 __transaction_state partitions; details); consumer offsets are one file per group (details). The coordinator routing Apache derives from internal-topic partition leadership, kaas derives from a hash over the broker set.

Why: (a) without replication, an internal-topic-as-log buys nothing over a file; (b) NFS close-to-open consistency means the file is the materialized state — failover is "open the file", no replay; (c) debuggability: a stuck transaction is cat slot-N.json.

Tiered storage / S3 (KIP-405) — deferred, not refused

Apache 3.6+: remote log storage with a local hot tier.

kaas: no remote tier. The tiered-storage-only API surfaces (EARLIEST_LOCAL_TIMESTAMP, EARLIEST_PENDING_UPLOAD_OFFSET in ListOffsets) are deliberately skipped — clients only send them when configured for remote tiers.

Why: the NFS substrate is already bulk-priced storage, and KIP-405 roughly doubles the cleanup/retention state machine.

What would change our mind: this is the one entry that's genuinely deferred — an S3 backend is intended later, and the storage engine's byte-opaque segments are designed not to preclude it.

Fetch sessions (KIP-227) — stateless by contract

kaas answers every Fetch with SessionID=0 — Apache's documented signal for "broker doesn't support sessions" — so clients send full fetch state per request. Echoing the client's session ID without maintaining session state was an actual bug (clients sent incremental deltas against state kaas didn't have and silently dropped partitions); SessionID=0 is the correct unsupported-marker, not a shortcut. The extra per-request CPU is fine at kaas's scale; session caching is a future optimisation, not a correctness gap.

The rest of the tracked non-goal KIPs

• KIP-48 (delegation tokens) — token auth targets large multi-tenant clusters brokering their own trust; kaas deployments authenticate via SCRAM or mTLS backed by Kubernetes-managed secrets, or — for token-shaped needs — OAuth bearer tokens on an OAUTHBEARER listener.
• KIP-664 (Describe/ListTransactions) — admin tooling over coordinator state; a follow-up. Until then the slot files on the volume are directly inspectable, which covers the debugging use case the KIP exists for.
• KIP-714 (client metrics push) — out of scope for the preview line; kaas's own observability is OTLP-push (Observability).
• KIP-848 / KIP-1071 (next-gen rebalance) — post-3.7 protocols; out of the 3.7 parity target by definition.
• KIP-932 (share groups) — Kafka 4.0+; the shell-tool suite marks the share-group tools as skipped with an explicit reason (Verification story).

Inter-broker surface

The Apache inter-broker/controller keys (LeaderAndIsr, StopReplica, UpdateMetadata, ControlledShutdown, the KRaft quorum and Envelope family) don't exist in kaas at all — there is no replication protocol to drive and no quorum to speak. kaas brokers coordinate through exactly two channels: the heartbeat gRPC stream and files on the shared volume (Controller).

Verification story

How parity claims are backed: the Apache shell-tool suite, the integration tests, the parity project board, and the bench methodology.

A compatibility claim is only as good as the thing that would catch it being wrong — which is also why this book can afford to be blunt about gaps: the machinery that backs every "implemented" would surface a regression just as loudly. kaas layers four:

1. The Apache shell-tool suite

The strongest evidence that unmodified Kafka tooling works against kaas is running unmodified Kafka tooling against kaas. 41 per-tool scripts (scripts/kafka-*.sh, at the repo root) run the actual Apache Kafka distribution's shell tools (kafka-topics, kafka-console-producer, kafka-consumer-groups, kafka-acls, kafka-{producer,consumer}-perf-test, kafka-verifiable-{producer,consumer}, …) against a live kaas cluster — the same binaries an operator would point at Apache, unchanged. Every script honours the same BOOTSTRAP / KAFKA_BIN overrides; defaults target the in-cluster Service DNS and /opt/kafka/bin.

Two honesty conventions:

• Skips are explicit, not silent. Scripts covering features that are non-goals or post-3.7 (KRaft tools, share groups, …) print a one-line reason and exit 77 — discoverable, and never pretending to test something that can't work.
• The baseline is recorded and diffable. scripts/.parity-baseline.txt pins the expected per-script result. Current baseline (captured 2026-07-19 against v0.2.4-preview, production 3-broker shape on NFS-RWX): 21 PASS / 20 SKIP / 0 FAIL. Every SKIP maps to a documented non-goal or a post-3.7 feature. Reruns assert against the baseline, so a PASS→FAIL downgrade is a regression, not an anecdote.

2. In-process integration tests

Run on every CI push (cargo test --workspace --all-features), against a broker started inside the test process:

• wire-level round trips (produce → fetch → metadata);
• the SASL/SCRAM handshake plus ACL enforcement;
• the byte-opacity tripwires, asserted to read zero after real traffic;
• multi-broker bring-up: assignment, takeover, coordinator routing;
• the full KIP-447 consume-process-produce-commit round trip (EOS v2);
• controller failover and the stale-epoch fence.

Plus per-crate unit tests — including the codec's fixture tests, which pin encode/decode byte-identity against captures from Apache Kafka 3.7.

3. The parity project board

The kaas-migration-parity GitHub project tracks the feature matrix item by item — the working surface where "what does 3.7 do here?" questions get resolved before they become code. When a feature is ambiguous, the default is match Apache Kafka 3.7, never invent kaas-specific semantics.

4. Docs that can't rot

Two CI gates keep this book honest (cargo xtask check-docs-drift):

• The API support matrix is generated from the same codec registry that builds the ApiVersions response; CI fails if the committed page drifts from the wire surface.
• Every crates/… / bins/… source path cited anywhere in the book is checked against the tree; a refactor that moves a file fails CI until the citation is fixed.

Performance verification

Benchmarks are treated with the same suspicion as compatibility claims: multi-run (5×) averaging with outlier exclusion, NFS RPC + network-rate snapshots as a NAS-liveness probe, and recorded reports under docs/perf-results/. Current standing and methodology live in Performance vs Strimzi.

Implementation notes (for contributors)

• Shared shell-suite helpers (BOOTSTRAP, KAFKA_BIN, skip) live in scripts/_common.sh, sourced by every kafka-*.sh script.
• The integration tests above, in list order: bins/kaas/tests/smoke.rs, bins/kaas/tests/auth_smoke.rs, bins/kaas/tests/oauth_smoke.rs, bins/kaas/tests/byte_opacity.rs, bins/kaas/tests/cluster_bringup.rs + cluster_smoke.rs, bins/kaas/tests/eos_v2.rs, bins/kaas/tests/retention.rs, and crates/kaas-controller/tests/controller_failover.rs + stale_controller_race.rs.

Workspace layout & crate dependency graph

Twelve library crates, two binaries, and an xtask runner — who depends on whom, and why the layering looks the way it does.

The workspace root carries Cargo.toml, rust-toolchain.toml (pinned toolchain, auto-installed by rustup), proto/ (the heartbeat gRPC schema), deploy/ (Helm chart + generated CRDs), scripts/ (the Apache shell-tool parity suite), and docs/ (this book). protoc is vendored via the broker's build script — a fresh checkout needs nothing beyond rustup.

The layering rule that shapes the graph: kaas-codec knows nothing about storage, storage knows nothing about Kubernetes, and nothing below kaas-broker knows about request handling. Wire bytes stay byte-opaque from codec through storage (the invariant Part II's wire-protocol chapter documents), and the Kubernetes-facing crates (kaas-k8s, kaas-operator-*) sit off the hot path entirely (runtime independence).

Each crate has its own short chapter in this part — what it owns, the invariants callers must hold, and where to start reading. If you've arrived from Parts I–II you already know what the system does and what it speaks on the wire; these chapters assume that and point back to the architecture pages instead of re-explaining semantics. The chapter order is a deliberate reading order, following the wire inward: kaas-codec → kaas-protocol → kaas-storage → kaas-coordinator → kaas-broker → kaas-controller, then the supporting crates (kaas-auth, kaas-k8s, kaas-observability), the operator pair (kaas-operator-api, kaas-operator-controllers), kaas-test-harness, and finally the two binaries that plug every seam together (kaas and kaas-operator).

Crate dependency graph

An arrow reads "depends on". Verified against each crate's Cargo.toml [dependencies] (runtime deps only, dev-dependencies excluded).

flowchart LR
    subgraph binaries
        kaas_bin["bins/kaas<br/>broker entrypoint"]
        op_bin["bins/kaas-operator<br/>operator entrypoint"]
    end

    broker["kaas-broker<br/>broker glue, Coordinator,<br/>takeover, handlers/*"]
    protocol["kaas-protocol<br/>dispatch, listener bring-up"]
    codec["kaas-codec<br/>wire frames, per-API codecs"]
    auth["kaas-auth<br/>SCRAM, mTLS, ACLs, quotas"]
    storage["kaas-storage<br/>engine, segments, idempotence"]
    coordinator["kaas-coordinator<br/>consumer groups + txns"]
    controller["kaas-controller<br/>election, balancer,<br/>assignment writer"]
    k8s["kaas-k8s<br/>endpoints, identity,<br/>topic watcher"]
    opapi["kaas-operator-api<br/>CRD types (kube-derive)"]
    opctl["kaas-operator-controllers<br/>reconcilers"]

    kaas_bin --> broker
    kaas_bin --> controller
    kaas_bin --> k8s
    kaas_bin --> protocol
    kaas_bin --> codec
    kaas_bin --> auth
    kaas_bin --> storage
    kaas_bin --> coordinator

    op_bin --> opctl
    op_bin --> opapi
    op_bin --> controller
    op_bin --> storage

    broker --> protocol
    broker --> codec
    broker --> auth
    broker --> storage
    broker --> coordinator
    broker --> opapi

    protocol --> codec
    protocol --> auth
    protocol --> storage

    controller --> broker
    controller --> coordinator
    controller --> storage

    k8s --> broker
    k8s --> coordinator
    k8s --> opapi

    opctl --> opapi
    opctl --> storage

Two crates are left off the diagram to keep it readable:

• kaas-observability is depended on by every crate above except kaas-codec and kaas-operator-api (and by both bins); its own single dependency is kaas-codec, for the byte-opacity tripwire counters.
• kaas-test-harness depends on nothing in the workspace and nothing depends on it — it is still an empty placeholder crate.

This graph is hand-maintained (checked against Cargo.toml on 2026-08-11). Auto-generating it from cargo metadata is a possible future gen-api-matrix-style xtask.

kaas-codec

The Kafka wire-protocol codec: frames, primitives, CRC32C, KIP-482 tagged fields, and per-API request/response types with the ApiSpec registry.

The wire boundary of the whole system. Everything here is pinned by fixture tests: every registered API key/version encodes and decodes byte-identically against captures from Apache Kafka 3.7.

Module map: frame.rs (length-prefixed framing, streaming FrameReader), primitives.rs + tagged.rs (Kafka primitive types and KIP-482 compact/tagged encoding), headers.rs (per-API header-version resolution), crc.rs (CRC32C batch verification), errors.rs, one module per API under src/api/, and src/api/registry.rs — the ApiSpec table.

The invariant callers must hold: RecordBatch payloads are byte-opaque. There is no Record struct in this crate, and none may be added — batch bytes travel as Option<bytes::Bytes>, and the only code reading past the fixed v2 header is CRC verification and the batch-header walker (recordbatch_count.rs). Any future violation must bump the counters in tripwires.rs, which tests assert are zero. The full rationale is in Wire protocol & framing.

The registry is load-bearing: src/api/registry.rs::ALL (40 entries, count-asserted by a unit test) drives the ApiVersions response, the header-version lookup, and the book's generated API matrix — adding an API without a registry row is structurally impossible to ship quietly.

Where the boundary sits: kaas-codec knows nothing about storage, Kubernetes, or request handling — its only workspace consumers are kaas-protocol (framing/dispatch) and the handler layer above.

Start reading at src/api/registry.rs, then one small API module (src/api/api_versions.rs) end to end.

kaas-protocol

Multi-listener TCP/TLS bring-up, the per-listener pre-auth dispatch gate, framing, and per-connection state.

The layer between the codec and the handlers: it owns sockets, connection lifecycle, and the decision of whether a request may be dispatched at all.

Module map: server.rs (multi-listener TCP/TLS accept loops — TCP_NODELAY on accept, mTLS principal-mapper wiring), frame.rs (Connection<S>: async stream + frame reader, request headers parsed via the codec registry, responses written with the right header version), dispatch.rs (the API-key router), connstate.rs (per-connection mutable state: listener name, SASL progress, principal).

The pre-auth gate is the crate's most consequential logic: on an authenticated listener, dispatch.rs refuses every API except SaslHandshake (17), ApiVersions (18), and SaslAuthenticate (36) until the connection's SASL exchange completes. The same gate enforces the KIP-368 re-authentication deadline — past the session lifetime it answers every request with error 58 until the client re-authenticates. Listener identity travels on ConnState as a free-form name matching the chart's listeners[] entries; the auth engine is selected per listener (architecture chapter).

Error contract: a request that fails dispatch-level checks gets a proper error-code response with the client's correlation ID — connections are not dropped for policy failures.

Where the boundary sits: kaas-protocol depends on kaas-codec and kaas-auth; it knows nothing about storage or Kubernetes. Handlers are registered into the dispatcher by bins/kaas — the crate defines the Handler seam, not the handlers.

Start reading at dispatch.rs, then server.rs for the listener bring-up.

kaas-storage

The disk storage engine: segments, manifest, cleaner, idempotent-producer state, the aborted-txn index, and the in-memory dev-mode engine.

Read Storage engine hot path and File-handle ownership before the engine code — the group-commit, manifest-lag, and single-FD semantics are deliberate and easy to misread as bugs from inside a single file.

Module map: engine.rs (the StorageEngine trait + DiskStorageEngine), disk.rs (engine-level orchestration), partition.rs (the partition core: mutex-guarded write path, ArcSwap read snapshot, per-partition committer task), segment.rs (epoch-prefixed segment files + sparse index), manifest.rs (tmp+fsync+rename state file), cleaner.rs (delete-policy retention only — compaction is still unimplemented, gh #158) + topicconfig.rs (retention/segment/compaction knobs from .config.json), recovery_checkpoint.rs (the bounded-recovery-scan checkpoint, gh #230), topic_identity.rs (the .topic-id.json incarnation stamp, gh #219), idempotence.rs + producer_snapshot.rs (per-PID dedupe rings and their persistence), txn_index.rs (aborted-transaction ranges for read-committed Fetch), memory.rs (dev-mode in-memory engine), atomic_write.rs, errors.rs, fs.rs (the filesystem seam that lets tests fault-inject). The crate also carries the multi-log-dir/volume-pool plumbing (parse_log_dirs_json, the PlacementResolver seam — gh #221/#224).

Invariants callers must hold:

• Single writer per partition is enforced by coordinator ownership + epoch-prefixed filenames, not by the filesystem — calling append on a partition you don't own is a protocol violation upstream, not something the engine can fully defend against.
• Batch bytes are opaque. The engine peeks fixed-size headers (idempotence info, offsets) and rewrites the base offset in place; nothing may decode records.
• The manifest lags by design — recovery reconciles from the log on open; treating manifest.json as current truth mid-flight is a bug.
• FDs belong to the leader: take_over opens handles, relinquish closes them; holding handles elsewhere reintroduces the NFS silly-rename problem (gh #76).

Start reading at partition.rs::append, following one batch from classification to the committer's sync_all.

kaas-coordinator

The consumer-group coordinator and offset store, plus the transaction coordinator's state store, marker queue, and fence log.

Two coordinators share this crate because they share a shape: in-memory protocol state + durable files on the shared volume + an ownership seam that decides which broker answers for which key.

Consumer-group side: group.rs (the join/sync/heartbeat/leave state machine, generation tracking, static-membership handling), manager.rs (group lookup/creation, ownership-filtered list/describe, offset deletion — including Manager::purge_topic_offsets, which drops a deleted topic's committed offsets from every owned group, gh #240), offset_store.rs (per-group JSON files under <KAAS_CLUSTER_DIR>/__consumer_offsets/<group>.json — moved under the cluster dir by gh #223 — plus the pending layer that stages transactional offsets keyed by (groupID, PID) until EndTxn commits them).

Transaction side: txn_state.rs (per-transactional.id entries, slot-sharded across txn_state/slot-N.json; all transitions are atomic slot-file rewrites), marker_queue.rs (cross-broker COMMIT/ABORT marker dispatch as files under marker_queue/to-<broker>/), fence_log.rs (cross-broker producer-epoch fence broadcast), plus the atomic_write.rs and errors.rs helpers. The architecture chapters on transactions and consumer groups explain why these are files and not RPCs.

The invariant callers must hold — the assignment-source indirection: group ownership comes from a GroupAssignmentSource, txn ownership from a txn-assignment source. In production both are backed by the broker Coordinator (hash-fallthrough over assignment.json); in single-broker tests they're local always-true stubs. Code in this crate must route every "do I own this?" question through the seam — hard-coding locality reintroduces the gh #92 chicken-and-egg that took two releases to unwind.

Start reading at manager.rs, then group.rs for the rebalance state machine, then txn_state.rs.

kaas-broker

Broker glue: the on-broker Coordinator, takeover drivers, topic registry, CR write paths, and one handler module per Kafka API.

The largest crate, but structurally simple: everything either answers a request (handlers/) or maintains the state requests read from (everything else).

State side: broker.rs (the Broker — the narrow shape every handler reads), coordinator.rs (the on-broker Coordinator: watches assignment.json, answers every ownership question with hash-fallthrough group/txn routing), takeover.rs + group_takeover.rs (drivers that diff assignments into storage-engine take-over/relinquish and group load/evict — including the gh #89 orphan sweep), group_hash.rs (deterministic coordinator routing), topic_registry.rs, self_fence.rs (stops acking writes when heartbeats stall), heartbeat_client.rs, fence_watcher.rs + marker_watcher.rs (shared-volume pollers applying peer fences and txn markers), producer_id.rs (persisted per-broker PID blocks, gh #219), txn_markers.rs (shared COMMIT/ABORT marker dispatch + the prepared-txn reconcile, gh #225), assignment.rs (the assignment.json schema), control_batch.rs (COMMIT/ABORT control-batch encoder), listener_advert.rs (the advertised-endpoint rule + broker catalog shared by Metadata and DescribeCluster), argocd.rs (ArgoCD coexistence annotations on broker-minted CRs), topic_config_defaults.rs (the DescribeConfigs defaults table), cli.rs (env/listener parsing), local_lease.rs (dev mode).

Write-back side: topic_cr_writer.rs, acl_cr_writer.rs, and user_cr_writer.rs — the only paths where serving a Kafka request writes to Kubernetes (CreateTopics/CreatePartitions/IncrementalAlterConfigs and Metadata auto-topic-creation → KafkaTopic, Create/DeleteAcls → KafkaUser, AlterUserScramCredentials → KafkaUser.spec.authentication.scram — KIP-554, gh #252). In dev mode the topic writer is a no-op impl that refuses politely; the ACL and user writers are simply not installed without a kube client.

Handlers: one module per API key under handlers/. The per-key behaviour — versions, semantics, deviations — is documented exhaustively in Part II's per-API reference; don't duplicate it here or there.

Invariant callers must hold: handlers never talk to Kubernetes or the per-listener auth engine directly — ownership comes from the Coordinator, authorization from the cluster-wide authorizer, and K8s writes go through the CR writers. That's what keeps the hot path runtime-independent.

Start reading at broker.rs, then coordinator.rs, then one thin handler (handlers/list_groups.rs) before the big ones (handlers/produce.rs, handlers/fetch.rs).

kaas-controller

Controller-side logic: Lease election, the partition/group balancer, the assignment writer, and the heartbeat gRPC server.

Everything that runs only on the broker currently holding the kaas-controller Lease (architecture).

Module map: election.rs (the election seam) + kube_election.rs (the Kubernetes Lease implementation; LocalElection is the dev-mode always-elected stub), balancer.rs (partition + consumer-group placement with deterministic smoothing, so recomputes move as little as possible), assignment_writer.rs (atomic assignment.json writes behind the TopicSource / BrokerSource / GroupSource trait seams), and heartbeat_server.rs (the bidi gRPC server side of proto/heartbeat.proto). The k8s_mirror.rs seam for a KafkaClusterAssignments debug mirror was removed along with that CRD in v0.3.2 — its kube-backed writer was never ported from Go.

The trait seams are the point: the balancer and writer are pure over their sources, which is what makes tests/controller_failover.rs and tests/stale_controller_race.rs possible without a cluster — the stale-epoch fence (a deposed controller's write rejected by its stale leaseTransitions epoch) is pinned by test, not by hope.

Invariant callers must hold: the assignment file is the only output channel. Nothing in this crate may instruct a broker directly — brokers follow assignment.json, and anything the controller wants to happen must be expressible as an assignment change.

Start reading at balancer.rs, then assignment_writer.rs.

kaas-auth

SCRAM-SHA-512, SASL PLAIN, and SASL/OAUTHBEARER, mTLS principal mapping, ACL evaluation, and debt-carrying client quotas — loaded from operator-written files with hot-reload.

The security crate, deliberately split along the axis the architecture chapter explains: authentication is per-listener, authorization and quotas are cluster-wide.

Authentication: scram.rs (the SCRAM-SHA-512 server state machine — SCRAM-SHA-256 is not implemented), plain.rs (SASL PLAIN, only offered over TLS), oauth.rs (SASL/OAUTHBEARER, KIP-255: RFC 7628 framing + JWT validation against a hot-swapped JWKS, alg allowlisted, fail-closed before the first key fetch — the fetch loop itself lives in bins/kaas), mtls.rs (peer-cert principal extraction) + principal_mapping.rs (Apache's ssl.principal.mapping.rules syntax, KIP-371 — parse errors fail at startup), engine.rs + selector.rs (the AuthEngine seam and its per-listener selection), credentials.rs (the Strimzi-shape credentials.json loader).

Authorization & quotas: acls.rs (the acls.json loader and ACL engine — deny overrides allow; literal, prefixed, and * pattern types per KIP-290), authorizer.rs (AllowAllAuthorizer + the super-user early-allow wrapper), quota.rs (token buckets with debt-carry — the gh #125 fix that stops N concurrent clients bursting at N× the configured rate), types.rs (Principal, Resource, Operation).

Operational contract: both JSON files are written by the operator and hot-reloaded by the broker — no restart on user/ACL changes, no Kubernetes call on the request path. KAAS_AUTH_DISABLED=true swaps in the allow-all engine everywhere.

Start reading at engine.rs for the seams, then scram.rs against an RFC 5802 refresher, then quota.rs (the debt-carry test multi_client_contention_carries_debt is the spec).

kaas-k8s

Broker-side Kubernetes helpers: peer-endpoint watching, pod identity, the topic watcher, and the partitions-ready readiness gate.

Every module follows the same two-layer split: a pure-state core that tests exercise directly, and a kube-bound pump (behind the default kube-watchers feature) that feeds it events.

Module map: identity.rs (BrokerIdentity — parses the ordinal out of the StatefulSet pod name; no kube dependency), endpoints.rs (BrokerRegistry over the headless Service's EndpointSlices — the broker's live view of its peers, feeding FindCoordinator and Metadata), topic_watcher.rs (a pure-state KafkaTopic cache + divergence detector), readiness.rs (the kaas.rs/PartitionsReady pod readiness gate), kube_watchers.rs (the pumps: lease watch, endpoint watch, topic watch, readiness patching).

An honesty note mirrored from Part II: the production topic pump is kube_watchers::run_topic_watch. Its TopicApply callbacks carry name, partitions, volume assignments (gh #221), the migrate-to annotation (gh #224), and Status.TopicID (gh #241 — fed into the TopicRegistry's incarnation side-map for the stale-dir gate), and the pump takes an on_synced callback fired on the first completed relist. The delete callback closes FDs and purges state immediately (engine.abandon_topic, gh #219; purge_topic_offsets, gh #240) rather than waiting for the assignment recompute. What remains not wired in is the standalone topic_watcher::TopicWatcher cache with its deletionTimestamp-immediate delete events — and Metadata still serves all-zero topic IDs (KIP-516).

The pump is self-healing, and has to be (gh #202). Two properties, both easy to omit and neither optional:

• It restarts its own stream with exponential backoff instead of returning when the stream ends. Kube ends streams routinely; the earlier version returned Ok(()) and trusted the caller to restart it, which the caller never did — so a single relist silently ended topic tracking for the life of the process.
• It reconciles on relist. Event::Init opens a set, InitApply fills it, InitDone retracts every previously-reported topic absent from it. A topic deleted while the watch was disconnected produces no Delete event, so the diff is the only thing that can notice it.

The state backing that diff deliberately outlives any single stream — that's the whole point, and it's why it lives in TopicWatchState rather than inside the stream loop. A relist cut short by a restart drops its partial set rather than retracting topics it never finished enumerating.

Invariant callers must hold: nothing here may block request handling — every consumer of this crate's state reads a cached view, and a dead API server only freezes that view temporarily, until the watch reconnects and reconciles (runtime independence).

Start reading at endpoints.rs (the cleanest example of the two-layer split), then kube_watchers.rs.

kaas-observability

The OTLP metrics + tracing bootstrap, the /healthz HTTP handler, and the byte-opacity tripwire counters.

The content of this crate — what gets exported where, the /healthz runtime view, why gauges read through lock-free snapshots — is covered in the Observability architecture chapter; this page is the code map.

Module map: bootstrap.rs (OTel SDK bring-up from OTEL_EXPORTER_OTLP_* env; metrics push OTLP/HTTP http/protobuf — the only dialect Prometheus's native OTLP receiver speaks; traces OTLP/gRPC), metrics.rs (the Arc-shared Metrics registry — global() returns a no-op registry before bootstrap, so pre-boot code and tests never nil-check), gauges.rs (the GaugeSource seam the broker feeds partition gauges through), health.rs (axum /healthz + /readyz; the RuntimeState trait), byteopacity.rs (the tripwire counters), otlp_push_observer.rs (a PushMetricExporter wrapper that makes OTLP push failures alertable, gh #121), k8s_api.rs (K8s API call metrics), topic_traffic.rs, tracing.rs (tracing-subscriber + OTel layer; every log line carries trace_id/span_id when a span is active).

Invariant callers must hold: gauge callbacks and RuntimeState implementations must never take hot-path locks — the gh #134 outage (a stuck NFS fsync starving the metrics pipeline) is why the storage engine exposes lock-free snapshots for exactly these readers.

Start reading at health.rs (the RuntimeState trait is a compact inventory of what the broker considers its own vital signs).

kaas-operator-api

The kube-derive CRD types — the source cargo xtask gen-crds renders into deploy/crds/ and the Helm chart.

Four CRD types, one module each: kafkacluster.rs (external-listener plumbing, plus the storage / internal-listener / gateway / service config shapes), kafkatopic.rs (partitions + spec.config + Status.TopicID, plus status.volumeAssignments and the kaas.rs/migrate-to-volume annotation contract), kafkauser.rs (Strimzi-shape authentication, inline spec.authorization.acls, and the honestly-named producerMaxByteRatePerBroker / consumerMaxByteRatePerBroker quota fields), plus the condition.rs / scheme.rs helper modules. The semantics — who reconciles what, why the quota names diverge from Strimzi, the no-finalizers model — live in Kubernetes integration.

The generation contract is the thing to internalize: every type derives kube::CustomResource + schemars::JsonSchema, and cargo xtask gen-crds walks them into deploy/crds/*.yaml mirrored into the chart. CI fails on drift — so editing anything in this crate means regenerating and committing both YAML trees in the same change. Field-level validation attributes are part of the wire contract with the apiserver, not decoration.

Both binaries depend on this crate — the operator to reconcile, the broker to read KafkaTopics and write admin changes back. It must stay free of behaviour: types, validation, defaults, nothing else.

Start reading at kafkauser.rs — it's the richest schema and shows every pattern the other three use.

kaas-operator-controllers

One reconciler per CRD, materializing state to files on the shared PVC, plus leader-elected orphan sweeps (at election, then every 5 minutes).

Two layers:

• Reconcilers — kafkatopic_controller.rs (partition directories + .config.json; mints Status.TopicID on first reconcile and never rotates it; refuses partition decrease), kafkauser_controller.rs (derives SCRAM entries into credentials.json, rebuilds acls.json, owns the <user>-kafka-credentials Secret; the gh #104 pre-derived passthrough enables zero-downtime rotation), kafkacluster_controller.rs (cert-manager Certificates, per-broker Services, Gateway TLSRoutes — all with OwnerReferences; also idempotently creates the per-cluster

• Helpers — credentials.rs + acls.rs (the file materializers), sweep.rs (the leader-elected orphan sweep — an initial pass at leader election plus a periodic re-run every 5 minutes, resumable across interrupted passes, gh #205 — dropping topic dirs and credential entries with no matching CR), conditions.rs (status conditions), errors.rs, observer.rs (reconcile counters).

The cleanup model is the crate's defining decision — no finalizers. Deleting CRs never blocks on the operator being alive; Kubernetes GC handles owned resources, and the sweep reclaims on-disk leftovers on the next pass. The ArgoCD cascade-delete deadlock that forced this design is told in Kubernetes integration.

Invariant callers must hold: reconcilers must stay idempotent and convergent — every materialized artifact is rebuilt from the full CR set, never incrementally patched into an unknown state, which is what makes the sweep safe to run blindly at startup.

Start reading at kafkauser_controller.rs (the richest reconcile), then sweep.rs.

kaas-test-harness

Shared test helpers — reserved as the only place in the workspace where a decoded-record representation would be allowed to live.

Still an empty placeholder crate (a doc comment, no code, no dependents), to be populated as integration tests need it. Its charter is narrow and deliberate: shared fixtures and record-construction helpers for tests. Production crates must never grow a decoded-record type — when a test needs one, it belongs here, where the tripwire counters can't be quietly bypassed (wire protocol).

kaas (broker binary)

The broker entrypoint: dispatcher registration, env wiring, the cluster runtime, and the graceful SIGTERM drain.

bins/kaas is deliberately thin on logic and thick on wiring — it's where every seam the library crates expose gets a production implementation plugged in.

main.rs: parses KAAS_LISTENERS (default: one plain listener on 0.0.0.0:9092), selects storage (KAAS_DATA_DIR set → disk engine; unset → in-memory dev mode), builds the per-listener auth engines and the cluster-wide authorizer/quota checker, registers every handler with the dispatcher, spawns the kube topic watch (in cluster mode), starts /healthz//readyz, and owns the graceful SIGTERM drain — relinquish every open partition (persisting manifests + producer snapshots, closing FDs), then flush remaining manifests as defence-in-depth (file-handle ownership). It also carries the --init init-container entry point, the retention-cleaner tick, the credentials/ACL hot-reload task, and the JWKS refresher loops for OAuth listeners.

cluster.rs: the cluster runtime — Lease election glue, heartbeat client/server wiring, broker-set watcher (2 s alive-set poll), topic-change notifier, the assignment loop when this broker is controller, the fence/marker watchers, the txn timeout reaper + gh #225 marker reconcile (same 10 s tick), and the assignment-source hot-swap (the boot-time always-true stub replaced by the real Coordinator-backed source once the runtime is up — the gh #92 dance described in Consumer-group coordination).

Mode selection is three-way (see the top comment in bins/kaas/src/cluster.rs): neither var set → in-memory dev mode (no cluster runtime, local-lease "I lead everything"); KAAS_DATA_DIR set but MY_POD_NAME unset → single-broker disk mode (a Coordinator + assignment loop is installed); MY_POD_NAME set → cluster mode.

Integration tests live in bins/kaas/tests/ — smoke.rs, auth_smoke.rs, byte_opacity.rs, cluster_bringup.rs, cluster_smoke.rs, eos_v2.rs, oauth_smoke.rs, retention.rs — the suites the verification story leans on.

Start reading at main.rs top to bottom — it reads as a map of the whole system.

kaas-operator (operator binary)

The operator entrypoint driving the reconcilers in kaas-operator-controllers.

A small binary by design: it boots the three reconcilers (KafkaTopic, KafkaUser, KafkaCluster) against a namespace-scoped kube::Client, runs the leader-elected orphan sweep (an initial pass at election, then every 5 minutes), serves /healthz + /readyz over axum, and shuts down cleanly on SIGTERM.

Configuration is all environment (chart-templated by deploy/helm/kaas/templates/operator-deployment.yaml): KAAS_DATA_DIR (shared PVC mount, default /data), KAAS_CLUSTER_DIR and KAAS_LOG_DIRS (gh #221), KAAS_NAMESPACE, KAAS_LOG_LEVEL / KAAS_LOG_FORMAT, the metrics bind address and HEALTH_PROBE_BIND_ADDRESS, and the standard OTEL_EXPORTER_OTLP_* variables consumed by kaas-observability's bootstrap.

Remember the architectural stance whenever tempted to grow this binary: the operator is a startup/admission component. Brokers serve traffic while it's down (runtime independence); anything that would put it on a request path belongs elsewhere.

Helm chart & listener configuration

Deploying with the chart: the Strimzi-shape listeners array, cluster-wide authorization values, and how the bundled CRDs are handled.

The chart at deploy/helm/kaas/ is the source of truth for production configuration — replicas, controller-Lease tuning, storage class, image repositories. It deploys the broker StatefulSet, the operator Deployment, and up to three classes of shared RWX PVC: the data volume, an optional dedicated control-plane volume (storage.controlPlane.enabled), and one per storage.pool[] entry. Installation, image derivation, and the smoke test live in the chart's own deploy/helm/kaas/README.md; this chapter covers the concepts that need more than a values table, and the chart values reference documents every key exhaustively.

helm install my-kaas oci://ghcr.io/kaas-rs/charts/kaas \
  --version 0.3.1-preview \
  --namespace kafka --create-namespace \
  --set storage.className=<your-rwx-class> \
  --set broker.replicaCount=3

The listeners array

.Values.listeners is a Strimzi-shape array: each entry declares name (free-form), port, type (internal / external), tls, an authentication.type (none / scram-sha-512 / mtls / plain / oauth), and an optional enabled flag (absence = enabled). The default values ship four entries — plain (9092, anonymous), external (9093, TLS, disabled by default), authed (9095, SCRAM, disabled by default), and oauth (9096, internal, TLS, SASL/OAUTHBEARER, disabled by default).

The templates iterate the array to emit the StatefulSet container ports, the KAAS_LISTENERS JSON env the broker parses, the Service ports, and the NOTES.txt bootstrap output. The three axes are orthogonal — see Listeners, authentication, authorization for how the broker treats them. Combination constraints: mtls authentication requires tls: true, and the broker refuses plain and oauth (SASL PLAIN / OAUTHBEARER) over non-TLS connections at runtime — both send reusable credentials on the wire.

Two behaviours to know before enabling an external listener:

• Only the first type: external listener drives the KafkaCluster CR plumbing (Certificates, per-broker Services, TLSRoutes) — the operator currently understands a single external listener.
• External listeners use per-broker hostnames on one SAN-per-broker certificate (works with HTTP-01 ACME; one DNS record per broker). cert-manager rotates the single Secret in place and brokers hot-reload it without a restart. Scaling replicaCount re-reconciles the Certificate's SAN list.

Cluster-wide authorization

Authorization is deliberately not per-listener: .Values.authorization.type ("" = off, simple = ACL enforcement) and .Values.authorization.superUsers (list of User:foo strings, emitted as KAAS_SUPER_USERS) apply to every listener. Authentication stays per-listener. Users, ACLs, and quotas are authored as KafkaUser CRs — see Kubernetes integration.

CRDs on install and upgrade

The chart bundles its CRDs in crds/. Helm installs them on first install but deliberately never upgrades CRDs from a chart's crds/ directory — on any release that changes them, apply the new CRDs explicitly before helm upgrade (exact commands in the chart README's "CRDs on upgrade" section).

Other levers

• broker.controllerLease.durationSeconds (default 15) — controller failover latency vs API-server write rate.
• broker.minReadySeconds (default 60) — how long a freshly Ready broker must stay Ready before the rollout proceeds to the next pod.
• broker.retentionCheckIntervalSeconds (default 300) — the retention sweep interval; 0 disables the sweep.
• broker.maxMessageBytes (default 1048588) — Apache's message.max.bytes, the cap on one Produce batch.
• broker.fsyncMaxLatencyMs (default 30000) — fsync watchdog deadline; 0 disables.
• auth.requireSasl (default false) — arms the SASL gate on anonymous listeners too, so every connection must authenticate.
• auth.sslPrincipalMappingRules — Apache's ssl.principal.mapping.rules for mapping mTLS subject DNs to principals.
• podDisruptionBudget.maxUnavailable (default 1) — keeps voluntary disruptions from taking multiple single-writer brokers down at once.
• storage.controlPlane.enabled (default false) — moves cluster-wide coordination state onto its own PVC, so a full data volume cannot take the control plane down.
• storage.pool[] (default empty) — additional named volumes for per-topic placement; see the volume pool page.
• storage.* — see Storage substrate requirements; the PVCs carry helm.sh/resource-policy: keep, so uninstall never deletes data.

Implementation notes (for contributors)

• The listeners array is the gh #126 shape, rendered by the helpers in the chart's _helpers.tpl. The KafkaCluster CR template still synthesizes the legacy single-listener shape from the first external entry (via kaas.firstByType); refactoring the operator to consume the array natively is open follow-up.
• The bundled CRDs are generated by cargo xtask gen-crds; the CI rust job fails on drift between the source types and the committed YAML.

Chart values reference

Every key in deploy/helm/kaas/values.yaml, what it does, and where it lands. The Helm chart chapter tells the narrative story (listeners, upgrades, CRDs); this page is the exhaustive reference. Defaults shown are the chart's.

Two Helm behaviors worth knowing before overriding anything:

• Helm deep-merges values, so restating a default changes nothing — and removing a chart default requires an explicit null, not omission.
• A handful of keys are dead — parsed by nothing, kept only until a cleanup release removes them. They are flagged ⚠ dead below rather than silently omitted, so you can tell "documented and inert" from "undocumented".

image and operator.image

operator

broker

broker.podSecurityContext

Default: non-root UID 65532, group 0, fsGroup: 0, fsGroupChangePolicy: OnRootMismatch. Two independent mechanisms both make the shared volume writable: on CSI drivers that honour fsGroup, the kubelet chowns the volume to group 0 (the broker runs 65532:0, the Strimzi "primary GID 0" convention); on shared NFS — where most CSI drivers skip that chown — the broker's partition-init init container is the floor, chowning the volume itself before the broker starts.

broker.controllerLease

The Kubernetes Lease behind controller election. Tighter = faster failover, more API-server writes. On an API server with high tail latency (small shared nodes), widen the ratios — a lease that flaps during pod churn moves the controller for no reason.

Durability, retention, limits

storage

All PVCs the chart renders carry helm.sh/resource-policy: keep — they survive helm uninstall and must be deleted explicitly.

storage.controlPlane

Enabling this on an existing cluster is a breaking change (pre-v1 policy): redeploy fresh, or copy __cluster/* onto the new volume with brokers and operator scaled down.

storage.pool[]

Default []. Named additional RWX volumes — "log dirs" in Kafka's KIP-113 vocabulary — mounted on every broker at /vols/<name> and selectable per topic. Each entry:

The full model — placement stickiness, selectors, explicit migration — is the volume pool chapter.

auth

Authentication/authorization architecture, including what each listener authentication.type means, is Listeners, authentication, authorization.

admin.argocd

Details and the reasoning: Kubernetes integration.

Scheduling and identity

observability

The wider observability story: Observability.

listeners[]

The Strimzi-shape listener array — each entry is one TCP listener on every broker, described by three orthogonal axes (type, tls, authentication.type). The Helm chapter and listeners architecture page cover the model; this is the field reference. Defaults ship four entries: plain (9092, anonymous), external (9093, TLS, disabled), authed (9095, SCRAM, disabled), oauth (9096, TLS + OAUTHBEARER, disabled).

external-type extras

Only the first type: external entry drives the operator's Certificate/Service/TLSRoute reconciliation today.

oauth authentication fields

Strimzi's KafkaListenerAuthenticationOAuth field names, verbatim:

authorization

Implementation notes (for contributors)

• Templates consuming these values live in deploy/helm/kaas/templates/; listener JSON assembly is the kaas.listenersJSON helper in deploy/helm/kaas/templates/_helpers.tpl, landing as the KAAS_LISTENERS env (gh #126). Env-var names for the broker knobs are parsed in crates/kaas-broker/src/cli.rs.
• The ⚠ dead keys (broker.ports.kafka/tls, auth.mechanisms, auth.tls.*, autoscaling.*) and the inert heartbeatRttP99Seconds threshold are tracked in gh #265; the external advertised-hostname gap is gh #263; the unapplied service.annotations is part of gh #266.
• This page documents chart defaults, not any specific deployment. When a key moves or dies, update this page in the same commit — the drift gates don't cover values keys.

CRD reference

Everything you would manage in Apache Kafka through the Admin API or ZooKeeper-era shell tools — topics, users, ACLs, quotas, cluster plumbing — is declared in kaas as Kubernetes custom resources under the kaas.rs/v1alpha1 API group. If you have run Kafka under Strimzi the shape is deliberately familiar; where a kaas CRD diverges from its Strimzi counterpart, the per-CRD page says so and why.

This chapter is the field-level reference. The architectural story — what reconciliation produces, why there are no finalizers, how brokers consume the CRs — lives in Kubernetes integration.

All three are namespaced. The examples in this chapter use the kafka namespace from Getting Started.

Installing and upgrading the CRDs

The CRD YAML ships bundled with the Helm chart (deploy/helm/kaas/crds/) and is also published standalone under deploy/crds/. Helm installs bundled CRDs on first helm install but deliberately never upgrades them — after upgrading the chart, apply the new CRD YAML yourself:

kubectl apply -f deploy/crds/

See Helm chart & listener configuration for the full upgrade procedure and the pre-v1 compatibility rules that go with it.

Conventions shared by every kaas CRD

• Status conditions. Each reconciled CRD reports a Ready condition in status.conditions, surfaced as a printer column, so kubectl get kafkatopics (or kafkausers, kafkaclusters) shows reconcile health at a glance. A Ready=False condition carries a reason and message naming what was rejected.
• No apiserver defaulting for enum-like strings. Fields such as an ACL's patternType or an issuer's kind are left empty in the stored object when you omit them; the operator applies the default at reconcile time. Your stored CR stays byte-for-byte what you wrote, which keeps GitOps diffs clean.
• No finalizers. Deleting a CR never blocks on the operator being alive. Owned Kubernetes resources are garbage-collected via OwnerReferences; on-disk state is reclaimed by a leader-elected sweep. The Kubernetes integration page explains the ArgoCD deadlock that motivated this.
• Deletion is destructive and unguarded. There is no spec-level "protection" flag yet: deleting a KafkaTopic deletes its data (and its committed consumer offsets, matching Apache), deleting a KafkaUser revokes its credential and ACLs.

Implementation notes (for contributors)

• CRD types are kube-derive structs in crates/kaas-operator-api/src/, one module per kind. cargo xtask gen-crds regenerates deploy/crds/ and the chart copy; the rust CI job fails on drift, so commit both when you touch the types.
• Reconcilers live in crates/kaas-operator-controllers/.

KafkaCluster

KafkaCluster is the top-level cluster CR — but it owns much less than a Strimzi user would expect, and that is the first thing to understand about it. Under Strimzi, the Kafka CR is the whole cluster: the operator builds the broker pods, storage, listeners, everything, from it. In kaas the Helm chart owns the broker workload — StatefulSet, volumes, listener array, environment — and the KafkaCluster CR carries only the parts the operator reconciles at runtime: external listener plumbing (cert-manager certificates, per-broker Services, Gateway-API TLSRoutes) and the cluster-scoped status.

You normally never author this CR by hand. The chart templates one per release (deploy/helm/kaas/templates/kafkacluster.yaml) from your Helm values; day-2 changes flow through helm upgrade, not kubectl edit. The reference below is for reading the object and for understanding what the operator does with it.

Spec

apiVersion: kaas.rs/v1alpha1
kind: KafkaCluster
metadata:
  name: kaas
  namespace: kafka
spec:
  replicas: 3
  storage:
    className: nfs
    size: 50Gi
  listeners:
    internal:
      port: 9092
    external:
      enabled: true
      port: 9093
      hostnamePattern: "broker-%d.kafka.example.com"
      bootstrapHostname: "kafka.example.com"
      tls:
        certManager:
          enabled: true
          issuerRef:
            name: letsencrypt
            kind: ClusterIssuer
      gateway:
        enabled: true
        gatewayRef:
          name: public-gateway
          namespace: gateway-system

Everything the operator creates from this CR carries an OwnerReference back to it, so deleting the KafkaCluster lets Kubernetes garbage-collect the Certificates, Services, and TLSRoutes with no operator involvement.

Note the asymmetry with the chart's listener model: the chart supports an arbitrary array of listeners (see Helm chart & listener configuration), while this CR still models the legacy single internal/external pair. The chart bridges the two by synthesizing this shape from the first listener of each type; refactoring the operator to consume the array natively is planned.

Status

kubectl get kafkaclusters prints Replicas, External (whether the external listener is enabled), and Ready.

What it does not do

• It does not create or scale the broker pods — Helm does.
• It does not define which listeners exist or their authentication — that is the chart's listener array, delivered to brokers by environment variable.
• Deleting it does not delete topic data; it tears down the external access path only.

Implementation notes (for contributors)

• Type: crates/kaas-operator-api/src/kafkacluster.rs; generated schema deploy/crds/kaas.rs_kafkaclusters.yaml.
• Reconciler: crates/kaas-operator-controllers/ (KafkaCluster reconciler, 300 s requeue).
• The chart→CR template is deploy/helm/kaas/templates/kafkacluster.yaml, using the kaas.firstByType helper to collapse the listener array into the legacy single-listener shape (gh #126 follow-up tracks consuming the array natively).

KafkaTopic

KafkaTopic declares a topic: its partition count, its per-topic configuration, and (optionally) which storage volumes hold its data. It is the kaas equivalent of Strimzi's KafkaTopic, and like Strimzi's it is bidirectional: you can author topics as CRs in git, or create them over the Kafka protocol (kafka-topics.sh --create, AdminClient, auto-creation) — a wire-created topic appears as a CR minted by the broker, and wire-level config changes (kafka-configs.sh --alter) are patches to the CR. Either way, kubectl get kafkatopics always shows the truth.

Spec

apiVersion: kaas.rs/v1alpha1
kind: KafkaTopic
metadata:
  name: orders
  namespace: kafka
spec:
  partitions: 12
  config:
    retentionMs: 604800000      # 7 days — also the enforced default
    segmentBytes: 268435456     # 256 MiB
  storage:
    volumes: [premium]          # optional; requires a volume pool

There is no replicas field and never will be: kaas has no replication — durability comes from the storage substrate. Wire-level creates asking for replicationFactor > 1 are accepted and clamped, since clients routinely hardcode 3.

spec.config

Each field maps 1:1 to a Kafka topic config; unset means the broker default. Values land in the topic's .config.json on the shared volume, which brokers re-read on use — config changes hot-reload, no restart, taking effect at the next retention sweep / segment roll / append.