08 · Replication, ISR & High Watermark

Source: Apache Kafka 4.4.0-SNAPSHOT (git 04bfe7d, 2026-06-15), KRaft mode. Derived from source code, not copied from official documentation.

Replication is the durability engine of Kafka: every topic-partition is a small replicated state machine whose authoritative metadata (assignment, leader, leader epoch, in-sync replica set) lives in the KRaft metadata log, while the actual log data is pushed from a single leader broker to its followers via the Fetch protocol. This chapter dissects how a broker materialises that metadata into live Partition objects through ReplicaManager.applyDelta, how the leader tracks each follower's log end offset and decides who is in-sync, how the high watermark (the committed-data boundary) advances as the minimum LEO over the ISR, how ISR changes are committed by round-tripping an AlterPartition RPC to the controller, how leader epochs and epoch-based truncation prevent silent log divergence, and how acks=all producers block in a purgatory until the high watermark crosses their offset.

Role & responsibilities

The replication subsystem answers one question for every byte a producer writes: is this record safe? "Safe" means it has been durably replicated to enough replicas that it survives the loss of any tolerated number of brokers and will never be silently dropped by a future leader. Concretely the subsystem owns:

• Per-broker partition ownership. ReplicaManager holds every hosted partition, applies metadata deltas to drive each into a leader or follower role, routes produce/fetch to the right UnifiedLog, and periodically checkpoints high watermarks to disk.
• Per-partition state & durability accounting. Partition is the state machine: it tracks the assignment, the ISR, the leader epoch, the local log, and one Replica object per remote follower carrying that follower's LEO and last-fetch timestamps.
• ISR maintenance. Shrinking the ISR when a follower falls behind (replica.lag.time.max.ms), expanding it when a follower catches up, and committing both through the controller via AlterPartition.
• High-watermark advancement, computing the committed offset as the min LEO over the (maximal) ISR, and unblocking acks=all produce and follower fetch requests when it moves.
• Divergence prevention, leader epochs, the epoch→start-offset cache, and epoch-based truncation so a returning follower trims exactly the records that were never committed.

Where it lives in the code

Core concepts & terminology

AR (assigned replicas)

The ordered broker-id list assigned to the partition. The first element is the preferred leader. Held in assignmentState.replicas (Partition.scala:196).

ISR (in-sync replicas)

The subset of AR the controller currently believes is caught up. Committed copy is partitionState.isr. Membership is the precondition for being elected leader without data loss (under clean election).

Maximal ISR

The ISR the leader uses locally for HW advancement and acks=all quorum, which during a pending expansion optimistically includes the not-yet-committed replica. Defined per pending state via PartitionState.maximalIsr().

LEO (log end offset)

The offset of the next record to be appended. The leader's LEO is its own; each follower's LEO is recorded from its Fetch requests in the corresponding Replica.

HW (high watermark)

The committed boundary: the largest offset known to be replicated to all of the (maximal) ISR. Consumers (READ_UNCOMMITTED) may only read below it. HW = min LEO over the ISR, gated by min ISR.

LSO (last stable offset)

The boundary below which there are no open transactions; READ_COMMITTED consumers read below it. Tracked by UnifiedLog, surfaced via fetchOffsetSnapshot.

Leader epoch

A monotonically increasing integer the controller assigns on every leadership change. Stamped into each record batch and cached as epoch→start-offset, it is the anchor for truncation (KIP-101).

Partition epoch

A second monotonic counter bumped on every partition metadata change (including ISR-only changes). It is the optimistic-concurrency token for AlterPartition.

ELR (eligible leader replicas)

Replicas the controller knows are complete up to the HW but which dropped out of the ISR (e.g. a clean shutdown). They can be elected cleanly even though they are not currently in the ISR (KIP-966).

Data structures

PartitionRegistration, the authoritative metadata record

PartitionRegistration (PartitionRegistration.java:153-163) is the in-memory image of one partition's controller-owned metadata, reconstructed by replaying PartitionRecord/PartitionChangeRecord from the metadata log. Its fields:

elr and lastKnownElr are persisted as tagged fields only when ELR is enabled and non-empty (PartitionRegistration.java:387-392), keeping records small on clusters that have not adopted KIP-966.

Replica & ReplicaState, the leader's view of a follower

