Kafka, from the inside out

Most Kafka tutorials teach you the API: create a producer, call send(), write a poll loop, done. That gets you to "it works on my laptop". It does not tell you why a consumer group freezes for thirty seconds, why a broker that lost its disk did not lose your data, or why acks=all is still not enough on its own. Those answers live one layer down, in the mechanisms. This guide is about that layer.

We will keep one picture in our heads the whole way through. Imagine a city's ledger archive. Every event in the city is written, in order, as a numbered line in a bound ledger book. Nobody ever erases or rewrites a line; they only add new lines at the end. The archive has several branches (buildings), each holding some of the books, and every important book has photocopies in other branches in case one burns down. A city council keeps the official minutes of which branch holds which book. Teams of clerks read the books and keep a bookmark of how far they got. That is Kafka: records, partitions, brokers, replicas, the KRaft controller quorum, consumer groups and committed offsets.

The guide has levels. Chapters and blocks carry a chip — L1 essentials, L2 practitioner, L3 expert internals, L4 mastery (operations and application design). Choose your depth with the switch in the bottom-left corner; deeper material hides until you want it. The map below jumps anywhere and ticks off what you've read.

Kafka's answer is the list. Everything in Kafka is built on one simple data structure: the log. Here "log" doesn't mean a debug log. It means an append-only sequence of records. You can add records at the end and you can read from any position. You can't change or insert a record in the middle.

Meet the city ledger archive

We'll use one picture for the whole course, so let's set it up carefully. Imagine a city's ledger archive, an old-fashioned records office. Clerks write every event that happens in the city into bound ledger books. They never erase a line and never squeeze a line between two others. A new event always goes on the next empty line.

Records

The unit of data is a record. The docs also call it an event or a message. A record says that "something happened". It has a key, a value, a timestamp and optional headers (small key/value metadata). The official introduction uses this example: key "Alice", value "Made a payment of $200 to Bob", and a timestamp. To Kafka, the key and value are just bytes. Your application decides what they mean (JSON, Avro, a plain string…).

Topics

Records are organized and durably stored in topics, for example payments or page-views. That's our ledger series. A topic is multi-producer and multi-subscriber: zero, one or many applications can write to it, and zero, one or many can read from it. And, unlike a classic message queue, reading a record does not delete it. Records stay until the topic's retention rule throws them away (chapter 6), so the same data can be read again and again, by different readers, at different times.

Partitions and offsets

One ledger book would be a bottleneck: one book can only be written by one hand at a time. So a topic is split into partitions, several books in the same series. The partitions can live on different servers, so many clients can read and write the topic in parallel. That's how Kafka scales.

Inside a partition, every record gets a sequential number called the offset: 0, 1, 2, 3… It's the printed line number in the book. Offsets are per partition. Partition 0 and partition 1 both have a record at offset 5, and the two are unrelated. To name a record uniquely you need three things: topic, partition and offset.

Which book does a new record go into? The producer picks. With the default rules, records with the same key always go to the same partition because the partition is chosen from a hash of the key. All of Alice's payments land in one book, in the order they were written. Records without a key are spread over partitions (chapter 3 shows how).

That sentence is the most important one in this chapter. The docs put it this way: any consumer of a given topic-partition will always read that partition's events in exactly the same order as they were written. Across two partitions there is no such promise. Play with it below. Type a key, append it, and watch where it lands. Then press the break-it button.

bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic payments --partitions 3

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic payments \
  --reader-property parse.key=true --reader-property key.separator=:
> alice:paid 10
> carol:paid 5
> alice:paid 7

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic payments \
  --from-beginning --formatter-property print.key=true \
  --formatter-property print.partition=true --formatter-property print.offset=true

One archive building can't hold every ledger in a big city, and one fire would destroy everything. So the city runs several branches. In Kafka, a branch is a broker: a server process that stores partitions and serves reads and writes for them. A group of brokers working together is a cluster.

The partitions of a topic are spread across the brokers. Each partition also has copies (replicas, chapter 5) on several brokers, and exactly one of those copies is the leader. All writes for a partition go to its leader. There are normally many more partitions than brokers, and Kafka spreads leadership around so each broker leads its fair share.

Finding the right branch: bootstrap + metadata

A client (producer or consumer) needs to know which broker leads which partition. It learns this in two steps:

1. Bootstrap. You configure bootstrap.servers, a list like host1:9092,host2:9092. It's used for the first connection (and again to re-bootstrap if every known broker becomes unreachable, the default metadata.recovery.strategy=rebootstrap). It doesn't need to list every broker, but the docs recommend listing more than one in case one is down.
2. Metadata. The client asks any broker for metadata. Every broker can answer: which brokers exist, and for each partition which broker is the leader, which brokers hold replicas, and which replicas are in sync.

Then the client connects directly to the leader of each partition it needs. There's no router or load balancer in the middle. In archive terms: you walk into any branch and read the notice board ("Payments book 1 is kept at branch B2"), then you walk to B2 yourself. This is a deliberate design choice. No extra hop, and no central bottleneck in the data path.

Who keeps the notice board up to date? The controllers

Somebody has to decide which broker leads which partition and notice when a broker dies. In Kafka 4.x that's the job of the controllers. They form a small quorum and use a Raft-based protocol called KRaft. In the analogy, they're the city council that keeps the official registry. Brokers register with the controller and send it regular heartbeats. If no heartbeat arrives within broker.session.timeout.ms (default 9000 ms), the broker is considered offline, and the controller picks a new leader for its partitions from the replicas that are in sync.

For learning you can run one process that is both broker and controller. The quickstart's config/server.properties sets process.roles=broker,controller. In production they're usually separate machines. Kafka 4.x has no ZooKeeper at all. Chapter 9 covers KRaft in depth.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments

A producer is the client that writes records to Kafka. In the archive, it's a courier. You hand it letters (records); it sorts them by destination book (partition) into sacks (batches), and a van (a background thread) drives full sacks to the right branch. Sending one letter per trip would be absurdly slow, so batching is at the heart of Kafka's performance.

What happens when you call send()

The Java KafkaProducer.send() is asynchronous. It puts the record into a buffer and returns right away with a Future. You can wait on that future or pass a callback to learn the outcome later. Inside, the record goes through a small pipeline:

1. Metadata. The producer needs to know the topic's partitions (chapter 2). The first send to a topic may wait for metadata.
2. Serializer. Your key and value objects become bytes via key.serializer and value.serializer, e.g. StringSerializer.
3. Partitioner. Picks the partition. If you set one explicitly, it's used. Otherwise, with a key, it's murmur2(key bytes) made positive, mod the partition count. Without a key, the producer uses a sticky partition that changes after at least batch.size bytes have been sent to it.
4. Record accumulator. The record is appended to the open batch for that partition, in a memory pool of size buffer.memory.
5. Sender thread. A background I/O thread (named kafka-producer-network-thread…) collects ready batches, groups them per leader broker into produce requests, and sends them.

A batch is "ready" when it's full (batch.size, default 16384 bytes) or it has waited linger.ms. Since Kafka 4.0 the default linger.ms is 5 ms (it was 0 before). The source says the change was made because larger batches usually give similar or lower latency despite the extra wait.

acks: when is a write "done"?

The acks setting decides how much confirmation the producer waits for. acks=0: don't wait at all. acks=1: the leader wrote it to its own log. acks=all (same as -1, and the default): the leader waits until all in-sync replicas have it. Chapter 5 shows what each one can lose.

Producers also retry automatically (retries defaults to Integer.MAX_VALUE, bounded in time by delivery.timeout.ms, default 120000 ms). Retries raise a problem: if an ack is lost, the retry could write the record twice. That's why enable.idempotence is true by default. The broker then drops the duplicate retry (chapter 12 shows how). Idempotence needs acks=all. If you set a conflicting value like acks=1 without explicitly enabling idempotence, it is quietly turned off.

A consumer reads records from topics. In the archive, a consumer is a clerk who reads a ledger book line by line and does something with each entry: updates a balance, sends an email, copies it into a data warehouse.

Pull, not push

Kafka consumers pull. The consumer sends fetch requests to the partition leaders and says "give me records starting at offset N". The broker never pushes data at a consumer. The design docs explain why: with push, a slow consumer gets flooded when production outpaces it. With pull, a slow consumer just falls behind and catches up when it can. Pull also batches naturally, because each fetch takes everything available after the position (up to a size limit).

What if there's nothing new? The consumer doesn't spin in a tight loop. A fetch can wait on the broker (a "long poll") until at least fetch.min.bytes (default 1 byte) is available or fetch.max.wait.ms (default 500 ms) has passed.

The poll loop

Your code calls consumer.poll(timeout) in a loop. Each call returns a batch of records (at most max.poll.records, default 500), you process them, then you poll again:

consumer.subscribe(List.of("payments"));
while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> r : records) {
        handle(r.key(), r.value());          // your business logic
    }
    // with enable.auto.commit=true (default), progress is committed in the background
}

Two rules to remember from the Javadoc. The KafkaConsumer is not thread-safe: use one instance per thread. And you must call poll() often enough. If more than max.poll.interval.ms (default 300000 ms, 5 minutes) passes between polls, the consumer is considered failed and its partitions are given to someone else.

Position vs committed offset: the bookmark

A consumer tracks two numbers per partition:

• Position: the offset of the next record it will receive. It's one more than the highest offset it has seen, and it advances automatically with every poll(). It lives in the consumer's memory.
• Committed offset: the last position stored safely in Kafka. If the consumer crashes and restarts (or another consumer takes over), it resumes from here.

In the archive, the committed offset is a bookmark the clerk slips into the book and registers at the front desk. The position is just where the clerk's finger is right now. If the clerk faints, the finger's position is lost, but the bookmark stays.

Where do bookmarks live? In an internal, compacted Kafka topic called __consumer_offsets. Each group has one broker acting as its group coordinator. The consumer finds it with a FindCoordinator request and sends its offset commits there. The coordinator writes each commit to __consumer_offsets and replies only after all replicas of that topic have it. By default the topic has 50 partitions (offsets.topic.num.partitions) and replication factor 3 (offsets.topic.replication.factor).

By default (enable.auto.commit=true), the consumer commits its position automatically every auto.commit.interval.ms (default 5000 ms). You can switch that off and call commitSync() or commitAsync() yourself.

The order of process and commit decides your delivery guarantee. Commit then process: a crash in between skips records (at-most-once). Process then commit: a crash in between re-delivers records (at-least-once). At-least-once is Kafka's default promise. So write handlers that can safely see the same record twice.

Consumer groups: a team of clerks

One clerk can't keep up with a big series. So consumers join a consumer group by sharing the same group.id. Kafka splits the topic's partitions among the group's members so that each partition is read by exactly one consumer in the group. A topic with four partitions and two consumers gives each consumer two partitions. When a consumer joins, leaves or crashes, the partitions are redistributed. That's called a rebalance.

Different groups are completely independent. Each group gets all the records and keeps its own bookmarks. That covers both classic messaging styles. Queue-like: all workers in one group share the load. Pub-sub-like: each application uses its own group and sees everything.

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic payments --group billing --from-beginning

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing

Disks die. Machines reboot. A whole rack loses power. If each partition lived on only one broker, any of those would make data unavailable or lose it. So Kafka keeps replicas: copies of each partition on several brokers. In the archive, every ledger book has photocopies kept in other branches, and the copies are updated as new lines are written.

Leader and followers

The number of copies is the topic's replication factor (RF). You set it per topic, and it includes the original. One replica is the leader. It receives all writes (and normally all reads). The others are followers. They fetch new records from the leader, just like a consumer would, and append them to their own copy. All replicas have the same records at the same offsets in the same order. A follower may just be a few records behind at the tail.

The broker-wide default for new topics is default.replication.factor=1, which means no copies at all. The docs call RF 3 a common production setting. Set it explicitly (for example --replication-factor 3).

