Kafka Replication: Fetching, the ISR, Leader Epochs and Truncation

New to Kafka? Start with How Kafka Moves a Message: topics, partitions, producers and consumers. This page zooms in on the copies of one partition.

The idea: one leader decides what is committed

Every Kafka partition is stored on several brokers, the replicas. One of them is the leader: producers write to it, and (by default) consumers read from it. The others are followers. A follower does not wait to be sent data: it asks the leader for it with Fetch requests, just as a consumer does. On the page, partition orders-0 has three replicas on brokers B1, B2 and B3, and B1 starts as the leader.

The leader decides when a record is committed: when every replica in the in-sync replica set (ISR) has it. The offset up to which that holds is the high watermark (HW). Consumers only see records below the HW, and an acks=all producer gets its answer when the HW passes its batch.

Fetching: the fetch offset is the acknowledgement

Each record has an offset, its position in the log. A replica's log end offset (LEO) is the offset of the next record it will write, which is the number of records it has. A follower sends Fetch(fetch offset = its LEO): "give me everything from here". There is no separate acknowledgement. When B2 asks for offset 4, the leader knows that B2 has offsets 0 to 3.

The leader keeps the last fetch offset of every follower and computes HW = min LEO of the ISR (its own LEO included). The HW only goes up. A follower that is caught up does not spin: its fetch waits at the leader (a long poll, replica.fetch.wait.max.ms) until there is something new. On the page these waiting fetches are not drawn (Demo 1: replication, LEO and HW).

The follower's HW is one fetch behind

The fetch response carries the records and the leader's current HW, and the follower sets its own HW to min(leader's HW, its LEO). But the leader only raises the HW after it has seen the follower's next fetch, so a new record takes two rounds: the first fetch brings the record, the second tells the leader that the follower has it (the HW moves and the producer gets its answer), and that second response brings the new HW to the follower. That one-round lag is harmless while everything runs. It mattered for truncation, as below shows.

The ISR and replica.lag.time.max.ms

A follower is in sync while it keeps up. The leader records when each follower last fetched up to the leader's LEO ("caught up"). If that was more than replica.lag.time.max.ms ago (30 s by default; 3 ticks on the page), the follower is too slow, maybe because of a long GC pause or a slow disk. The leader then sends AlterPartition to the controller with a smaller ISR. Once the controller has accepted it, the HW no longer waits for the slow replica, and waiting acks=all requests are answered. When the follower catches up again (its LEO reaches the HW), the leader sends another AlterPartition to add it back (Demo 2: slow follower leaves the ISR).

The ISR is stored by the controller (in KRaft, in the metadata log) because it decides who may become leader. Only a replica in the ISR is known to have every committed record. A dead broker is removed from the ISR by the controller as soon as its heartbeats stop (broker.session.timeout.ms). The last member is never removed, so the controller remembers who has the data.

acks, min.insync.replicas and NOT_ENOUGH_REPLICAS

A shrinking ISR makes commits easier: with an ISR of one, the leader alone "commits" a record. min.insync.replicas puts a floor under that for acks=all producers (Demo 3: acks=all below min.insync):

Producer	ISR = 3 or 2	ISR = 1 (min.insync.replicas = 2)
acks=all	answered when the HW passes the batch	refused with NOT_ENOUGH_REPLICAS, nothing appended
acks=1	answered after the leader's own append	accepted; committed with one copy
acks=0	no answer at all	accepted, no answer

If the ISR shrinks after the append, the answer is NOT_ENOUGH_REPLICAS_AFTER_APPEND. The record is in the log, but the producer is told it failed. With replication factor 3, min.insync.replicas=2 and acks=all, every acknowledged record is on at least two brokers, and the partition still takes writes with one broker down.

Leader epochs and the leader-epoch-checkpoint

When the leader dies, the controller picks the first live replica of the ISR and bumps the partition's leader epoch, a counter of leader changes. The new leader stamps every record it writes with the epoch, and every replica keeps a small file, leader-epoch-checkpoint, with one line per epoch: (epoch, start offset), the offset of the first record written in that epoch. On the page cells are coloured by epoch, and the checkpoint is listed in each replica's panel.