On the leader, each remote follower is a Replica holding a single AtomicReference<ReplicaState> (Replica.java:36,42). All updates are lock-free compare-and-set via replicaState.updateAndGet(...); readers grab an immutable snapshot through stateSnapshot(). ReplicaState is a record (ReplicaState.java:32-39):

The "caught up" rule is computed in updateFetchStateOrThrow (Replica.java:84-91): if the follower's fetch offset reached the leader's current LEO, lastCaughtUpTime jumps to now; else if it reached the leader's LEO as of the previous fetch (lastFetchLeaderLogEndOffset), it advances to the previous fetch time; otherwise it is unchanged. This two-tier rule lets a follower remain "in sync" even though its fetch offset chronically trails a busy leader by a few records, what matters is that it keeps closing the gap that existed when it last asked.

LeaderEpochFileCache, epoch → start-offset

Each log keeps a TreeMap<Integer,EpochEntry> mapping leader epoch → start offset of that epoch, guarded by its own ReentrantReadWriteLock and persisted to a leader-epoch-checkpoint file (LeaderEpochFileCache.java:59-60). assign(epoch, startOffset) appends a new entry and flushes synchronously; truncation flushes asynchronously to avoid fsync stalls on the fetcher/log threads (LeaderEpochFileCache.java:335-355,367-388). The pivotal query is endOffsetFor(requestedEpoch, logEndOffset) (LeaderEpochFileCache.java:284-328): it returns the largest cached epoch <= the request and the exclusive end offset of that epoch (= start offset of the next epoch, or the LEO for the latest epoch). That is precisely the answer to the OffsetsForLeaderEpoch RPC and the truncation pivot for a returning follower.

LogOffsetMetadata, an offset plus its physical position

HW, LEO and LSO are not bare longs; they are LogOffsetMetadata (LogOffsetMetadata.java:33-35): a messageOffset, the segmentBaseOffset of the segment containing it, and the relativePositionInSegment. Carrying the segment position lets the leader answer fetches and compute byte deltas without re-resolving the offset, and lets maybeIncrementHighWatermark advance the HW even when the offset is unchanged but the log has rolled to a new segment (UnifiedLog.java:573-574).

The ISR state hierarchy

partitionState is a @volatile PartitionState implementing a tiny state machine. CommittedPartitionState (steady state) reports maximalIsr() == isr() and isInflight() == false. While an AlterPartition is outstanding the field is swapped (under the write lock) to a pending record:

The asymmetry (PendingExpandIsr.java:43-47 vs PendingShrinkIsr.java:41-43) is a deliberate safety choice: optimistically counting the new replica during an expansion only makes the HW condition stricter, so if the controller rejects the expansion no committed offset was over-advanced; conversely a shrink is not applied to the maximal ISR until the controller confirms it, so a rejected shrink cannot have prematurely advanced the HW past a replica that is still required.

Architecture & control/data flow

Detailed mechanics

Becoming leader or follower: applyDelta

Metadata changes arrive as a TopicsDelta. ReplicaManager.applyDelta (ReplicaManager.scala:2360) takes replicaStateChangeLock, processes deletions first (so a recreated topic with the same name is not clobbered), then splits the local changes into leaders and followers and applies each:

• Leaders (applyLocalLeadersDelta, ReplicaManager.scala:2418): remove any follower fetcher for the partition, then call partition.makeLeader(registration, isNew, checkpoints...).
• Followers (applyLocalFollowersDelta, ReplicaManager.scala:2449): call partition.makeFollower(...); if the leader epoch changed, (re)start a ReplicaFetcherThread targeting the new leader at initialFetchOffset(log).

makeLeader (Partition.scala:588) runs under the per-partition leaderIsrUpdateLock write lock. It rejects stale registrations (partitionRegistration.partitionEpoch < partitionEpoch), updates the assignment and ISR from the registration via updateAssignmentAndIsr, ensures the local log exists, and, only if the leader epoch advanced, records the new epoch's start offset in the epoch cache (leaderLog.assignEpochStartOffset, Partition.scala:649), resets every follower's ReplicaState via resetReplicaState, and sets leaderEpoch/leaderEpochStartOffsetOpt. It finishes by calling maybeIncrementLeaderHW because an ISR that just dropped to one replica may allow the HW to jump to the local LEO. makeFollower (Partition.scala:694) is simpler: it sets leaderReplicaIdOpt/leaderEpoch, clears the local ISR (a follower does not track ISR), and returns whether the leader epoch changed so the caller knows to restart the fetcher.