In-sync replicas (ISR)

Not every follower is always up to date. The leader tracks the in-sync replicas (ISR): the replicas that are caught up. A replica is in sync if its broker has a live session with the controller and it keeps up with the leader. A follower that can't catch up to the leader's log end within replica.lag.time.max.ms (default 30000 ms) is removed from the ISR. It rejoins after catching up fully.

A record is committed when all current ISR members have it and the ISR has at least min.insync.replicas members (more on that setting below). Only committed records are handed to consumers. If the leader dies, the controller picks the new leader from the ISR. Every ISR member has every committed record, so nothing committed is lost. With RF f+1, Kafka can survive f failed replicas without losing committed records.

acks and min.insync.replicas: your durability dial

The producer's acks decides when it hears "done":

• acks=0: the record is put in the socket buffer and considered sent. No guarantee the broker got it; retries don't apply. The returned offset is always -1.
• acks=1: the leader wrote it to its own log. If the leader dies before a follower copies it, it's lost.
• acks=all (default): the leader waits for the full ISR. Not lost as long as one in-sync replica survives.

There's a trap in acks=all. "All" means all current ISR members, and the ISR can shrink. If only the leader is left in the ISR, acks=all is effectively acks=1. The topic setting min.insync.replicas (default 1) closes the gap. With acks=all, if the ISR is smaller than this number, the write is rejected with NotEnoughReplicas (or NotEnoughReplicasAfterAppend) instead of being accepted with too few copies. The Kafka docs give the typical recipe: RF 3, min.insync.replicas=2, acks=all. A majority must persist each write, and you can still lose one broker and keep writing.

Reading doesn't delete, so something else must, or disks fill up forever. That's retention. The archive has a policy like "we keep ledger entries for seven days". Old records are removed based on time or size, never based on whether someone read them. A consumer that is days behind can still catch up, as long as it stays inside the retention window.

Two cleanup policies

Each topic has a cleanup.policy:

• delete (the default): discard old data when it's older than retention.ms or when the partition is bigger than retention.bytes. Good for event streams where each record stands alone: clicks, logs, payments.
• compact: keep at least the latest record for each key and remove older records with the same key. The archive only keeps the newest entry per account. Good for "current state" topics, like a customer's latest address. Chapter 14 covers compaction.
• Both at once, delete,compact: compact, and also drop old segments by time/size.

The defaults

The broker-level equivalents (log.retention.hours = 168, log.retention.bytes, log.cleanup.policy…) set the default for every topic. The topic-level settings override them for one topic.

Deletion happens per segment

Remember from chapter 1 that a partition is a series of segment files (the book's volumes). Time and size retention never delete single records. They delete whole segments, oldest first. For time-based retention, the largest timestamp in the segment counts. A segment goes only when even its newest record is older than retention.ms. So data often lives longer than retention.ms: up to about one extra segment's worth of time, plus the time until the next retention check.

If both limits are set, a segment is deleted when either the time or the size rule says so.

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --entity-type topics --entity-name payments --add-config retention.ms=259200000

In Part I a partition was one bound ledger book. Time to confess: nobody binds a book that never ends. In the archive, each "book" is really a shelf of thin volumes. Clerks only ever write in the last volume; when it gets thick, they shelve it and start a fresh one. Kafka calls those volumes segments.

On disk, a topic called my-topic with two partitions is simply two directories, my-topic-0 and my-topic-1. Inside each directory the segment files are named after the offset of the first record they contain, zero-padded to 20 digits, so the first one is always 00000000000000000000.log. That file name is the segment's base offset.

A segment is not one file but a small family that shares the same base-offset name:

Next to them live a couple of per-partition files such as leader-epoch-checkpoint and partition.metadata. You will meet the leader-epoch file again in Part III.

Rolling: when does a new volume start?

The active segment is rolled (closed, and a new one started) when it would grow past segment.bytes (default 1 GiB), when it is older than segment.ms (default 7 days), or when one of its index files is full. Why care? Because retention and compaction work one whole segment at a time. Time-based retention looks at the largest timestamp in a segment. If a topic has retention.ms of one hour but keeps writing into the same 1 GiB segment for a day, that segment's newest record is always fresh, so its oldest records stay on disk far longer than an hour. Records leave in whole volumes, never one line at a time.

Reading: three hops to any offset

A consumer says "give me offset 2 600". The broker does three cheap hops. 1. Pick the segment: segments are kept in a sorted map keyed by base offset, and the broker takes the floor entry (the largest base offset ≤ 2 600). 2. Look in that segment's .index (memory-mapped, binary-searched) for the largest indexed offset ≤ 2 600, which gives a byte position. 3. Read the .log forward from that position until it meets the batch containing 2 600. Try it:

Notice what the index is not: it is not an entry per record. It is sparse. The broker adds an entry only after roughly index.interval.bytes (default 4096) of data has been appended since the last one. So the index stays tiny and hot in memory, and the forward scan is at most a few kilobytes. That is the whole trick: a binary search over a small array plus a short sequential read.

Why "just files" is fast: page cache, sequential I/O, zero-copy

Kafka's design notes are blunt about it: "don't fear the filesystem". Linear reads and writes are the pattern disks and operating systems handle best (the docs quote about 600 MB/s linear versus about 100 kB/s random writes on one old disk array). The OS already spends all free memory on the page cache, so Kafka appends into that cache and doesn't build its own heap cache. That avoids storing everything twice, avoids GC pressure, and the cache stays warm when the broker process restarts.

On the way out, the broker uses sendfile (Java's FileChannel.transferTo): bytes go from page cache to the socket without a round-trip through user space. The usual path is four copies and two system calls; with sendfile only the final copy to the NIC buffer remains. Because the on-disk format equals the wire format, there is nothing to convert. The docs point out that when consumers are caught up, you often see no disk reads at all, because everything is served from cache.

The fsync question

By default Kafka never forces an fsync on a schedule: flush.messages and flush.ms both default to Long.MAX_VALUE. An acknowledged write may sit only in page cache. Is that reckless? No. Kafka gets durability from replication, not from the disk of one machine. A write acknowledged with acks=all is in the page cache of every in-sync replica, and a crashed replica must fully re-sync before it may rejoin the ISR (Chapter 10). The design notes put the cost of fsync-per-write at "two to three orders of magnitude" of throughput. The config docs literally say "we recommend you not set this".

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic seg-demo \
  --partitions 1 --config segment.bytes=1048576
bin/kafka-producer-perf-test.sh --topic seg-demo --num-records 30000 --record-size 100 \
  --throughput -1 --bootstrap-server localhost:9092
ls -l /tmp/kraft-combined-logs/seg-demo-0/
bin/kafka-dump-log.sh --files /tmp/kraft-combined-logs/seg-demo-0/00000000000000000000.log | head
bin/kafka-dump-log.sh --files /tmp/kraft-combined-logs/seg-demo-0/00000000000000000000.index | head

Kafka never really stores "a message". It stores record batches. Even a single send() becomes a batch of one. In archive terms, a batch is a sealed envelope delivered to the ledger office: the clerk stamps the envelope once (offset, checksum, producer id) instead of stamping every letter inside. This is the "message set" idea from the design docs. Group messages so that network round-trips, disk writes and checksums are paid per batch, not per record.

The same bytes travel producer → broker → disk → consumer unchanged whenever possible. That's why the format is a stable, versioned interface: the current magic value is 2 (introduced in Kafka 0.11; older formats are documented only for history). Here is the envelope, field by field. Click any field.

Records inside the envelope: deltas and varints

Each record inside the batch is tiny on purpose. It has a varint length, one unused attributes byte, a timestampDelta (varlong) and an offsetDelta (varint), both relative to the batch's base values, then key, value and headers, each prefixed by a varint length. Varints use the same encoding as Protobuf, so small numbers cost one byte. A record whose offset is baseOffset + 3 and that was produced 2 ms after the first costs one byte for each delta instead of 8 + 8.

Headers (key/value pairs of metadata, like tracing ids) are part of every record. Their key is never null, the value may be null, and their order is preserved from producer to consumer.

Compression: end-to-end and batch-level

Compressing one message at a time compresses badly: the redundancy is between messages (the same JSON field names, the same user agents). So Kafka compresses the records section of a whole batch. The codec is recorded in bits 0–2 of attributes: 0 none, 1 gzip, 2 snappy, 3 lz4, 4 zstd. The compressed data follows directly after recordsCount.

And it stays compressed. The design docs describe the path: the producer compresses; the broker decompresses only to validate (for example, that the record count matches the header); the batch is written to disk still compressed; it is sent to consumers compressed; the consumer decompresses. Compression CPU is paid at the edges, and every broker hop, disk byte and replica fetch benefits.

The level configs exist both on the producer and as topic configs.

for c in none zstd; do
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic comp-$c --partitions 1
  bin/kafka-producer-perf-test.sh --topic comp-$c --num-records 20000 --record-size 200 \
    --throughput -1 --bootstrap-server localhost:9092 \
    --command-property compression.type=$c linger.ms=20
done
du -sh /tmp/kraft-combined-logs/comp-*-0
bin/kafka-dump-log.sh --files /tmp/kraft-combined-logs/comp-zstd-0/00000000000000000000.log | head -3

Who decides which branch holds which book, which photocopy is current, which branches are open today? In our city, the council does, and it doesn't send letters around. It keeps minutes: a numbered, append-only record of every decision. Branches read the minutes and update their own copy of the registry. That's KRaft. Since Kafka 4.0 it's the only mode; ZooKeeper is gone.

Concretely, cluster metadata (topics, partitions, leaders, ISRs, broker registrations, configs, ACLs, feature levels) lives in a single-partition internal topic called __cluster_metadata. Every change is a record in that log: TopicRecord, PartitionRecord, PartitionChangeRecord, RegisterBrokerRecord, BrokerRegistrationChangeRecord, ConfigRecord, FeatureLevelRecord and so on. The current state of the cluster is simply "replay the log from the start" (or from a snapshot).

Roles: controllers and brokers

Each server sets process.roles to broker, controller, or broker,controller ("combined", fine for development, not recommended for critical deployments). Controllers form the metadata quorum: one is the active controller (the Raft leader), the others are hot standbys that replicate the log. You typically run 3 or 5. A majority must be alive, so 3 tolerate 1 failure and 5 tolerate 2 (the docs' rule: 2N + 1 controllers survive N failures).

Brokers are observers: they replicate the metadata log too but never vote. In kafka-metadata-quorum.sh describe --status you literally see CurrentVoters (controllers) and CurrentObservers (brokers).

Raft, Kafka-style: pull, not push

Textbook Raft has the leader push entries to followers. KRaft's implementation (KafkaRaftClient) describes itself as a "Kafkaesque version of the Raft protocol". Leader election is more or less pure Raft (a Vote request carrying the candidate's last log offset), but replication is driven by followers fetching, using the same Fetch API that partition replicas use (Chapter 10). Because nobody pushes, a new leader sends BeginQuorumEpoch so voters learn who to fetch from. A voter that hasn't fetched successfully from the leader for controller.quorum.fetch.timeout.ms (default 2 000 ms) becomes prospective and runs a pre-vote (KIP-996: a Vote request with PreVote=true, which isn't persisted). Only if a majority grants it does it become a candidate and start a real election.

An entry is committed once a majority of voters have it. That's the metadata log's high watermark. The active controller only completes a request once its records are durable in the metadata log, so a decision it has acknowledged can't be "un-made" by a controller failover.

Broker liveness: heartbeats and fencing

When a broker starts, it registers with the active controller (a RegisterBrokerRecord lands in the log) and then sends a heartbeat every broker.heartbeat.interval.ms (default 2 000 ms). If the controller gets no heartbeat within broker.session.timeout.ms (default 9 000 ms), the broker's lease has expired. The controller fences it (a BrokerRegistrationChangeRecord with fenced = FENCE) and moves leadership of its partitions to other ISR members. The design docs define "alive" exactly like this: an active session with the controller, plus keeping up with the leader when acting as a follower.

Static vs dynamic quorums (KIP-853)

Older clusters list every controller in controller.quorum.voters: a static quorum, where membership can't change without a full reconfiguration. With kraft.version 1 (release 4.1+ for upgrades of existing clusters), quorums are dynamic: nodes find the quorum through controller.quorum.bootstrap.servers (like bootstrap.servers for clients), and you add or remove controllers online with kafka-metadata-quorum.sh add-controller / remove-controller. The quorum is formatted as dynamic when controller.quorum.voters is absent and you format with --standalone, --initial-controllers or --no-initial-controllers.

bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3
bin/kafka-dump-log.sh --cluster-metadata-decoder \
  --files /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000.log | grep -A2 orders

Part I gave you the rules: replicas, a leader, an ISR, and acks=all waits for the ISR. Now let's watch the machinery that enforces them. It runs on two numbers per replica and one simple loop.

Followers pull. A follower is basically a consumer of its leader. It sends Fetch requests with a FetchOffset ("give me everything from offset N"), appends what it gets to its own log, and asks again. The design docs point out the bonus: pulling lets the follower batch naturally. In the archive: each branch that keeps a photocopy sends a runner to the main branch saying "I have lines up to 41, give me 42 onward".

Two numbers you must know

• LEO (log end offset): the offset of the next record a replica would write. Each replica has its own.
• HW (high watermark): the offset up to which records are committed, meaning every in-sync replica has them. On the leader, the HW is the smallest LEO among the ISR (and caught-up replicas), and it only moves while the ISR has at least min.insync.replicas members.

The clever part: the leader learns each follower's LEO for free, because a follower's FetchOffset is its LEO. And followers learn the HW for free, because every fetch response carries the leader's HighWatermark. There's no separate "ack" message. The fetch loop is the replication protocol.

What the high watermark controls

Consumers only see committed records. A fetch from a normal consumer returns data only below the HW. Records above it exist on the leader but might vanish if the leader dies before followers copy them, so Kafka won't hand them out. The docs put it plainly: "only committed messages are ever given out to the consumer", regardless of the producer's acks. Records become visible only once they're on all in-sync replicas and the ISR has at least min.insync.replicas members.

acks=all waits for the HW. The leader appends the batch, then parks the produce request until the HW has passed the batch's last offset, i.e. until every ISR member has fetched it. Only then does the producer get its acknowledgement. acks=1 answers right after the leader's local append; acks=0 doesn't wait at all.

min.insync.replicas guards the door. With acks=all, if the ISR is smaller than min.insync.replicas, the leader rejects the write with NOT_ENOUGH_REPLICAS. The partition-level code also only advances the HW when the ISR is at least min.insync.replicas. Without this, "acks=all" would degrade silently to "acks=leader-only" when followers drop out. Writes with acks=0 or 1 are still accepted below the minimum, but they stay invisible to consumers until the ISR is back up to it.

ISR shrink and expand

The leader periodically checks each follower. A follower that hasn't caught up to the leader's LEO within replica.lag.time.max.ms (default 30 000 ms) is out of sync. This covers both a stuck follower (no fetches at all) and a slow one (fetching, but never reaching the end). The leader asks the controller to shrink the ISR (an AlterPartition request; the change becomes a metadata record). Once the ISR is smaller, the HW can move again, and acks=all producers unblock.

When the lagging follower catches up (its LEO reaches the HW within the current leader epoch), the leader asks to expand the ISR again. Note the time-based definition: Kafka doesn't count "messages behind". A follower 10 000 records behind that's catching up fast stays in; one that's only 5 behind but hasn't reached the end for 30 s goes out.

In Part I a consumer group was a team of clerks sharing ledger books: each book goes to exactly one clerk, and each clerk keeps a bookmark (the committed offset). The hard part isn't reading. It's changing who reads what when a clerk arrives, leaves or collapses, without two clerks reading the same book and without a book left unattended. That handover is a rebalance, and Kafka has two very different protocols for it.

The group coordinator and __consumer_offsets

Every group has a group coordinator: a broker-side component on the broker that leads one partition of the internal topic __consumer_offsets. Which partition? abs(groupId.hashCode()) % offsets.topic.num.partitions (default 50 partitions, replication factor 3). A consumer finds its coordinator with a FindCoordinator request, then sends all group traffic (joins, heartbeats, offset commits) there. Commits and group state are written as records into that partition, so they're replicated like any other data, and a new coordinator can rebuild them by replaying the partition after a failover.

Classic protocol: JoinGroup / SyncGroup

The classic protocol (the client default in 4.3: group.protocol=classic) puts the brains in the client:

1. Something changes (a member joins or leaves, a member misses heartbeats, subscriptions change). The coordinator moves the group to PreparingRebalance and every member must send JoinGroup.
2. The coordinator waits for members to rejoin, picks one as the group leader and gives it everyone's subscriptions.
3. The leader runs the client-side assignor (partition.assignment.strategy) and sends the plan in SyncGroup. The coordinator hands each member its share.

That's a global synchronization barrier: nobody gets a new assignment until everyone has joined. How painful it is depends on the rebalance flavour:

• Eager (e.g. RangeAssignor, RoundRobinAssignor, StickyAssignor): each consumer must revoke all its partitions before joining. The whole group stops. That's the famous "stop-the-world" rebalance.
• Cooperative (CooperativeStickyAssignor): consumers keep their partitions while rebalancing. The assignor withholds partitions that must move; their owners revoke just those, and a second rebalance gives them to the new owner. Everyone else keeps working.

The default partition.assignment.strategy is [RangeAssignor, CooperativeStickyAssignor]. That means eager Range by default, but you can switch to cooperative with a single rolling bounce that removes RangeAssignor from the list.

The new consumer protocol (KIP-848)

KIP-848 is GA since Kafka 4.0 and enabled by default on the server. Clients opt in with group.protocol=consumer. It moves the brains to the broker and removes the barrier:

• Members send a periodic ConsumerGroupHeartbeat (with MemberEpoch, subscribed topics or a regex, and the partitions they own). There's no JoinGroup/SyncGroup.
• The coordinator computes a target assignment with a server-side assignor (group.consumer.assignors: uniform by default, or range).
• Each member is reconciled individually through heartbeat responses: first "revoke these", then "you now own these". A partition is only given to its new owner once the old owner has confirmed it released it. That's what keeps two clerks off the same book.
• Heartbeat interval and session timeout become broker configs (group.consumer.heartbeat.interval.ms 5 000, group.consumer.session.timeout.ms 45 000). The client configs session.timeout.ms, heartbeat.interval.ms and partition.assignment.strategy can't be used with it.

Failure detection: two different clocks

A consumer can be "dead" in two ways. Crashed: heartbeats stop, and after the session timeout (session.timeout.ms, default 45 s, classic; group.consumer.session.timeout.ms, default 45 s, new protocol) the coordinator removes it. Until then, its partitions are not read by anyone. Stuck: the process heartbeats happily, but your code hasn't called poll() for more than max.poll.interval.ms (default 300 000 ms = 5 min). The consumer is considered failed, leaves the group, and a rebalance hands its partitions to others. When the slow code finally commits, it gets a commit failure because it no longer owns the partitions.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic rb-demo --partitions 6
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic rb-demo --group g1 \
  --command-property group.protocol=consumer
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic rb-demo --group g1
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list --type
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g1 --members --verbose
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g1 --state

Here's the classic distributed-systems trap. The producer sends a batch; the leader appends it; the acknowledgement gets lost on the way back (a timeout, a dropped connection). The producer can't tell "not written" from "written but not acknowledged", so it retries. Before Kafka 0.11 that meant a duplicate in the log: at-least-once delivery. The design docs compare it to inserting into a database table with an auto-generated key.

The fix is the one every good clerk uses: number the envelopes. If each sender writes "envelope #7" on every envelope, the clerk can say "I already filed #7" and hand back the old receipt instead of filing it twice. Kafka does exactly this, and it's on by default (enable.idempotence=true).

The three numbers

• Producer ID (PID): a 64-bit id the producer gets from the cluster at start-up (InitProducerId request). It identifies this producer session.
• Producer epoch: a 16-bit counter for the same PID. It increases when the session is "reset" (for example, after certain errors), so older in-flight writes from a previous incarnation can be recognized.
• Sequence number: per PID and per partition, starting at 0 and increasing by one per record. A batch carries its baseSequence, and its last sequence follows from the record count.

All three live in the batch header you met in Chapter 8 (producerId, producerEpoch, baseSequence). They cost nothing extra on the wire.

What the broker checks

For every partition, the leader keeps a small producer state per PID: the current epoch and metadata about the last 5 batches it appended (sequence range and offsets). When a batch arrives:

• If its sequence range matches one of those last 5 batches, it's a duplicate. The broker doesn't append it; it answers with the offsets of the original batch, and the producer is happy.
• If its first sequence is exactly last sequence + 1, it's the next batch: append it.
• If there's a gap (or it's out of order), the broker rejects it with OUT_OF_ORDER_SEQUENCE_NUMBER.
• If its epoch is older than the one the broker knows, it's rejected with INVALID_PRODUCER_EPOCH.