A leader can die with records that no follower has: with acks=1 it already answered, and with acks=all it was still waiting. Those records sit above the HW, so they were never committed, and the new leader writes different records at the same offsets. When the old leader comes back as a follower, it must find out where its log stops agreeing with the new leader's. It sends OffsetsForLeaderEpoch(its last epoch). The leader looks that epoch up in its own checkpoint and answers where it ended, which is the start offset of the next epoch. The returning replica truncates everything from there and then fetches normally (Demo 4: failover, old leader truncates). Followers that were alive during the election do the same check against the new leader.

KIP-279 fixed a corner case: if the leader never had the follower's last epoch (two quick failovers in a row), it answers with its largest older epoch and where that ended, the follower truncates to the smaller of the two ends, and the two sides repeat the exchange with older epochs until they agree.

Why truncating to the HW lost data

Before Kafka 0.11 there were no leader epochs. A follower that restarted (or got a new leader) simply cut its log back to its own HW, on the theory that everything above the HW was uncommitted. But a follower's HW lags one fetch behind the leader's. KIP-101 describes how that lost acknowledged data (Demo 5: HW truncation loses a record):

1. B3 is down, so the ISR is [B1, B2]. The leader B1 writes m1; B2 fetches it (B2's LEO is 2, but its HW is still 1).
2. B2's next fetch tells B1 that B2 has m1. The HW moves to 2 and the producer gets its acks=all answer: m1 is committed.
3. B2 restarts before the response with HW 2 arrives. It truncates to its own HW, 1, and m1 is gone from B2.
4. B1 dies. B2 is in the ISR, so it becomes leader, without m1. An acknowledged record is lost.

With leader epochs (Demo 6: leader epochs keep it) the restarted B2 asks B1 OffsetsForLeaderEpoch(0), learns that epoch 0 ends at B1's LEO 2, and keeps m1. The HW is never used for truncation any more. The same rule also fixes the second KIP-101 problem, where two replicas could end up with different records at the same offset and never notice.

Unclean leader election

If every replica in the ISR is down, the partition is offline: no writes, no reads. Out-of-sync replicas may still be alive, but they are missing committed records. unclean.leader.election.enable decides what happens next (Demo 7: unclean election loses records):

	false (default)	true
Who may lead	only an ISR member	any live replica, if no ISR member is alive
While the ISR is down	partition offline, waits	an out-of-sync replica leads at once
Committed records	kept	those the new leader lacks are lost; the old leader truncates them when it returns
Choose it for	durability (payments, orders)	availability (metrics, logs you can afford to lose)

The setting is a topic config that can be changed on a live cluster. Turning it on while a partition is offline lets the controller elect at once, which is what the demo does. Leader epochs still keep the replicas consistent after an unclean election: the returning ISR member truncates its committed records to match the new leader. Consistency is kept, but those records are gone.

How the page simplifies

• One partition, three replicas, one producer. Times are scaled: one tick is one fetch round, and replica.lag.time.max.ms is 3 ticks.
• The controller notices a dead broker at once. In a real cluster that takes broker.session.timeout.ms (9 s by default), and a follower learns about a new leader from the metadata log, not from a message.
• Fetches that are caught up wait at the leader and are not drawn. Every other fetch, response, AlterPartition and OffsetsForLeaderEpoch is.
• A "slow" follower stops fetching entirely, and a quick restart (Demos 5 and 6) keeps the broker in the ISR.

See also How Kafka Moves a Message for the whole path from producer to consumer and Kafka Consumer Groups for rebalancing and exactly-once. The other Kafka internals pages cover log segments, log compaction, the KRaft controller quorum and zero-copy fetches. Raft uses the same ideas in a consensus protocol: terms play the part of leader epochs, and a follower deletes conflicting entries after an election. Database replication compares leader-follower setups, and quorum reads and writes shows the leaderless alternative.

What the page leaves out

Producer retries and idempotence (a retried batch after a failover is deduplicated by sequence number). Eligible leader replicas (KIP-966, Kafka 4.x), which keep track of replicas that are safe to elect even when they have dropped out of the ISR. Fetching from the closest follower (KIP-392, replica.selector.class). Preferred leader election and partition reassignment. Follower throttling, tiered storage, and the log start offset that retention moves. The ReplicaFetcherThread batching many partitions into one fetch, and the replica-offset-checkpoint file that stores each replica's HW on disk.

References

Kafka documentation: replication (ISR, unclean leader election, min.insync.replicas)

KIP-101: Alter Replication Protocol to use Leader Epoch rather than High Watermark for Truncation

KIP-279: Fix log divergence between leader and follower after fast leader fail over

Confluent: Hands-free Kafka replication (replica.lag.time.max.ms)