Recording follower progress and advancing the high watermark

When a follower's Fetch is served, the leader calls updateFollowerFetchState (Partition.scala:767). Under the ISR read lock it pushes the follower's new offsets into its Replica (updateFetchStateOrThrow), then attempts two things outside that read lock: maybeExpandIsr (might this follower now rejoin the ISR?) and, if the follower's LEO actually advanced, maybeIncrementLeaderHW.

maybeIncrementLeaderHW (Partition.scala:1010) is the heart of commit. It refuses to advance while the partition is under min ISR (isUnderMinIsr), then walks the leader's LEO downward against every remote replica:

newHW = leader.LEO
for each remote replica r:
    if r.LEO < newHW and
       (maximalIsr.contains(r) or (r.isCaughtUp(leader.LEO, now, lagMs) and isReplicaIsrEligible(r))):
        newHW = r.LEO
leaderLog.maybeIncrementHighWatermark(newHW)   // monotonic; only moves up

So the HW is the minimum LEO over the maximal ISR, but it additionally waits for any replica that is "caught up and eligible" even though it has not yet been committed to the ISR. The comment at Partition.scala:992-1004 spells out the reason: a replica that is considered caught-up but whose LEO momentarily trails the HW should hold the HW back, otherwise it could be elected leader and serve a stale view; and counting newly-added replicas (the maximal ISR) makes the commit condition strictly safer (KIP-497). maybeIncrementHighWatermark (UnifiedLog.java:563-581) enforces monotonicity and that the new HW never exceeds the LEO.

Shrinking the ISR (lag detection)

A scheduler task isr-expiration runs every replicaLagTimeMaxMs / 2 ms (ReplicaManager.scala:285) and calls maybeShrinkIsr() on every online partition (ReplicaManager.scala:2100-2107). Partition.maybeShrinkIsr (Partition.scala:1089) first checks under the read lock (cheap, no write contention) whether any follower is out of sync via getOutOfSyncReplicas(replicaLagTimeMaxMs) (Partition.scala:1155), which filters ISR members (excluding the leader) by !isCaughtUp(...). If so it re-checks under the write lock, builds a PendingShrinkIsr, swaps partitionState, and submits an AlterPartition. The lag covers two failure shapes (Partition.scala:1142-1151): a stuck follower whose LEO has not moved for maxLagMs, and a slow follower that never closes the gap to the leader's LEO within maxLagMs, both collapse to the single lastCaughtUpTimeMs test.

Expanding the ISR

Expansion is driven by fetches, not a timer. After recording a follower's progress, maybeExpandIsr (Partition.scala:876) checks canAddReplicaToIsr (not already in ISR, not in-flight, and isReplicaIsrEligible) and isFollowerInSync (Partition.scala:907-912): the follower's LEO must be >= the leader's HW and >= the leader-epoch start offset. The second clause is essential, a replica must have caught up within the current leader epoch before it joins, otherwise it could be elected and lose data that sits between the HW and LEO (Partition.scala:862-872). Eligibility (isReplicaIsrEligible, Partition.scala:914-932) additionally requires the broker not be fenced, not be in controlled shutdown, and that its cached broker epoch matches the broker epoch carried in its Fetch (or the Fetch sent -1 to bypass).

Committing the change: the AlterPartition round-trip

Both shrink and expand call submitAlterPartition(pendingState) (Partition.scala:1687), which builds a LeaderAndIsr carrying the proposed ISR with each member's broker epoch (addBrokerEpochToIsr, Partition.scala:1661) and hands it to AlterPartitionManager.submit. DefaultAlterPartitionManager (DefaultAlterPartitionManager.java:53) batches updates: a ConcurrentHashMap holds at most one pending item per partition, and an AtomicBoolean inflightRequest ensures only one AlterPartition RPC is on the wire at a time (DefaultAlterPartitionManager.java:60-64,118-125). The request is sent to the active controller through the NodeToControllerChannelManager and is never timed out, it retries indefinitely until answered or until a newer metadata update supersedes it (DefaultAlterPartitionManager.java:138-140).