Why "max 5 in flight"?

The broker only remembers the last 5 batches per producer per partition. So the producer must never have more than 5 unacknowledged requests on one connection. Otherwise a retry could refer to a batch the broker has already forgotten, and duplicate detection would fail. That's why idempotence requires max.in.flight.requests.per.connection ≤ 5 (default 5), plus retries > 0 and acks=all. And the producer docs promise that with idempotence on, ordering is preserved for any allowed value. The broker's refusal to accept sequence gaps is what makes that possible: a batch can't jump ahead of an earlier one that has to be retried.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic idem-demo --partitions 1
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic idem-demo
> a
> b
bin/kafka-dump-log.sh --files /tmp/kraft-combined-logs/idem-demo-0/00000000000000000000.log

The idempotent producer from Part II solved one problem: a retry of the same batch to the same partition doesn't create a duplicate. That's it. It says nothing about two different partitions, nothing about a producer that crashes and restarts, and nothing about the consumer offset you want to move at the same time as you write your output.

Think about the classic "read → process → write" loop. You read record 41 from orders, write a result to invoices, then commit offset 42. Crash between the write and the commit, and on restart you read 41 again and write a second invoice. Commit first, crash before the write, and the invoice is never written. Both orders are wrong. You need the output write and the offset move to be one atomic act.

Kafka's answer: write everything immediately, decide visibility later. The records go into the logs straight away, flagged as "part of transaction X". When the producer commits, a tiny control record — a commit or abort marker — is appended to every partition that took part. Readers that ask for isolation.level=read_committed hold back transactional records until they see the marker, and skip the ones that end up aborted. In ledger terms: the clerk writes in pencil-grey ink, and at the end a notary stamps "VALID" or "VOID" in every book touched.

The cast

A transactional producer is an idempotent producer with a name. You set transactional.id (setting it implies enable.idempotence). That name is stable across restarts — that's the whole point. It's hashed onto one partition of the internal topic __transaction_state (50 partitions by default, replication factor 3, min ISR 2), and the leader of that partition acts as the producer's transaction coordinator. The coordinator keeps the transaction's state machine and writes every state change to __transaction_state, so a new coordinator can pick up after a failover.

The API is five calls: initTransactions() once, then per transaction beginTransaction(), send()…, optionally sendOffsetsToTransaction(offsets, consumer.groupMetadata()), then commitTransaction() or abortTransaction(). On the consumer side you set isolation.level=read_committed and enable.auto.commit=false — the offsets travel inside the transaction instead.

What happens on commit

The coordinator's state machine moves Empty → Ongoing → PrepareCommit → CompleteCommit (or PrepareAbort → CompleteAbort). The moment PrepareCommit is durably written to __transaction_state, the decision is final — even if the coordinator dies, the next one will finish the job. Then it sends WriteTxnMarkers requests to the leaders of every partition in the transaction (including the __consumer_offsets partition holding your group's offsets). Once all markers are written, it records CompleteCommit.