On the controller, ReplicationControlManager validates the proposal (ReplicationControlManager.java:1300-1356): the partition epoch must match (else INVALID_UPDATE_VERSION); the new ISR must be a subset of the assignment and must contain the leader (else INVALID_REQUEST); and every proposed member must be eligible, alive, unfenced, with a matching broker epoch, else INELIGIBLE_REPLICA. If valid, it appends a PartitionChangeRecord (bumping partitionEpoch) and returns the committed LeaderAndIsr.

Back on the leader, the completion handler (Partition.scala:1693) runs under the write lock. If partitionState no longer equals the proposed state (a leader election or metadata replay moved it on), the response is ignored as stale. On success, handleAlterPartitionUpdate (Partition.scala:1794) installs a fresh CommittedPartitionState, updates the local partitionEpoch, and re-runs maybeIncrementLeaderHW (a shrink may now permit the HW to advance). On error, handleAlterPartitionError (Partition.scala:1733) maps the code to a retry-or-give-up decision; notably INELIGIBLE_REPLICA and OPERATION_NOT_ATTEMPTED are authoritative "not applied" signals that let the leader safely reset to the last committed state, while transient codes trigger a resubmit.

The acks=all produce path

A produce with acks=-1 must not be acknowledged until its records are committed (HW >= the produced offset). ReplicaManager.appendRecords (ReplicaManager.scala:638) appends to the leader's log via Partition.appendRecordsToLeader (Partition.scala:1219), which under the read lock first guards durability, if the ISR is smaller than the effective min ISR and acks=-1, it throws NotEnoughReplicasException before writing (Partition.scala:1234-1238), then appends and re-checks the HW. If the append did not already satisfy the request, maybeAddDelayedProduce (ReplicaManager.scala:878) parks a DelayedProduce in the produce purgatory keyed by each partition. The operation completes when checkEnoughReplicasReachOffset(requiredOffset) (Partition.scala:947) sees leaderLog.highWatermark >= requiredOffset; if at that moment the maximal ISR has fallen below min ISR it completes with NOT_ENOUGH_REPLICAS_AFTER_APPEND (the data is durable but the durability contract weakened). The trigger that re-checks the purgatory is HW movement: whenever an append or follower fetch increases the HW, an action enqueued by addCompletePurgatoryAction (ReplicaManager.scala:853-876) calls delayedProducePurgatory.checkAndComplete.

acks=0

Producer does not wait; the broker does not even send a response body. No durability guarantee.

acks=1

Leader responds as soon as its local append succeeds; delayedProduceRequestRequired is false (ReplicaManager.scala:1361-1367). Survives leader loss only if a follower had already replicated the record.

acks=-1 (all)

Leader waits for HW >= offset, i.e. replication to the whole (maximal) ISR, and refuses to write at all if ISR < min.insync.replicas.

Leader epochs & epoch-based truncation

Before KIP-101, a follower truncated to the leader's HW on becoming a follower, which could diverge if leadership changed faster than the HW propagated. Now every batch carries the leader epoch in which it was written, and the LeaderEpochFileCache records where each epoch began. When a follower (re)starts fetching it asks the leader, via OffsetsForLeaderEpoch, for the end offset of the follower's latest epoch; endOffsetFor returns the start offset of the next epoch (the exact divergence point), and the follower truncates there. In the fetch path the leader detects divergence inline: readRecords (Partition.scala:1367) compares the follower's lastFetchedEpoch against the cache and, if they diverge, returns an empty fetch carrying a divergingEpoch (Partition.scala:1401-1413) instead of records, prompting the follower to truncate. The server-side answer is Partition.lastOffsetForLeaderEpoch (Partition.scala:1585).

Unclean leader election & Eligible Leader Replicas

If every ISR member is unavailable, the partition has no safe leader. With unclean.leader.election.enable=false (the default, LogConfig.java:139) the partition stays offline until an ISR member returns. If unclean election is permitted, the controller may elect an out-of-sync replica, which can lose committed records; that replica's leaderRecoveryState becomes RECOVERING, and the controller refuses any AlterPartition that grows the ISR beyond one replica or flips RECOVERED→RECOVERING while recovering (ReplicationControlManager.java:1326-1344). PartitionRegistration.electionWasUnclean keys off this state (PartitionRegistration.java:165-167).

KIP-966 narrows the gap between "lose data" and "stay offline": the controller tracks an ELR, replicas it knows still hold all data up to the HW even after they leave the ISR. When the ISR empties, PartitionChangeBuilder can elect from the ELR cleanly (PartitionChangeBuilder.java:319-336): a replica is an acceptable leader if it is in the ISR, or, when the ISR is empty, in the ELR; on election from the ELR it becomes the sole ISR member and is removed from the ELR. As a last resort, lastKnownElr retains the last good leader so a single-member recovery can proceed even when both ISR and ELR are empty (PartitionChangeBuilder.java:247,270,299-312). ELR/lastKnownElr are persisted only when the feature is enabled.

Concurrency & threading

Each Partition is thread-safe via a ReentrantReadWriteLock leaderIsrUpdateLock (Partition.scala:183), with a documented lock-ordering contract (Partition.scala:148-162): the partition lock is acquired before the log lock. The read lock guards the common hot path (appends, follower fetches, HW reads); the write lock guards leadership transitions and ISR mutations. The double-checked pattern in maybeExpandIsr/maybeShrinkIsr (test under read lock, re-test and mutate under write lock) avoids serialising the fetch path on the rare ISR change. Per-follower state needs no partition lock at all: Replica uses an AtomicReference updated by CAS.

Configuration reference

Note that min.insync.replicas and unclean.leader.election.enable are topic-level overridable; the fetcher tunings are broker-level. The HW checkpoint filename is replication-offset-checkpoint (ReplicaManager.scala:89).

Failure modes, edge cases & recovery

• Leader crash. The controller fences the broker, elects a new leader from the ISR (or ELR), bumps the leader epoch, and replays the change to all brokers. Followers restart fetchers and truncate to the new leader's epoch boundary; no committed record (below the old HW) is lost because every ISR member held all committed data.
• Follower falls behind. Dropped from the ISR after replica.lag.time.max.ms; the HW can then advance using the smaller ISR. The follower re-enters once its LEO reaches the HW and the current epoch's start offset.
• ISR shrinks below min ISR. acks=all produces are refused with NotEnoughReplicas before writing; in-flight delayed produces may complete with NOT_ENOUGH_REPLICAS_AFTER_APPEND. The HW stops advancing (maybeIncrementLeaderHW returns early when isUnderMinIsr).
• AlterPartition rejected. INVALID_UPDATE_VERSION means another change won the race, the leader awaits fresh metadata. INELIGIBLE_REPLICA/OPERATION_NOT_ATTEMPTED are safe "not applied" signals and reset to the last committed state. Transient errors retry after 50 ms (DefaultAlterPartitionManager.java:174).
• Returning zombie leader. A deposed leader that still thinks it leads will have its AlterPartition rejected (stale partition epoch) and its produce/fetch rejected by epoch checks; its followers' fetches carry a newer leader epoch and it learns it is no longer leader.
• Log directory failure (JBOD). handleLogDirFailure (ReplicaManager.scala:2157) marks affected partitions offline, removes their fetchers, drops the dir from the HW checkpoint map, and notifies the controller via the directory event handler so leadership moves elsewhere.
• Append offset below log start (delete-records race). A follower whose first fetched batch straddles the leader's advanced log start offset triggers truncateFullyAndStartAt to restart its log from the batch's base offset (Partition.scala:1197-1216).
• Stale HW after unclean election. If the HW lags the epoch start offset, ListOffsets for "latest" returns OFFSET_NOT_AVAILABLE until the HW recovers (Partition.scala:1455-1460).

Invariants & guarantees

Interactions with other subsystems

• The Log Storage Engine, appends, HW/LEO/LSO bookkeeping, and the leader-epoch checkpoint all live in UnifiedLog; Partition is a thin replication wrapper over it.
• The Fetch Path & Replica Fetchers, the follower side that pulls records, performs epoch-based truncation, and reports LEO back; the leader's fetchRecords emits diverging-epoch responses.
• The KRaft Controller, owns the authoritative ISR/leader/ELR, validates and commits AlterPartition, runs leader election and reassignment.
• Metadata Propagation & Broker Lifecycle, delivers the TopicsDelta that applyDelta turns into leader/follower transitions; broker registration/fencing feeds ISR eligibility.
• Request Processing (KafkaApis), dispatches Produce/Fetch into ReplicaManager and wires the delayed-operation purgatories.
• Transactions & Exactly-Once Semantics, the LSO and the min-ISR durability gate underpin transactional commit visibility.
• The Producer Client, the acks setting selects which of the durability paths above the producer relies on.

Design rationale & evolution

Gotchas / operational notes