On the reading side, every partition tracks a last stable offset (LSO): the point before which every transactional record has been decided. A read_committed consumer only gets records below the LSO, and the fetch response carries a list of AbortedTransactions so the client can drop records from aborted ones. That's why one long-running open transaction stalls every read_committed consumer of that partition — even consumers that don't care about that producer. Records behind the open transaction are withheld until it finishes.

Zombie fencing — why the id must be stable

Imagine instance A of your service stalls in a long GC pause. The platform thinks it's dead and starts instance B with the same transactional.id. B calls initTransactions(); the coordinator bumps the producer epoch for that id and aborts whatever A left open. When A wakes up and tries to write with the old epoch, the partition leader rejects it with INVALID_PRODUCER_EPOCH (surfaced as InvalidProducerEpochException, which asks the app to abort); as soon as A talks to the coordinator again — to commit or abort — it gets the fatal ProducerFencedException and must close. The epoch is the notary's seal: only the newest seal is honoured. Without a stable id there's nothing to fence on, and A and B would happily write duplicates side by side.

KIP-890 and transaction version 2

The original protocol had a gap: a produce request that arrived late (say, a retry delayed in the network) could land after a transaction finished and silently become part of the next one, or linger as a "hanging" transaction that pins the LSO forever. KIP-890 ("transactions server-side defense") closes this in two steps. First, partition leaders verify with the coordinator that a partition really is part of an ongoing transaction before accepting transactional data (transaction.partition.verification.enable, default true). Second, with transaction.version=2 (TV2), the producer epoch is bumped at the end of every transaction, so a straggler from transaction N carries a stale epoch and is rejected by construction.

TV2 also trims round trips. The client no longer sends AddPartitionsToTxn itself: the partition leader adds the partition to the transaction on the server side while handling the produce request. sendOffsetsToTransaction likewise skips the separate AddOffsetsToTxn call and goes straight to TxnOffsetCommit. The server side has been on since 4.0 and is controlled by the transaction.version feature; 4.x producers switch to it at the start of their next transaction once they learn the cluster supports it — no restart needed. Downgrades are safe too.

bin/kafka-features.sh --bootstrap-server localhost:9092 describe
bin/kafka-transactions.sh --bootstrap-server localhost:9092 list
bin/kafka-transactions.sh --bootstrap-server localhost:9092 describe --transactional-id billing-1
# watch only committed data:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic invoices \
  --isolation-level read_committed --from-beginning

Time-based retention is blunt: after seven days the whole segment goes, whether or not it holds the only copy of a customer's current address. For a stream of state — "account 123 now has email X" — you want something smarter: forget the superseded entries, keep the latest one for every key, forever. That's cleanup.policy=compact.

In the ledger archive, a compacted book is one where an archivist periodically recopies old pages, leaving out every line that a later line for the same account has overruled. The crucial detail: the line numbers are copied too. Line 38 stays line 38, even if lines 36 and 37 were dropped. Offsets never change; compaction only creates gaps.

Head, tail and the cleaner

A compacted log has two regions. The tail (older, already cleaned) holds at most one record per key. The head (newer, "dirty") is an ordinary log with dense offsets. The log cleaner — a pool of background threads, log.cleaner.threads, default 1 — picks the log with the highest dirty ratio, builds a compact map of key → last offset for the dirty section, and recopies segments from the beginning, dropping every record whose key appears again later. Cleaned segments are swapped in as it goes, so the extra disk needed is about one segment, not a copy of the log.

Answer to the brain question, part one: every offset stays a valid position. Reading from a compacted-away offset simply returns the next record that still exists — the docs' example is that a read starting at 36, 37 or 38 all begin at 38.

Tombstones: deleting in an append-only world

To delete a key you append a record with that key and a null value — a tombstone. The cleaner removes every earlier record for the key. The tombstone itself must linger for a while, or consumers that are rebuilding state would never learn the key was deleted. How long? delete.retention.ms, default 24 hours. In the v2 batch format the cleaner stamps a delete horizon into the batch the first time it sees the tombstone (now + delete.retention.ms) and drops the tombstone on a later pass after that time.

What compaction promises (and doesn't)

From the design docs: a consumer that keeps up with the head sees every record; ordering is never changed; offsets never change; a consumer reading from the start sees at least the final value of every key. What it does not promise is "exactly one record per key right now". The active segment is never cleaned, nothing past the last stable offset is cleaned, and records younger than min.compaction.lag.ms are protected. So a compacted topic can contain duplicates at any time — your consumer must treat it as "latest wins", never as a unique-key table.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic accounts \
  --partitions 1 --config cleanup.policy=compact --config segment.ms=10000 \
  --config min.cleanable.dirty.ratio=0.01 --config delete.retention.ms=60000

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic accounts \
  --reader-property parse.key=true --reader-property key.separator=: \
  --reader-property null.marker=NULL
> A:1
> B:1
> A:2
> B:NULL        # tombstone for B

bin/kafka-dump-log.sh --files /tmp/kraft-combined-logs/accounts-0/00000000000000000000.log --print-data-log

Every produce, fetch, heartbeat and metadata call is a request that travels through the same assembly line inside the broker. When latency spikes, "Kafka is slow" is useless; "produce requests spend 900 ms in the request queue" tells you exactly which part of the line is jammed. This chapter gives you that vocabulary.

Picture the archive branch's front desk. A doorman (the acceptor) lets visitors in and hands each to one of a few counter clerks (network processors). Clerks don't do the work — they read the form and drop it into an in-tray (the request queue). Behind the scenes, archivists (I/O handler threads) take forms from the tray, write in the ledgers or look things up, and put answers in each clerk's out-tray (response queue). Clerks hand the answers back. And some forms can't be answered yet — "tell me when three branches have a copy" — so they go to a waiting room: purgatory.

The stages

1. Acceptor. One thread per listener accepts TCP connections and assigns each to a processor. The docs describe the model as "a single acceptor thread and N processor threads which handle a fixed number of connections each".

2. Network processors (num.network.threads, default 3 — per listener). Each runs a non-blocking selector, reads complete requests off its sockets, and puts them in the shared request queue. After reading a request from a connection it mutes that connection until the response is sent, which is how Kafka keeps per-connection ordering: one request per connection is being processed at a time on the broker.

3. Request queue (queued.max.requests, default 500). When full, network threads block — they stop reading new requests. This is back-pressure, deliberately.

4. I/O handler threads (num.io.threads, default 8). They run KafkaApis.handle: validate, append to the log (page cache), or read for a fetch. This is where disk, CPU for decompression/validation, and lock contention show up.

5. Purgatory. If a request can't be completed right now, it's parked as a delayed operation and the I/O thread moves on. A produce with acks=all waits in the produce purgatory until the in-sync followers have fetched the data (or the request's timeout — the producer's request.timeout.ms, default 30 s — expires). A fetch waits in the fetch purgatory until fetch.min.bytes (default 1) are available or fetch.max.wait.ms (default 500) passes. That's how long-polling works without burning a thread per waiting consumer.

6. Response queue → processor → socket. Each processor has its own response queue; it writes the response and unmutes the connection.

Reading the latency breakdown

For each request type (Produce, FetchConsumer, FetchFollower, …) the broker publishes TotalTimeMs, split into:

The monitoring docs suggest both idle-percent gauges stay "ideally > 0.3". RequestChannel's RequestQueueSize and the purgatory gauges (PurgatorySize for Produce and Fetch) complete the picture — the docs note the produce purgatory is "non-zero if ack=-1 is used" and the fetch one's "size depends on fetch.wait.max.ms in the consumer".

bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers \
  --entity-default --alter --add-config num.io.threads=16
bin/kafka-jmx.sh --jmx-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi \
  --object-name kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent

Part II showed the happy path of replication: followers fetch, the high watermark (HW) advances, committed records are safe. This chapter is about the unhappy path: a leader dies holding records its followers never got, a new leader writes different records at the same offsets, and the old leader comes back. Two photocopies of the same ledger book now disagree about what's on line 6. Somebody has to tear out pages — the question is who, and how many.

Why the high watermark wasn't enough

Early Kafka truncated a restarting follower to its own HW. KIP-101 describes what went wrong. Because a follower learns the new HW only on its next fetch, its local HW lags. If it truncates to that stale HW and then becomes leader before re-fetching, records that were already committed vanish. And after a multi-broker power loss, replicas could end up with different records at the same offsets — divergent lineages that no amount of fetching would repair.

Leader epochs: numbering the reigns

The fix: every time the controller elects a new leader for a partition, the leader epoch goes up (the KRaft controller also bumps it on some other partition changes, such as certain reassignments), and every batch the leader writes is stamped with it (the partitionLeaderEpoch field in the batch header you met in Part II). Each replica also keeps a small leader epoch cache — a list of epoch → first offset of that epoch, persisted in the partition directory as leader-epoch-checkpoint.

Now a returning follower doesn't guess. It asks the leader: "the last epoch I have is 1 — where did epoch 1 end in your log?" The leader looks in its cache: epoch 2 started at offset 6, so epoch 1 ended at 6. The follower truncates everything from offset 6 up and fetches the leader's version. Only records that genuinely diverge are removed. In the ledger analogy: each reign of a head archivist uses a different ink colour, and the register at the front of the book says on which line each colour started. Comparing registers tells you exactly where two copies part ways.

Unclean leader election: availability over truth

If every in-sync replica is gone, Kafka faces the dilemma the design docs spell out: wait for an ISR member to come back (maybe never), or promote whatever replica is alive, even if it's missing committed records. unclean.leader.election.enable (default false) chooses between them. When it's on and an out-of-sync replica becomes leader, its log becomes the source of truth: committed records it lacked are gone, and the other replicas truncate to match when they return. Consumers that had already read past that point detect it — with no automatic reset policy the consumer throws LogTruncationException with the first diverging offset.

In KRaft, turning the flag on dynamically doesn't act instantly: the controller checks leaderless partitions every unclean.leader.election.interval.ms (default 5 minutes). To force it now, the config docs point to kafka-leader-election.sh with the unclean option.

ELR: electing safely outside the ISR (KIP-966)

There's a middle ground the ISR alone can't see. With min.insync.replicas enforced, the HW can't advance while the ISR is smaller than min ISR. So a replica that was in the ISR when it dropped below min ISR still has every committed record — nothing new could be committed after it left. Such replicas are safe leaders even though they're not in the ISR any more.

Eligible Leader Replicas track exactly those. The KRaft controller stores them in the partition's metadata record (EligibleLeaderReplicas, plus LastKnownElr). When the ISR shrinks below min ISR, former ISR members move into the ELR — except replicas that had an unclean shutdown, which may have lost unflushed data. When the ISR is back at or above min ISR, the ELR is cleared. Leader election order becomes:

1. If the ISR is not empty, pick from the ISR.
2. Else, if the ELR is not empty, pick an unfenced ELR member.
3. Else, the last known leader if it's unfenced (roughly the pre-4.0 behaviour when all replicas were offline).

ELR arrived in 4.0 behind the eligible.leader.replicas.version feature (off by default in 4.0) and is enabled by default on new clusters since 4.1. Enabling it has side effects on min.insync.replicas: it gets a cluster-level value, the broker-level setting is removed and can't be altered, and changing it at cluster or topic level clears ELR state.

bin/kafka-features.sh --bootstrap-server localhost:9092 describe      # eligible.leader.replicas.version
bin/kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature eligible.leader.replicas.version=1
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders   # shows Elr / LastKnownElr
bin/kafka-leader-election.sh --bootstrap-server localhost:9092 \
  --election-type unclean --topic orders --partition 0

A Kafka broker's disk does two jobs at once: it's the hot write path, and it's the long-term archive. Most reads are tail reads served from the page cache; old data is touched rarely, for backfills and recovery — the docs make exactly this point. Yet keeping 90 days of data means buying 90 days of fast local disk, replicated three times, and moving all of it whenever you add a broker or replace a failed one.

Tiered storage splits the two jobs. The local tier is the log you know. The remote tier is an external store (the docs mention HDFS or S3) that receives completed segments. In the archive: the branch keeps the last few volumes on the reading-room shelves; finished volumes are shipped to a central warehouse, and a card index says which warehouse box holds which lines.

The moving parts

RemoteStorageManager (RSM) — a plugin interface that copies segment data and its indexes to the remote store, fetches them back, and deletes them (copyLogSegmentData, fetchLogSegment, fetchIndex, deleteLogSegmentData). Kafka ships no production implementation: you configure a vendor or open-source plugin with remote.log.storage.manager.class.name and .class.path.

RemoteLogMetadataManager (RLMM) — keeps strongly consistent metadata about remote segments. The default, TopicBasedRemoteLogMetadataManager, stores it in the internal topic __remote_log_metadata; you must set remote.log.metadata.manager.listener.name so its internal clients know which listener to use.

RemoteLogManager (RLM) — the broker component that runs per-partition tasks: a copy task on the leader, an expiration task that deletes remote segments past retention, and a follower task. Segments eligible for copying are those that are not the active segment and whose end offset is below the last stable offset — "as remote storage should contain only committed/acked messages". That answers the brain question.

Two retention clocks

A tiered topic (remote.storage.enable=true) has two retention settings. local.retention.ms / local.retention.bytes decide how long segments stay on the broker; retention.ms / retention.bytes decide how long the data exists at all (in the remote tier). The local values default to -2, meaning "use the full retention" — so you must set them explicitly to get any benefit, and they must not exceed the total retention. Crucially, a local segment becomes eligible for deletion only after it has been uploaded.

Reading old data

Consumers don't change at all. When a fetch asks for an offset that's no longer local, the leader serves it from the remote store: the read runs on the remote-reader thread pool (remote.log.reader.threads, default 10) and the fetch waits in a dedicated DelayedRemoteFetch purgatory (up to remote.fetch.max.wait.ms, default 500). Expect higher latency for these cold reads and budget for the object store's request costs.

bin/kafka-topics.sh --create --topic tieredTopic --bootstrap-server localhost:9092 \
  --config remote.storage.enable=true --config local.retention.ms=1000 --config retention.ms=3600000 \
  --config segment.bytes=1048576 --config file.delete.delay.ms=1000
bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic tieredTopic \
  --num-records 1200 --record-size 1024 --throughput -1
ls /tmp/kafka-remote-storage/kafka-tiered-storage/        # uploaded .log/.index/.timeindex/... files
bin/kafka-console-consumer.sh --topic tieredTopic --from-beginning --max-messages 1 \
  --bootstrap-server localhost:9092 --formatter-property print.offset=true

Consumer groups give you ordering and scale by handing each partition to exactly one member. That's perfect for ordered streams — and awkward for a work queue. Want 50 workers on a topic with 8 partitions? Tough: 42 sit idle. One slow record blocks everything behind it in that partition. And there's no built-in notion of "this record failed three times, stop trying".

Share groups are a second kind of group for exactly those workloads. They were previewed in 4.1 and declared production-ready in 4.2. The design docs list the differences: consumers cooperatively consume and a partition may be assigned to several consumers; there can be more consumers than partitions; records are acknowledged individually (optimised for batches); and delivery attempts are counted, so unprocessable records can be handled automatically. The price: no ordering guarantee between records handed to different consumers.

In the archive analogy: instead of assigning each clerk a whole book, the clerks share the book, and there's a sign-out sheet next to it. A clerk signs out a few lines, works on them, and ticks them "done" — or hands them back. If a clerk vanishes, their sign-outs expire and the lines go back into the pool. The sheet also counts how often each line was signed out; after too many attempts the line is stamped "unprocessable" and set aside.

Acquisition locks and the record lifecycle

When a share consumer fetches, the broker hands it available records and marks them acquired with a time-limited acquisition lock — 30 s by default (share.record.lock.duration.ms group config, broker default group.share.record.lock.duration.ms). While acquired, no other member of the group gets them. The holder then does one of:

• ACCEPT — processed successfully; the record becomes acknowledged.
• RELEASE — give it back for another delivery attempt.
• REJECT — unprocessable; archive it, never deliver again.
• RENEW — still working on it; extend the lock.
• nothing — the lock expires and the record becomes available again.

Each acquisition bumps the record's delivery count. When a record would become available again but has already reached the limit (share.delivery.count.limit, broker default group.share.delivery.count.limit = 5), it's archived instead. That's your built-in poison-pill guard. In 4.3.1, archived simply means "no more delivery attempts": dead-letter queues for share groups (KIP-1191) exist only as groundwork in the source — a ShareGroupDLQ interface with a no-op implementation and the ARCHIVING record state — with no configuration and nothing actually written to a DLQ topic yet.

The broker also caps in-flight work per partition: group.share.partition.max.record.locks (default 2000 acquired records). Hit it and fetches return nothing until locks are released or time out.

Where the state lives

For each share-partition (group × topic-partition) the partition leader keeps the in-flight window between the share-partition start offset (SPSO) — everything below is done — and the share-partition end offset (SPEO). The durable version of this is written by the share coordinator to the internal topic __share_group_state (50 partitions, RF 3, min ISR 2 by default): a snapshot has the start offset plus state batches of (firstOffset, lastOffset, deliveryState, deliveryCount), where the state is 0 Available, 2 Acked or 4 Archived. Membership and assignment are handled by the group coordinator, with a heartbeat protocol (ShareGroupHeartbeat) similar to KIP-848.

bin/kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic jobs --group workers
bin/kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic jobs --group workers --reject
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type groups --entity-name workers \
  --add-config share.delivery.count.limit=3,share.auto.offset.reset=earliest
bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group workers --members
bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group workers --offsets

Kafka Streams is a library, not a cluster. You write a topology — "read orders, group by customer, count, write order-counts" — and run as many copies of your app as you like. The library splits the work, keeps local state, survives crashes and can give you exactly-once, using nothing but the Kafka features from this and the previous parts: consumer groups, compacted topics and transactions.

Archive analogy: each clerk keeps a private notebook (a state store) with running totals. Every change to the notebook is also written as a line in a dedicated ledger book in the archive (the changelog topic, compacted). If the clerk quits, a replacement rebuilds the notebook by reading that ledger — or, faster, takes over from a colleague who has been keeping a shadow copy all along (a standby replica).

Topology, sub-topologies, tasks

Your topology is grouped into sub-topologies: sets of connected processors that are linked to each other only through Kafka topics — typically a repartition topic, created when data must be re-keyed. Each sub-topology is instantiated once per input partition as a task, named <subtopology>_<partition> — so 0_2 is sub-topology 0 on partition 2. The docs stress that "the assignment of partitions to tasks never changes": a task is the fixed unit of parallelism, and the maximum parallelism is bounded by the number of input partitions. Threads (num.stream.threads, default 1) and instances only decide where tasks run.

State stores and changelogs

A stateful task owns a local store (RocksDB or in-memory). For each store Kafka Streams maintains a changelog topic named <application.id>-<storeName>-changelog, with one partition per task, and log compaction enabled (windowed stores use delete,compact). Repartition topics use delete with infinite retention; Streams purges them itself once records are consumed. When a task moves to another instance, its store is restored by replaying its changelog partition before processing resumes — that restore is usually the dominant cost of a rebalance.

Standby replicas and warm-up

num.standby.replicas (default 0) keeps shadow copies of each task's store on other instances by continuously reading the changelog. On failover the task is assigned where a standby exists, so only a small tail needs replaying. Since 2.6 the classic assignor also only assigns a stateful task to an instance with a caught-up copy if one exists (within acceptable.recovery.lag, default 10000 records), warming others up in the background with up to max.warmup.replicas (default 2) extra "warm-up" tasks.

Exactly-once v2

processing.guarantee is at_least_once by default; exactly_once_v2 turns on the Chapter-13 machinery end to end: output records, changelog writes and input offsets are committed in one transaction per commit, consumers read with read_committed, and commit.interval.ms drops from 30000 to 100 ms by default. The docs warn that EOS with a replication factor below 3 "effectively voids EOS" — use RF 3 and min.insync.replicas=2 for all involved topics.

The Streams rebalance protocol (KIP-1071)

Classically, Streams runs on a consumer group with a client-side assignor (StreamsPartitionAssignor) that computes task assignments during a group-wide rebalance. KIP-1071 applies the KIP-848 idea to Streams: the app registers a dedicated streams group, sends its topology to the broker in StreamsGroupHeartbeat (sub-topologies, source, changelog and repartition topics), and the broker computes active and standby task assignments continuously and incrementally (the heartbeat has a WarmupTasks field, but warm-up tasks are not supported yet). The core feature set is production-ready in 4.2, and the feature is enabled by default on new 4.2 clusters (streams.version=1); clients opt in with group.protocol=streams.

The docs list what's not yet supported: static membership, significant topology changes (need a new group), the high-availability assignor (so no warm-up tasks or rack awareness — only a sticky assignor), regex subscriptions, online migration, and supplying the main consumer via a custom KafkaClientSupplier. Migration is offline only; the upgrade notes recommend against migrating classic → streams on 4.2.0 because of a broker bug fixed in 4.2.1.

bin/kafka-features.sh --bootstrap-server localhost:9092 describe          # streams.version
bin/kafka-streams-groups.sh --bootstrap-server localhost:9092 --list
bin/kafka-streams-groups.sh --bootstrap-server localhost:9092 --describe --group counter-app --members --verbose
bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type groups \
  --entity-name counter-app --add-config streams.num.standby.replicas=1

Up to now your archive served one well-behaved city department. Real clusters are shared: payments, search, analytics and one intern's load test all write to the same branches. The official docs call the result "noisy neighbors": one tenant saturates the network or the request threads, and everyone else's latency goes through the roof.

In the ledger-archive picture, a quota is a counter rule: "each patron group may carry out at most X pages per second at this branch". A patron who goes faster isn't turned away. The clerk just says "sit down for a moment" and doesn't serve them until they're back under the limit. That is exactly how Kafka works: quotas throttle by delaying, not by rejecting (with one exception, below).

What can you limit?

Kafka has two families of client quotas. Network bandwidth quotas are byte-rate thresholds: producer_byte_rate and consumer_byte_rate, in bytes/sec. Request rate quotas (request_percentage) cap the share of broker network and I/O thread time a group may use. A quota of n% means n% of one thread, so the total capacity is (num.io.threads + num.network.threads) × 100%. On top of those, controller_mutation_rate limits how fast a client may create topics, create partitions and delete topics, counted in partitions.

There are also broker-side connection limits, which aren't tied to a client identity: max.connections.per.ip, max.connection.creation.rate, and a per-IP connection_creation_rate quota. All of them default to "unlimited" (Integer.MAX_VALUE).

Who is "a client"? Users, client-ids and precedence

A quota applies to a group of clients. The identity is the authenticated user principal, the application-chosen client.id, or the pair (user, client-id). Every connection in a quota group shares the quota. If (user=test-user, client-id=test-client) gets 10 MB/s, all producer instances with that pair split those 10 MB/s together. For a given connection, the most specific matching quota wins, in this order:

1. matching user and client-id
2. matching user, default client-id
3. matching user
4. default user, matching client-id
5. default user, default client-id
6. default user
7. matching client-id
8. default client-id

Quota overrides are written to the KRaft metadata log. Every broker reads them and they take effect immediately, so you never need a rolling restart to re-tune a tenant. By default there is no quota at all: clients get unlimited throughput until you set one.

How throttling actually works

When a group goes over its quota, the broker computes how long the client must stay silent for its average rate to drop back under the limit. It returns the response immediately, with that delay in the ThrottleTimeMs field (a fetch response carries no data in that case). Then it mutes the channel: it stops reading from that client's socket until the delay is over. A modern client sees a non-zero throttle time and holds back on its own. An old or misbehaving client that keeps sending anyway still gets nowhere, because its requests sit unread in a muted socket. The docs describe this as blocking the client "from both sides".

Why per broker and not cluster-wide? A cluster-wide limit would need the brokers to share their quota usage with each other in real time. The design docs say bluntly that this "can be harder to get right than the quota implementation itself!" So a 10 MB/s quota means 10 MB/s at each broker. A tenant whose partitions sit on six brokers can move up to six times that in total.

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --add-config 'producer_byte_rate=1048576' --entity-type clients --entity-name loadtest
bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe \
  --entity-type clients --entity-name loadtest
bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 200000 --record-size 1000 --throughput -1 \
  --command-property client.id=loadtest

throttleTimeMs = (observed − bound) / bound × windowSize

Every Kafka performance "secret" you've read on a forum is one of four trades: latency for throughput (batch more), CPU for bytes (compress), safety for speed (acknowledge less), and memory and file handles for parallelism (more partitions). This chapter gives you the knobs for each trade, their real defaults, and a method for turning them without guessing.

In the archive, the clerks are fastest when patrons bring bundles of pages, not single sheets. They're slower when every bundle must be photocopied at two other branches before the patron leaves. And they're slowest of all when their desks are cluttered with ten thousand half-used ledger books. Keep those three images in mind; they explain almost every row below.

The method (before any knob)

1. Reproduce with the perf tools: kafka-producer-perf-test.sh and kafka-consumer-perf-test.sh give you a baseline you can compare against.
2. Find the bottleneck, don't assume it. The broker's request metrics split TotalTimeMs into RequestQueueTimeMs, LocalTimeMs, RemoteTimeMs, ResponseQueueTimeMs and ResponseSendTimeMs (chapter 15's pipeline). High queue time means the threads are busy. High local time means the disk or log append is slow. High remote time on produce means you're waiting for followers (acks=all).
3. Check thread idleness: NetworkProcessorAvgIdlePercent and RequestHandlerAvgIdlePercent should ideally stay above 0.3 according to the monitoring docs.
4. Change one knob, re-run, and keep a log. Two knobs at once teach you nothing.

Producer knobs

Consumer knobs

Broker knobs

Partitions: the parallelism you pay for

The partition count is the maximum parallelism of a consumer group, and each partition must fit entirely on one server. So more partitions means more parallel consumers and spreads the load across more brokers. But each partition's segments cost file descriptors and memory map areas. The OS docs recommend at least 100000 file descriptors for the broker as a starting point. Each segment uses 2 map areas (index and time index), and the docs warn that 50000 partitions on one broker means 100000 map areas, which "likely" crashes it with OutOfMemoryError (Map failed) at a typical vm.max_map_count of about 65535. Remember too that you can add partitions but never remove them, and adding them remaps keys.

OS and disks

Kafka leans on the page cache. The hardware docs give a back-of-the-envelope memory estimate: buffer about 30 seconds of writes, write_throughput × 30. Keep Kafka data on dedicated drives (not shared with application logs), mount with noatime, and prefer XFS. In the docs' comparison XFS gave "160ms vs. 250ms+" request local time against the best EXT4 configuration. Keep the default flush settings (no application fsync): durability comes from replication, and forcing fsync gives the OS less room to reorder writes and adds latency.

bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 500000 --record-size 200 --throughput -1
bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic perf \
  --num-records 500000 --record-size 200 --throughput -1 \
  --command-property linger.ms=50 batch.size=262144 compression.type=lz4

At 3 a.m. nobody reads the design docs. You get a symptom ("lag is climbing", "producers throw NotEnoughReplicasException") and you need to go from that symptom to a cause and a safe fix in minutes. This chapter is that map. It leans on everything you already know: ISR and high watermark (ch. 10), rebalances (ch. 11), idempotence (ch. 12), the request pipeline (ch. 15), leader epochs and ELR (ch. 16), KRaft (ch. 9).

Archive version: an inspector walks in, checks a few gauges on the wall (metrics), asks the front desk a few questions (CLI tools), and only then moves furniture. The worst incidents come from moving furniture first, like adding consumers, deleting data or lowering min.insync.replicas, before knowing which gauge is red.

The symptom table

The dashboard: metrics worth an alert

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-min-isr-partitions
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

Replication inside a cluster (ch. 5, 10) keeps photocopies in other branches of the same archive, coordinated by one city council. Geo-replication is different: two independent archives in two cities, each with its own council, its own offsets and its own consumer groups. Between them sits a copying office that reads one archive and writes into the other. In Kafka that office is MirrorMaker 2 (MM2). The original MirrorMaker 1 was removed in 4.0.

People use it for disaster recovery, for feeding edge clusters into a central one, for separating production from testing, for cloud migrations, and for legal requirements about where data lives. MM2 replicates topic data and topic configurations, consumer group offsets and ACLs. It preserves partitioning and discovers new topics and partitions automatically.

Built from Connect parts

MM2 is not a separate engine. It's a set of Kafka Connect source connectors, which is why it inherits Connect's scaling (run more MM2 processes and they share the work) and Connect's metrics:

• MirrorSourceConnector: "Replicate data, configuration, and ACLs between clusters." Consumes from the source, produces to the target, and writes offset syncs that pair upstream offsets with downstream ones.
• MirrorCheckpointConnector: "Replicate consumer group state between clusters. Emits checkpoint records." It translates each group's committed offsets into target offsets.
• MirrorHeartbeatConnector: "Emits heartbeats to Kafka." A tiny heartbeats topic that proves the flow is alive and lets tools trace the replication topology.

Flows, aliases and remote topics

You name clusters with aliases and enable directional flows source->target. A flow is off until you set {source}->{target}.enabled = true. Topologies are just sets of flows: active/passive A->B, active/active A->B, B->A, aggregation A->K, B->K, fan-out K->A, K->B.

clusters = us-west, us-east
us-west.bootstrap.servers = broker1-west:9092
us-east.bootstrap.servers = broker3-east:9092
us-west->us-east.enabled = true
us-west->us-east.topics = orders.*

With the DefaultReplicationPolicy, a replicated ("remote") topic is renamed {source}.{topic}, so orders from us-west becomes us-west.orders in us-east. That prefix is not decoration. It stops two clusters from writing into the same topic-partition, and it's how active/active avoids infinite loops: MM2 knows us-west.orders came from us-west and won't send it back. IdentityReplicationPolicy keeps names unchanged, but its javadoc warns that MM2 then "is not able to prevent cycles", so use it only for one-way topologies.

Offsets don't survive the trip

This is the single most important idea in the chapter. The target cluster assigns its own offsets. If replication starts when the source partition begins at 1000, the remote partition begins at 0. Filtered or aborted records widen the gap further. A consumer group's committed offset "1007" means nothing in the other city.

So MM2 keeps a translation table. The source connector writes offset syncs (upstream ↔ downstream pairs) to mm2-offset-syncs.<target>.internal, which by default lives on the source cluster (offset-syncs.topic.location defaults to source). The checkpoint connector reads each group's committed offsets, translates them, and writes checkpoints to <source>.checkpoints.internal on the target. Applications can read them via RemoteClusterUtils.translateOffsets(...). Alternatively, set sync.group.offsets.enabled=true (default false) and MM2 writes the translated offsets into the target's __consumer_offsets, "as long as no active consumers in that group are connected to the target cluster".

Translation is deliberately conservative. MM2 only knows the sync points exactly. For a group that's past the nearest sync, it translates to "sync's downstream offset + 1". The code comment explains why: overestimating could skip records and lose data. The result is that after a failover you re-read some records, never skip them. Plan for at-least-once and idempotent processing.

Upgrading a Kafka cluster changes two separate things, and most upgrade mistakes come from mixing them up. The first is the software version: the jars on each node. The second is the feature levels the whole cluster has agreed to use: metadata.version and a handful of feature flags, stored as records in the KRaft metadata log.

In the archive, the software is each branch's clerks and furniture. The feature level is the edition of the city rulebook that the council has formally adopted. You retrain the clerks branch by branch, and the old rulebook stays in force while you do. Only when every branch can handle the new edition does the council vote it in. After that vote, some new rules write records the old clerks can't read, so there's no going back.

The rolling upgrade, per the 4.3 docs

1. Check the preconditions. 4.x is KRaft-only. Brokers must already run KRaft with software and metadata at least 3.3.x. KRaft clusters older than 3.3.x should go to 3.9.x first. ZooKeeper clusters must migrate to KRaft using a bridge release, and "the last bridge release is Kafka 3.9".
2. Roll the binaries: stop one broker, update the code, restart it, and repeat. The cluster still speaks the old metadata.version, so a mixed-version cluster is fine.
3. Verify behavior and performance at your leisure. This is your cheap rollback window: just reinstall the old binaries.
4. Finalize: bin/kafka-features.sh --bootstrap-server localhost:9092 upgrade --release-version 4.3

Finalizing is where the one-way doors are. Every MetadataVersion carries a boolean that says whether it changes the metadata format. IBP_4_3_IV0(30, "4.3", "IV0", true) does, so after finalizing 4.3 you cannot downgrade the metadata. IBP_4_2_IV1(29, "4.2", "IV1", false) doesn't, which is why the 4.2 docs say downgrade is supported. A downgrade is possible only if no version between current and target has metadata changes.

Feature flags: one switch per protocol

Since KIP-584, metadata.version isn't the only versioned feature. Each big protocol change has its own flag, so it can be enabled, and in most cases downgraded, independently. --release-version X sets all features to their defaults for that release. The tool's own help says "3.9-IV0 will set metadata.version=21 and kraft.version=1". --feature name=level changes just one.

bin/kafka-features.sh --bootstrap-server localhost:9092 describe
bin/kafka-features.sh --bootstrap-server localhost:9092 version-mapping --release-version 4.1
bin/kafka-features.sh --bootstrap-server localhost:9092 upgrade --release-version 4.3 --dry-run

What 5.0 will take away

5.0 isn't out, but the 4.x notes already name what it removes. Clean these up now: group.coordinator.rebalance.protocols (in 5.0 all protocols are always on, and group.version/streams.version/share.version alone control them), the kafka-streams-scala library, the CLI options deprecated by KIP-1147 (--producer-props, --consumer.config, --property…, replaced by --command-property, --command-config, --formatter-property, --reader-property), --max-partition-memory-bytes in the console producer, MX4J support, the misspelled PARTITIONER_ADPATIVE_PARTITIONING_ENABLE_CONFIG constant, the tag-less app-info metric names, and PrincipalConnectorClientConfigOverridePolicy (the new allowlist policy becomes the default).

Every Kafka application eventually meets three kinds of trouble: the network or cluster hiccups (a leader moves, an ISR shrinks), the data is bad (a record your code can't parse or can't process), or your own process misbehaves (too slow, killed mid-batch, fenced by a newer instance). Each needs a different reaction, and the most common production bug is treating all three the same way — usually with a try { … } catch (Exception e) { retry(); }.

The producer: retriable vs non-retriable

The Java client encodes the first split directly in its exception hierarchy. Every exception that extends org.apache.kafka.common.errors.RetriableException is, in the client's own words, "a transient exception that if retried may succeed". NotEnoughReplicasException, TimeoutException and the metadata errors (InvalidMetadataException and friends, such as NotLeaderOrFollowerException) live there. Everything else — RecordTooLargeException, InvalidTopicException, AuthorizationException, UnknownServerException — is non-retriable: the Callback javadoc calls them "fatal, the message will never be sent".

Here is the part many people miss: the producer already retries retriable errors for you. retries defaults to Integer.MAX_VALUE, and the real budget is time, not attempts: delivery.timeout.ms (default 120 000 ms = 2 minutes) is "an upper bound on the time to report success or failure after a call to send() returns". It covers the time the record waits in the accumulator, the wait for the broker's acknowledgement and all retries. Between attempts the client backs off exponentially from retry.backoff.ms (100 ms) up to retry.backoff.max.ms (1000 ms). Each single request is bounded by request.timeout.ms (30 000 ms), and the docs say delivery.timeout.ms should be ≥ request.timeout.ms + linger.ms.

So when your callback receives an exception, the client has already done what it could. A TimeoutException in your callback is technically "retriable", but it usually means the whole two-minute budget is gone. The other case is that send() already waited max.block.ms for metadata or buffer space, and doSend() reports that timeout through the callback too. Re-sending it blindly in the callback is how you get unbounded memory growth and records reordered behind newer ones. The right reaction is almost always: record the failure somewhere durable (a local outbox, a log, a metric + alert) and let a human or a controlled process decide.

// Kafka clients 4.3.1 — a producer callback that classifies instead of blindly retrying
producer.send(record, (metadata, e) -> {
    if (e == null) return;                                // acknowledged
    if (e instanceof RetriableException) {
        // the client already retried until delivery.timeout.ms ran out
        outbox.park(record, e);                           // durable, replayable later
    } else if (e instanceof RecordTooLargeException) {
        deadLetters.record(record, e);                    // will never fit: don't retry
    } else {
        fatal.set(e);                                     // auth, unknown server error…: stop, alert
    }
});

The transactional and idempotent producer adds one more category. ProducerFencedException (a newer instance with the same transactional.id took over — Part III) extends ApplicationRecoverableException, whose javadoc says the error "is fatal to the producer, and the application needs to restart the producer after handling the error". The send() javadoc lists the transactional errors that abortTransaction() cannot fix — ProducerFencedException, OutOfOrderSequenceException, UnsupportedVersionException, AuthorizationException — and says "the only option left is to call close()". Any other failed transactional send surfaces at commitTransaction(); you abort and carry on.

The consumer: poison pills, lost commits, wake-ups

A poison pill is a record your consumer can't deserialize — someone wrote JSON into an Avro topic, or a producer shipped a new schema early. The consumer deserializes inside poll(), so the failure surfaces as a RecordDeserializationException thrown from poll(). Its message ends with the key sentence: "If needed, please seek past the record to continue consumption." The position does not move past the bad record by itself, so a loop that just logs and polls again hits the same record forever. Since Kafka 3.8 (KIP-1036) the exception carries everything you need to quarantine it: topicPartition(), offset(), origin() (KEY or VALUE), keyBuffer(), valueBuffer(), headers(), timestamp().

// Kafka clients 4.3.1 — quarantine a poison pill and move on
while (running) {
    ConsumerRecords<String, Order> records;
    try {
        records = consumer.poll(Duration.ofMillis(500));
    } catch (RecordDeserializationException e) {
        dlqProducer.send(new ProducerRecord<ByteBuffer, ByteBuffer>(
                "orders-dlq", null, e.keyBuffer(), e.valueBuffer(), e.headers()));
        consumer.seek(e.topicPartition(), e.offset() + 1);   // step over it
        continue;
    }
    for (ConsumerRecord<String, Order> r : records) process(r);   // must be idempotent
    consumer.commitSync();
}

The second classic is CommitFailedException. Its javadoc explains it exactly: the commit failed "with an unrecoverable error. This can happen when a group rebalance completes before the commit could be successfully applied" — the partitions may already belong to another member. The default message even names the usual cause: the time between poll() calls exceeded max.poll.interval.ms (default 300 000 ms), so fix it "either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records" (default 500). Do not retry the commit: the records you processed will be re-delivered to the new owner. That is why consumers must be idempotent.

RebalanceInProgressException from commitSync() is milder: the javadoc suggests completing the rebalance with poll() and reconsidering the commit afterwards, with the caveat that your assignment may have changed.

Finally, WakeupException is not an error at all — it's the only thread-safe way to interrupt a consumer. KafkaConsumer "is NOT thread-safe", except for wakeup(), which makes the blocked call throw WakeupException. The javadoc pattern: set a closed flag, call consumer.wakeup() from the shutdown hook, catch WakeupException in the poll loop, rethrow it only if you weren't closing, and close() in finally. Thread interrupts also work, but the javadoc discourages them "since they may cause a clean shutdown of the consumer to be aborted".

Dead-letter queues and idempotent consumers

Kafka's plain consumer has no built-in dead-letter queue (DLQ): a consumer group tracks one offset per partition, not a per-record fate. A DLQ in a consumer-group app is a pattern you build — a separate topic you produce the failed record to, with headers describing why, before you commit past it. Kafka Connect has one built in for sink connectors (errors.deadletterqueue.topic.name with errors.tolerance=all), and since 4.2 Kafka Streams has one too (errors.dead.letter.queue.topic.name, KIP-1034). Spring for Apache Kafka gives you one for listeners (next chapter). Share groups (Part III) take a different route: they count delivery attempts per record, and a rejected record gets no further delivery attempts.

A good DLQ record is replayable: the original bytes (not a toString()), the original topic/partition/offset, the exception class and message, a timestamp, and the consumer group. Someone will fix the bug and want to push those records back through — design for that day.

Because the consumer gives you at-least-once delivery (the KafkaConsumer javadoc: "each record will likely be delivered one time but in failure cases could be duplicated"), processing must tolerate duplicates. Common designs: a natural idempotent write (upsert by business key), a processed-ids table updated in the same database transaction as the effect, or storing the consumed offset alongside the result and seeking to it on assignment (the javadoc's "storing offsets outside Kafka" pattern). For Kafka-to-Kafka pipelines, transactions (Part III) make the read-process-write cycle exactly-once.

That dialogue is the central trade-off of this Part. Inside a partition, order is only preserved if you process strictly in sequence. Any retry that lets later records go first (a retry topic, a DLQ you replay later, a share group redelivery) breaks per-key order. Blocking retries keep order but stall every key in the partition and risk max.poll.interval.ms. Decide per topic which one you can afford.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ints --partitions 1
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ints
> hello
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ints --from-beginning \
  --formatter-property value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer

Spring for Apache Kafka wraps the clients you just learned in a listener container: it runs the poll loop, calls your @KafkaListener method, commits offsets, catches exceptions, retries, dead-letters and shuts down cleanly. The patterns from the last chapter are all there, already built. The catch is that its defaults were chosen for "works out of the box", not for "correct for your topic". This chapter is pinned to Spring for Apache Kafka 4.1.1 (2026-08-20). That release builds against kafka-clients 4.2.1 and Spring Framework 7.0.9 (from its gradle/libs.versions.toml), and its quick tour lists Java 17 as the minimum.

The container: threads, partitions, commits

A ConcurrentKafkaListenerContainerFactory builds a ConcurrentMessageListenerContainer per @KafkaListener. With concurrency n it creates n KafkaMessageListenerContainers, each with its own KafkaConsumer and thread — consumers are not thread-safe, so one consumer per thread. Kafka's group protocol then spreads the partitions. In the archive: n clerks from the same team, each holding whole ledger books. More clerks than books means idle clerks. The docs warn about a subtler version: with three topics of five partitions each and concurrency=15, you "see only five active consumers" under the classic RangeAssignor default. Size concurrency to your partition count (summed over instances) and check the assignor.

Spring sets enable.auto.commit to false unless you set it (since 2.3), and commits for you according to the container's AckMode. The default is BATCH.

New in 4.1: @KafkaListener(ackMode = "MANUAL") overrides the factory's mode per listener, so you no longer need a second factory bean just for that. With transactions, offsets are sent to the transaction instead, and semantics equal RECORD or BATCH depending on the listener type.

Blocking retries: DefaultErrorHandler

When the listener throws, the container calls a CommonErrorHandler, by default the DefaultErrorHandler. It remembers the failed record, seeks the partition back so the record (and everything after it) is redelivered by the next poll(), and waits according to a Spring BackOff. The default is FixedBackOff(0, 9): ten deliveries with no delay, then the record is logged at ERROR and skipped. That answers the brain-power question. Without a recoverer, the default ends by skipping and logging, which is quiet data loss.

Two built-in guard rails matter:

• Fatal exceptions skip retries. DeserializationException, MessageConversionException, ConversionException, MethodArgumentResolutionException, NoSuchMethodException and ClassCastException go straight to the recoverer on the first failure. Add your own with addNotRetryableExceptions(…), for example validation errors.
• Only RuntimeExceptions are handled. The docs warn that anything extending Error "bypass[es] the error handler entirely, causing the consumer to terminate immediately", and your health check may still say UP.

Blocking retries keep order: nothing behind the failing record in that partition is processed until it succeeds or is recovered. The price is a stalled partition and the poll-interval clock. The container sleeps between attempts, but the consumer must still poll() within max.poll.interval.ms (300 s by default) or it is kicked out of the group. For long back-offs, the docs point to ContainerPausingBackOffHandler, which "pauses the listener container until the back off time passes", for delays "longer than the max.poll.interval.ms".

Dead letters: DeadLetterPublishingRecoverer

Give the error handler a recoverer and exhausted records get somewhere to go. DeadLetterPublishingRecoverer publishes the failed record with a KafkaTemplate. By default it goes to <originalTopic>-dlt and the same partition, so the DLT "must have at least as many partitions as the original topic". (The suffix became -dlt in 3.3. Older apps used .DLT and must opt in to keep it.) Each dead letter carries headers such as KafkaHeaders.DLT_EXCEPTION_FQCN, DLT_EXCEPTION_MESSAGE, DLT_EXCEPTION_STACKTRACE, DLT_ORIGINAL_TOPIC, DLT_ORIGINAL_PARTITION, DLT_ORIGINAL_OFFSET and DLT_ORIGINAL_CONSUMER_GROUP, which is everything the last chapter said a replayable DLQ record needs.

// Spring for Apache Kafka 4.1.1 (Spring Framework 7)
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
import org.springframework.kafka.listener.*;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;

@Configuration
class KafkaErrorConfig {

    @Bean
    DeadLetterPublishingRecoverer recoverer(KafkaOperations<?, ?> template) {
        return new DeadLetterPublishingRecoverer(template);        // -> orders-dlt, same partition
    }

    @Bean
    DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer) {
        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);                            // 1 s, 2 s, 4 s, then DLT
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(IllegalArgumentException.class);   // validation: no retry
        return handler;
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
            ConsumerFactory<String, Order> consumerFactory, DefaultErrorHandler errorHandler) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, Order>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);                                  // <= partitions per instance
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

Poison pills: ErrorHandlingDeserializer

Remember that deserialization happens inside poll(), before Spring sees anything, so "Spring has no way to handle the problem". The fix is to wrap your real deserializer in ErrorHandlingDeserializer. If the delegate fails, it returns null and puts a DeserializationException (with the raw bytes) in a header. The container sees the header, skips your listener and calls the error handler. Because DeserializationException is on the fatal list, the record goes straight to the recoverer. Paired with DeadLetterPublishingRecoverer, the DLT record gets the original bytes back as its value. Configure the delegate with ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS (spring.deserializer.value.delegate.class). In 4.x, the Jackson 3 JacksonJsonDeserializer replaces the deprecated Jackson 2 JsonDeserializer.

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JacksonJsonDeserializer.class);
props.put(JacksonJsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.Order");
props.put(JacksonJsonDeserializer.TRUSTED_PACKAGES, "com.example");

Non-blocking retries: @RetryableTopic and @DltHandler

The alternative to stalling a partition is to move the failing record aside. With @RetryableTopic on a @KafkaListener, a failure forwards the record to a retry topic with a back-off timestamp. The retry topic's consumer pauses that partition until the record is due, then processes it again. After the last attempt, the record goes to the DLT, handled by a @DltHandler method (or a default consumer "which only logs"). The framework creates the topics and containers for you.

@RetryableTopic(attempts = "4",
        backOff = @BackOff(delay = 1000, multiplier = 2.0, maxDelay = 10000))
@KafkaListener(topics = "payments", groupId = "payments-svc")
void onPayment(Payment p) { gateway.charge(p); }             // may throw

@DltHandler
void onDeadPayment(Payment p, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    alerts.raise(p, topic);
}
// creates payments-retry-1000, payments-retry-2000, payments-retry-4000, payments-dlt

Defaults to know: attempts = "3" (the number of attempts made before the message goes to the DLT), a fixed back-off of 1000 ms, retry suffix -retry, DLT suffix -dlt, numPartitions = "1" and replicationFactor = "-1" (broker default) for auto-created topics, and SameIntervalTopicReuseStrategy.SINGLE_TOPIC. So a fixed back-off uses one …-retry topic. (4.1 changed RetryTopicConfigurationBuilder's default to SINGLE_TOPIC too, to match the annotation.) In 4.0 the @Backoff from Spring Retry became Spring Kafka's own @BackOff, because Spring Retry was dropped in favour of Spring Framework 7's core retry support.

Two hard limits from the docs: non-blocking retries are not supported with batch listeners, and they cannot combine with container transactions. And the big caveat, in the docs' own capitals: "IMPORTANT: By using this strategy you lose Kafka's ordering guarantees for that topic." Record 18 overtakes the retried record 17. That is exactly what the figure's toggle shows.

That combination is real: extend RetryTopicConfigurationSupport and override configureBlockingRetries(…) to retry chosen exceptions in place first (the docs' example: a database hiccup that "would likely trigger errors on the next records as well"), with non-blocking retries after that. There's a global list of fatal exceptions for retry topics too; those skip retries and go straight to the DLT.

Transactions

Give the DefaultKafkaProducerFactory a transactionIdPrefix and it keeps a cache of transactional producers (transactional.id = prefix + n). The prefix "must be unique per instance". With Spring Boot, setting spring.kafka.producer.transaction-id-prefix auto-configures a KafkaTransactionManager and wires it into the listener container. The container then starts a transaction before calling your listener, your KafkaTemplate sends join it, and on success it sends the consumed offsets with sendOffsetsToTransaction() before committing. That's the consume-transform-produce pattern from Part III, done for you. On an exception it rolls back and repositions the consumer. Repeated failures go to the DefaultAfterRollbackProcessor, the transactional twin of DefaultErrorHandler. Spring Kafka 3.0+ only supports EOSMode.V2 (KIP-447 fencing). The docs are careful to scope exactly-once: the read → process → write sequence completes once; "the read and process have at least once semantics". Side effects outside Kafka (an HTTP call, an email) can still happen twice.

For a database write plus a Kafka send, annotate the method @Transactional with your DataSourceTransactionManager. The KafkaTemplate synchronizes with it, and the DB commits first, then Kafka. That is two transactions, not one atomic one. If the second commit fails, the docs say the exception reaches the caller and you "should take remedial action … to compensate". For outside-world side effects, pair transactions with idempotent consumers.

Sending: KafkaTemplate returns a CompletableFuture

Every send(…) returns CompletableFuture<SendResult<K, V>>. Attach whenComplete((result, ex) -> …) for non-blocking handling. The Throwable can be cast to KafkaProducerException, whose getFailedProducerRecord() gives you the record. If you must block, the docs recommend get() "with a timeout". By default the template has a LoggingProducerListener that only logs errors, so a send you never check fails silently apart from a log line. The producer-side rules from the last chapter still apply underneath: keep completion handlers short, and move slow work to your own executor (e.g. whenCompleteAsync(…, executor)).

Share consumers (Kafka queues) in Spring

Spring Kafka 4.0 added early-access share consumers. The docs say that since 4.1, "backed by Apache Kafka 4.2, share consumers are promoted to full production support". You use a DefaultShareConsumerFactory, a ShareKafkaListenerContainerFactory and a plain @KafkaListener(containerFactory = "shareKafkaListenerContainerFactory"). Acknowledgement follows ContainerProperties.ShareAckMode (new in 4.1, replacing a boolean):

• EXPLICIT (default): the container sends ACCEPT when your listener returns. On an exception, a ShareConsumerRecordRecoverer decides ACCEPT, RELEASE or REJECT, and the default is REJECTING.
• MANUAL: your listener takes a ShareAcknowledgment and must call exactly one of acknowledge(), release(), reject(). Unacknowledged records block the next poll. renew() (KIP-1222) extends the lock for slow work.
• IMPLICIT: maps to share.acknowledgement.mode=implicit; every acquired record is accepted "regardless of processing outcome".

Since 4.1, undeserializable records are caught at poll level and REJECTed, so a poison pill can't kill the thread. Current limits: no batch listeners and no message converters for share consumers. The broker's delivery-count limit (the share.delivery.count.limit group config, broker default 5 via group.share.delivery.count.limit) is your built-in "poison pill" cutoff, and there is no ordering promise to lose.

Observability and testing

If Micrometer is on the classpath with a single MeterRegistry, the listener container and KafkaTemplate record timers automatically: spring.kafka.listener with tags name, result (success/failure) and exception, and spring.kafka.template for sends. Set observationEnabled to true on ContainerProperties and KafkaTemplate to switch to Micrometer Observation (see the Micrometer Tracing reference for tracing). MicrometerProducerListener/MicrometerConsumerListener bind the clients' own metrics (KafkaClientMetrics). One 4.1.1 behaviour change: KafkaAdmin now remembers a failed cluster-id lookup for clusterIdRetryInterval (default 5 minutes). Previously, an observation-enabled container could block on every record while the cluster id was unavailable.

For tests, @EmbeddedKafka starts an in-JVM broker. Since 4.0 that's only EmbeddedKafkaKraftBroker, KRaft only; the ZooKeeper variant and the JUnit 4 EmbeddedKafkaRule are gone. The docs recommend a single broker across tests with a different topic per test, via Spring's context caching. For tests closer to production (real broker image, real networking), the Testcontainers Kafka module's org.testcontainers.kafka.KafkaContainer runs the apache/kafka or apache/kafka-native images, and getBootstrapServers() gives you the address to point your factories at.

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic payments-dlt --from-beginning \
  --formatter-property print.headers=true --formatter-property print.offset=true

After twenty-six chapters it would be easy to believe every messaging problem is a Kafka problem. It isn't. Our city ledger archive is superb at one thing: keeping an ordered, replayable, replicated record that many teams can read independently at their own pace. Some jobs look more like a post office: hand each parcel to exactly one courier, route it by address, and forget it once delivered. Others want the archive's model with a different building underneath. This chapter compares Kafka 4.3 with the main alternatives, fairly and with dates.

The one question that sorts everything: log or queue?

A log keeps records after they are read. Consumers track their own position, many independent readers can read the same data, and you can rewind (replay). Order is kept per partition. A queue hands each message to one consumer, tracks per-message acknowledgement, and deletes the message once it's acknowledged. Kafka consumer groups are log semantics. RabbitMQ classic and quorum queues are queue semantics. The interesting systems now offer both: Kafka added share groups (queue-like consumption over a log), RabbitMQ added streams (a log inside a broker built around queues), and Pulsar exposes both through subscription types.

RabbitMQ: queues, quorum queues, streams

RabbitMQ's model starts with routing. Publishers send to exchanges, which route to queues by binding rules. A queue "is an ordered collection of messages", FIFO, but priorities, multiple consumers and requeues after negative acknowledgements can change the order. After a consumer acknowledges, the message is removed. Quorum queues are "a modern queue type which implements a durable, replicated queue based on the Raft consensus algorithm" and the docs call them "the default choice when needing a replicated, highly available queue". They have per-message poison handling built in: since RabbitMQ 4.0 the delivery limit defaults to 20, after which the message is dropped or dead-lettered.

RabbitMQ streams are RabbitMQ's answer to the log: "an append-only log of messages that can be repeatedly read until they expire", with "non-destructive consumer semantics". They are always replicated and persistent, retention is set with x-max-length-bytes / x-max-age, consumers attach at an offset, timestamp or first/last, single active consumer gives ordered exclusive consumption, and super streams partition one logical stream across several streams. The docs' use cases are large fan-outs, replay/time-travel, throughput and large backlogs.

Apache Pulsar: brokers without disks

Pulsar separates serving from storage. Brokers are stateless ("The Pulsar message broker is a stateless component"), message data lives in Apache BookKeeper bookies as append-only ledgers, and a metadata store coordinates the cluster (Oxia recommended, ZooKeeper, or RocksDB). The Kafka analogy is a branch clerk who keeps no books at all: every page is written to a separate vault service. That makes brokers easy to scale out and replace, at the cost of running more distinct components. On top it has built-in multi-tenancy (tenants/namespaces), geo-replication, and four subscription types: Exclusive, Failover, Shared (round-robin, queue-like) and Key_Shared (messages with the same key go to one consumer). It also has negative acknowledgement, a retry letter topic and a dead letter topic built in. By default acknowledged messages are deleted; retention policies keep them for replay.

Redpanda: the Kafka API on a different engine

Redpanda is not a different messaging model. It's a different implementation of Kafka's model. Its docs say it is "compatible with the Kafka API", is "Built on C++", is "packaged as a single binary: it doesn't rely on any external systems", uses a thread-per-core model via the Seastar library, runs "Raft consensus … for data replication" where "every topic partition forms a Raft group", and offers tiered storage. Your Kafka clients and most tools connect unchanged. What you're comparing is operations, cost, performance and licensing, not semantics. Check which newer Kafka protocol features you depend on before assuming parity. Redpanda's Kafka-compatibility page says it "does not implement the server-side portion of KIP-890", so Kafka 4.x clients fall back to the original transaction protocol (no Transactions V2). KIP-848 consumer groups and share groups (KIP-932) are not mentioned in its compatibility docs.

Managed cloud streams: Kinesis and Pub/Sub

Amazon Kinesis Data Streams is a log, like Kafka. A stream is a set of shards (≈ partitions), records are routed by a partition key hashed with MD5, and each shard supports writes of up to 1,000 records/s or 1 MB/s and reads of up to 2 MB/s. Retention defaults to 24 hours and can be raised to 8,760 hours (365 days) at extra cost. The Kinesis Client Library keeps consumer state in DynamoDB tables. Google Cloud Pub/Sub is closer to a queue: at-least-once delivery by default, an optional exactly-once delivery feature, and ordering disabled by default. You enable it per subscription at creation and publish related messages with the same ordering key, limited to 1 MBps of publishing per key. Both remove the operations work. You give up protocol portability (Kafka clients don't speak them) and the Kafka ecosystem (Connect, Streams). Managed Kafka services are the middle ground: the Kafka API and semantics without running brokers. Compare their Kafka version and enabled features against what you need.

…and the option inside Kafka: share groups

Before adding a second broker technology for "we need a work queue", remember Part III. Since Kafka 4.2, share groups are production-ready. Consumers cooperatively consume the same partitions, there can be more consumers than partitions, records are acknowledged individually (accept / release / reject / renew), and delivery attempts are counted, with a broker default limit of 5 (group.share.delivery.count.limit) and a 30 s acquisition lock by default. You keep one platform, one security model and replay of the underlying topic. You give up ordering: the Kafka docs say to use share groups "in cases where records are processed one at a time, rather than as part of an ordered stream". What they don't give you (as of 4.3) is RabbitMQ-style routing or a broker-managed dead-letter topic. A rejected record simply isn't delivered again, so capture it yourself if you need it.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic thumbnails --partitions 3
bin/kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic thumbnails --group thumbs
# start the same command in 3 more terminals, then produce 20 records

One or more questions per chapter. Questions tagged (L2)–(L4) belong to the deeper levels; the wrong answers are the misconceptions the chapters took apart. Click an answer to see why.

Cheat sheet