Digging into Kafka's Source 9: Replication — The ISR, the High Watermark and the Wait Behind acks=all

What this post covers

This is the ninth post in the "Digging into Kafka's Source" series. In part 6 we saw how the leader broker writes the records of a Produce request to its own log, in part 7 how that log is laid out on disk, and in part 8 how Fetch requests read it back. At the end of part 8 we said that "consumers and followers use the same Fetch"; this post is the follower's side of that story. It follows a record written on the leader as it moves to other brokers, and how that fact travels back to the leader to decide both the producer's acks=all response and how far consumers can read.

A close-up of area D (partitions and replication) on the big map from part 2. The blue boxes are the main characters of this post, the yellow boxes hold state, and the grey boxes belong to the posts before and after it.

There are four questions to answer.

• How does a follower get records from the leader — which thread, and with which request?
• When does the high watermark move? Why can't a consumer read a record the leader has just written?
• What decides when a follower leaves the ISR and when it comes back, and who makes that decision final?
• What exactly does acks=all wait for, and where does min.insync.replicas reject a request?

All line numbers in this post refer to this commit. Part 1 shows how to check out the same commit and build the Docker image used for the examples.

Partition.scala, the heart of replication, is 1,856 lines; AbstractFetcherThread.scala is 873 and ReplicaManager.scala is 2,577. Moving logs between directories (JBOD), tiered storage and reassignment all live in the same files. This post picks only the main path a newcomer can follow; what it leaves out is listed by name and location in "What this post leaves out" at the end.

Words to know first

Replica, leader, follower: one partition is placed on several brokers as copies (replicas), and one of them, the leader, takes the writes and reads. The other replicas, the followers, fetch the leader's log and copy it (part 1). The controller decides who the leader is and spreads that through the metadata log (part 2, part 10).

LEO (log end offset): the offset the next record written to that replica's log will get. A log holding offsets 0–9 has an LEO of 10. Every replica has its own.

High watermark (HW): the boundary offset meaning "everything before this is held by the whole ISR". Consumers can only read records before the HW: with an HW of 10 they see up to offset 9. The leader computes it, and followers copy down the value the leader tells them.

ISR (in-sync replicas): the set of replicas that are "keeping up well enough" with the leader. The leader itself is in it. It's the Isr: 1,2,3 in kafka-topics.sh --describe. Section 4 of this post reads the source for what "well enough" means exactly.

Leader epoch: a generation number that goes up by one every time the partition's leader changes (in some cases it goes up even when the leader stays, such as when a reassignment removes replicas — PartitionChangeBuilder.triggerLeaderEpochBumpForReplicaReassignmentIfNeeded()). The leader stamps it on every record batch it writes (the partitionLeaderEpoch field of the batch header from part 6). After a leader change it tells you "which record was written under which leader", which is how diverged logs are reconciled (section 6). Separately there is a partition epoch, which goes up whenever the partition's state — its ISR or leader — changes.

Purgatory and DelayedProduce: a waiting room where a response is held back until a condition is met. Where part 8's DelayedFetch waited "until enough data piles up", this post's DelayedProduce waits "until replication is done".

AlterPartition: the request a leader broker sends to the controller saying "please change this partition's ISR to this". The leader decides the ISR, but the controller confirms and records it.

1. The follower side — a ReplicaFetcherThread sends Fetch requests to the leader

When a broker receives metadata saying "you are a follower of this partition", ReplicaManager.applyLocalFollowersDelta() calls Partition.makeFollower() and attaches the partition to a fetcher (ReplicaManager.scala lines 2458–2563). The fetcher threads are created in ReplicaFetcherManager, a 59-line class.

  override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaFetcherThread = {
    val threadName = s"ReplicaFetcherThread-$fetcherId-${sourceBroker.id}"
    val logContext = new LogContext(s"[ReplicaFetcher replicaId=${brokerConfig.brokerId}, leaderId=${sourceBroker.id}, " +
      s"fetcherId=$fetcherId] ")
    val endpoint = new BrokerBlockingSender(sourceBroker, brokerConfig, metrics, time, fetcherId,
      s"broker-${brokerConfig.brokerId}-fetcher-$fetcherId", logContext)
    val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id)
    val leader: LeaderEndPoint = new RemoteLeaderEndPoint(logContext.logPrefix, endpoint, fetchSessionHandler, brokerConfig,
      replicaManager, quotaManager, metadataVersionSupplier, brokerEpochSupplier)
    new ReplicaFetcherThread(threadName, leader, brokerConfig, failedPartitions, replicaManager,
      quotaManager, logContext.logPrefix)
  }

core/src/main/scala/kafka/server/ReplicaFetcherManager.scala lines 41–52

Reading the Scala — override def f(...): T = { ... } redefines a parent's method, and the last expression in the block (new ReplicaFetcherThread(...)) is the return value. s"...$x...${a.b}" drops values into a string (string interpolation). val is a variable you can't reassign.

The thread name is ReplicaFetcherThread-<fetcherId>-<leader broker id>. Notice there is no partition in the name. There is one fetcher thread per "leader broker × fetcherId", not one per partition. AbstractFetcherManager.addFetcherForPartitions() groups partitions by (leader broker, getFetcherId(partition)) to choose the thread, and getFetcherId() is abs(31 * hash of the topic name + partition number) % num.replica.fetchers (AbstractFetcherManager.scala lines 111–115 and 120–150). num.replica.fetchers defaults to 1 (ReplicationConfigs.java lines 95–96), so with default settings one fetcher thread per leader broker handles every partition fetched from that leader. Whether you follow 100 partitions led by broker 1 or just one, there is a single thread, ReplicaFetcherThread-0-1.

BrokerBlockingSender is a thin wrapper that creates one NetworkClient and one Selector from the clients module (at most one request in flight), sends a request and waits for the response (BrokerBlockingSender.scala lines 71–123). The same Selector the broker used on the receiving side in part 4 is used here on the sending side. A follower broker behaves like a client toward the leader.

Drawn from broker 2's point of view. The blue boxes are the thread and the objects it calls; the yellow box is the per-partition state the thread holds.

One loop: truncate if needed, then send a Fetch

ReplicaFetcherThread inherits from ShutdownableThread and keeps calling doWork() until told to stop. The parent class's doWork() is two lines.

  override def doWork(): Unit = {
    maybeTruncate()
    maybeFetch()
  }

core/src/main/scala/kafka/server/AbstractFetcherThread.scala lines 110–113

Reading the Scala — : Unit is Java's void. A method with no arguments is called with parentheses, like maybeTruncate().

maybeTruncate() only handles partitions that need the "truncation" we'll see in section 6; usually it does nothing. maybeFetch() builds a request with leader.buildFetch(partitionStates.partitionStateMap) and sends it via processFetchRequest() (lines 115–134). If there is no partition to fetch, it waits on a condition variable for replica.fetch.backoff.ms. The subclass ReplicaFetcherThread adds one line: at the end of each loop it wakes the DelayedFetches of partitions whose high watermark went up (completeDelayedFetchRequests(), ReplicaFetcherThread.scala lines 107–110). That is for consumers that read from a follower ("fetch from the closest replica", left out of this post).

What goes into the request is in RemoteLeaderEndPoint.buildFetch().

          val logStartOffset = replicaManager.localLogOrException(topicPartition).logStartOffset
          val lastFetchedEpoch = fetchState.lastFetchedEpoch()
          builder.add(topicPartition, new FetchRequest.PartitionData(
            fetchState.topicId().orElse(Uuid.ZERO_UUID),
            fetchState.fetchOffset(),
            logStartOffset,
            fetchSize,
            Optional.of(fetchState.currentLeaderEpoch()),
            lastFetchedEpoch))

core/src/main/scala/kafka/server/RemoteLeaderEndPoint.scala lines 184–192

Five things are sent per partition: fetchOffset ("give me records from here"), its own log start offset, the maximum bytes, the leader epoch it currently knows (currentLeaderEpoch) and the leader epoch of the last batch it received (lastFetchedEpoch). The request as a whole is built with FetchRequest.Builder.forReplica(version, brokerId, brokerEpoch, maxWait, minBytes, ...) (lines 212–213). That forReplica is the only difference from a consumer's Fetch: the request carries the follower's broker id and broker epoch, so the leader receiving it knows "this is a follower's Fetch". maxWait and minBytes are replica.fetch.wait.max.ms (default 500) and replica.fetch.min.bytes (default 1) (ReplicationConfigs.java lines 74–80). The long polling from part 8, where the leader holds a Fetch for up to half a second when there is nothing new, applies to followers too.

Append the received records as they are, and note the HW the leader sent

When the response arrives, processFetchRequest() checks each partition's error code (NONE, OFFSET_OUT_OF_RANGE, FENCED_LEADER_EPOCH …, lines 279–388) and, if all is well, calls processPartitionData().

    // Append the leader's messages to the log
    val logAppendInfo = partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = false, partitionLeaderEpoch)

    if (logTrace)
      trace("Follower has replica log end offset %d after appending %d bytes of messages for partition %s"
        .format(log.logEndOffset, records.sizeInBytes, topicPartition))
    val leaderLogStartOffset = partitionData.logStartOffset

    // For the follower replica, we do not need to keep its segment base offset and physical position.
    // These values will be computed upon becoming leader or handling a preferred read replica fetch.
    var maybeUpdateHighWatermarkMessage = s"but did not update replica high watermark"
    log.maybeUpdateHighWatermark(partitionData.highWatermark).ifPresent { newHighWatermark =>
      maybeUpdateHighWatermarkMessage = s"and updated replica high watermark to $newHighWatermark"
      partitionsWithNewHighWatermark += topicPartition
    }

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala lines 132–146

Reading the Scala — var is a variable you can reassign. opt.ifPresent { x => ... } passes a Scala lambda (x => ...) to Java's Optional.ifPresent(x -> ...). buf += x adds one element to a mutable collection. Arguments can also be passed by name, like isFuture = false.

Two things happen. The first is the append. appendRecordsToFollowerOrFutureReplica() ends up calling UnifiedLog.appendAsFollower(), which calls append(..., AppendOrigin.REPLICATION, false, ...) — the third argument, assignOffsets, is false (UnifiedLog.java lines 1093–1103). The batches the leader stamped with offsets and a leader epoch in part 6 are appended by the follower byte for byte, without renumbering. This is also when the follower records the leader epoch stamped on each batch in its own epoch cache (assignEpochStartOffset(batch.partitionLeaderEpoch(), batch.baseOffset()), lines 1226–1229). That is why every replica's log ends up identical, down to the offsets.

The second is the HW. The response carries the leader's HW, and the follower adopts it with maybeUpdateHighWatermark(). But the updateHighWatermark() inside it clips the value to no more than its own LEO.

    public long updateHighWatermark(LogOffsetMetadata highWatermarkMetadata) throws IOException {
        LogOffsetMetadata endOffsetMetadata = localLog.logEndOffsetMetadata();
        LogOffsetMetadata newHighWatermarkMetadata = highWatermarkMetadata.messageOffset < logStartOffset
            ? new LogOffsetMetadata(logStartOffset)
            : highWatermarkMetadata.messageOffset >= endOffsetMetadata.messageOffset
                ? endOffsetMetadata
                : highWatermarkMetadata;

        updateHighWatermarkMetadata(newHighWatermarkMetadata);
        return newHighWatermarkMetadata.messageOffset;
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 542–552

A follower's HW is min(the HW the leader sent, my LEO). The leader's HW is "where the whole ISR has the data", so it is usually at or below the follower's LEO anyway, but this guard keeps the follower from ever using an offset it doesn't have as its HW. The comment above the method also says "upper-bounded by the log end offset" (lines 583–591).

Example 1-1: the fetcher thread names

The examples in this post were run with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411), in a Docker container (Eclipse Temurin JDK 25). One controller (node 100) and three brokers (nodes 1, 2, 3) run in one container (ex/u09-isr-hw/run.sh), and we create a topic rep with its replicas pinned to 1:2:3. --replica-assignment 1:2:3 fixes even the order of the replica list, so the first one, broker 1, becomes the leader (the "first live replica" rule from part 2).

kafka-topics.sh --bootstrap-server $BOOTSTRAP --create --topic rep --replica-assignment 1:2:3
jstack $(kafka_pid 2) | grep -o '^"ReplicaFetcherThread[^"]*"'

These two lines are picked and shortened from run.sh (the real script prints each broker's jstack result joined onto one line). Here is the thread part of the output.

broker 2: "ReplicaFetcherThread-0-1" 
broker 3: "ReplicaFetcherThread-0-1" 
broker 1: (none: leader of rep-0)
broker 1: "broker-1-to-controller-alter-partition-channel-manager" 

Brokers 2 and 3, the followers, each have one ReplicaFetcherThread-0-1 fetching from the leader (broker 1); broker 1, the leader, has none. Instead broker 1 has a broker-1-to-controller-alter-partition-channel-manager thread that will send ISR changes to the controller (section 4).

2. The leader side — the Fetch offset itself says "I have everything up to here"

A follower never sends a separate "I got the records" acknowledgement. The fetchOffset of its next Fetch is that report. Asking for fetchOffset=11 means "I have up to 10 in my log", that is, "my LEO is 11". On the leader, ReplicaManager.readFromLog(), which handles Fetch requests, calls Partition.fetchRecords() for each partition, and that is where follower Fetches take their own branch.

    if (fetchParams.isFromFollower) {
      // Check that the request is from a valid replica before doing the read
      val (replica, logReadInfo) = inReadLock(leaderIsrUpdateLock, () => {
        val localLog = localLogWithEpochOrThrow(
          fetchPartitionData.currentLeaderEpoch,
          fetchParams.fetchOnlyLeader
        )
        val replica = followerReplicaOrThrow(
          fetchParams.replicaId,
          fetchPartitionData
        )
        val logReadInfo = readFromLocalLog(localLog)
        (replica, logReadInfo)
      })

      if (updateFetchState && !logReadInfo.divergingEpoch.isPresent) {
        updateFollowerFetchState(
          replica,
          followerFetchOffsetMetadata = logReadInfo.fetchedData.fetchOffsetMetadata,
          followerStartOffset = fetchPartitionData.logStartOffset,
          followerFetchTimeMs = fetchTimeMs,
          leaderEndOffset = logReadInfo.logEndOffset,
          fetchParams.replicaEpoch
        )
      }

core/src/main/scala/kafka/cluster/Partition.scala lines 1301–1325

Reading the Scala — val (a, b) = ... unpacks a two-value tuple into two variables at once. The last expression of the lambda () => { ... }, (replica, logReadInfo), is that tuple.

isFromFollower checks whether the request's replicaId is a valid broker id (0 or more) (FetchParams.java lines 67–69). The read itself is the same readFromLocalLog() a consumer gets, but how far it reads differs. FetchIsolation.of() picks LOG_END (to the end of the log) for followers and HIGH_WATERMARK (up to the HW) for consumers (FetchIsolation.java lines 31–39, the branch we saw in part 1). Followers must take records that aren't replicated yet, or replication would never progress; consumers must only see what is already replicated.

And only for a follower's Fetch does it go on to updateFollowerFetchState(). The updateFetchState in the condition is passed by ReplicaManager as !readFromPurgatory (ReplicaManager.scala lines 1838–1844), so the state is updated once, when the request first arrives and is read. When a long-polled request is read again later, the state is not touched.

What the leader keeps for each follower: Replica and lastCaughtUpTimeMs

The leader's Partition holds one Replica object per follower (remoteReplicasMap, line 181), and inside it swaps an immutable record, ReplicaState, through an AtomicReference. Its fields are the follower's log start and end offsets, the time of its last Fetch, the leader's LEO at that time, lastCaughtUpTimeMs, and the broker epoch carried in the Fetch (ReplicaState.java lines 32–39). The heart of Replica.updateFetchStateOrThrow(), called from updateFollowerFetchState(), is the three-way choice of lastCaughtUpTimeMs.

            long lastCaughtUpTime;
            if (followerFetchOffsetMetadata.messageOffset >= leaderEndOffset) {
                lastCaughtUpTime = Math.max(currentReplicaState.lastCaughtUpTimeMs(), followerFetchTimeMs);
            } else if (followerFetchOffsetMetadata.messageOffset >= currentReplicaState.lastFetchLeaderLogEndOffset()) {
                lastCaughtUpTime = Math.max(currentReplicaState.lastCaughtUpTimeMs(), currentReplicaState.lastFetchTimeMs());
            } else {
                lastCaughtUpTime = currentReplicaState.lastCaughtUpTimeMs();
            }

server/src/main/java/org/apache/kafka/server/replica/Replica.java lines 84–91

lastCaughtUpTimeMs is "the last time this follower had fully caught up with the leader". If this Fetch's offset is at or past the leader's current LEO, now is that time (first branch). Even if not, if this Fetch's offset is at or past the leader's LEO at the previous Fetch, the leader accepts that "at the time of the previous Fetch, it had caught up to the end as it was then" and uses the previous Fetch time (second branch). The comment above the method explains why the second branch is needed: when small produce requests arrive non-stop, the follower's Fetch offset is always a little behind the leader's LEO, and with the first branch alone it would never count as "caught up" (comment on lines 53–68).

This value is used for the ISR decisions in section 4. After recording it, updateFollowerFetchState() goes on to check with maybeExpandIsr(replica) whether to grow the ISR, checks with maybeIncrementLeaderHW() whether the HW can move if the follower's LEO changed, and if anything went up, wakes waiting requests with tryCompleteDelayedRequests() (Partition.scala lines 798–815). This one method is where all three outcomes of replication — the ISR, the HW and the acks=all response — get moved.

Drawn as if there were only one follower. With two followers, step 4 has to come from both before the HW moves (section 3).

What to notice in the picture is steps 2 and 4: two Fetches are needed. The Fetch in step 2 carries the record to the follower, but the leader only learns that "the follower has written it" when the Fetch in step 4 arrives. That round trip is part of acks=all's response time. The follower's HW (step 6) lags the leader's by a beat. When there is nothing new, a follower's Fetch is held at the leader for replica.fetch.wait.max.ms before it returns (a follower's DelayedFetch works off LOG_END, so an HW increase alone doesn't complete it, DelayedFetch.java lines 129–133), so a follower's HW can trail by up to about half a second.

Easy to get wrong: the follower LEO the leader keeps is an "inference"

The follower LEO the leader records is not where the follower has actually written on disk, but the offset of the last Fetch the follower sent. If a follower stops right after writing some records, the leader doesn't know the follower has them. Conversely, within one leader epoch, the follower's real LEO is never smaller than what the leader thinks, because the Fetch offset is an end that has already been written (the case where a follower truncates its log after a leader change comes in section 6). That is why the HW calculation can be wrong on the "too low" side but never on the "too high" side.

3. The high watermark — the smallest LEO in the ISR

The only place that raises the leader's HW is Partition.maybeIncrementLeaderHW(). The comment above the method lists the two occasions it is called: when the ISR changes and when some replica's LEO changes (lines 987–1012).

  private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, currentTimeMs: Long = time.milliseconds): Boolean = {
    if (isUnderMinIsr) {
      trace(s"Not increasing HWM because partition is under min ISR(ISR=${partitionState.isr})")
      return false
    }
    // maybeIncrementLeaderHW is in the hot path, the following code is written to
    // avoid unnecessary collection generation
    val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
    var newHighWatermark = leaderLogEndOffset
    remoteReplicasMap.forEach { (_, replica) =>
      val replicaState = replica.stateSnapshot

      def shouldWaitForReplicaToJoinIsr: Boolean = {
        replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
        isReplicaIsrEligible(replica.brokerId)
      }

      // Note here we are using the "maximal", see explanation above
      if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
          (partitionState.maximalIsr.contains(replica.brokerId) || shouldWaitForReplicaToJoinIsr)
      ) {
        newHighWatermark = replicaState.logEndOffsetMetadata
      }
    }

    leaderLog.maybeIncrementHighWatermark(newHighWatermark).toScala match {

core/src/main/scala/kafka/cluster/Partition.scala lines 1013–1038

Reading the Scala — currentTimeMs: Long = time.milliseconds is an argument with a default value. return false ends the method on the spot, as in Java (it's rare in Scala). forEach { (_, replica) => ... } is Java's Map.forEach((k, v) -> ...), and _ is an "unused argument". You can define a small method with def inside another method.

It reads simply. Start from the leader's own LEO, walk the followers, and if a follower has a smaller LEO, pull the value down to it. Two kinds of followers get to pull it down: those in the ISR (maximalIsr, section 4 below), and those not yet in the ISR but caught up and about to join (shouldWaitForReplicaToJoinIsr). The comment (lines 994–999) explains the second kind: if the leader is alone in the ISR and doesn't wait for a follower that is catching up, the HW (= the leader's LEO) keeps running ahead and the follower never makes it into the ISR.

The isUnderMinIsr check at the very top is worth noticing too. If the ISR is smaller than min.insync.replicas (or the replica count, if that is smaller; effectiveMinIsr(), lines 246–248), the HW doesn't move at all. Even records accepted with acks=1 become invisible to consumers. This ties into the min.insync.replicas discussion in section 5.

The UnifiedLog.maybeIncrementHighWatermark() on the last line only changes the value if the new one is larger (UnifiedLog.java lines 563–581); the one exception is the same offset on a newly rolled segment (because that offset's file position changed). So the HW the leader updates on this path never goes down. (Followers are different: when the truncation of section 6 drops the LEO below the HW, the HW is lowered to the LEO, UnifiedLog.java lines 2391–2392.) When a follower drops out of the ISR and the "smallest LEO" gets bigger, the HW jumps up all at once.

Green is records held by all three replicas (before the HW); yellow is records held by only some. If broker 3 drops out of the ISR, the HW jumps to 12.

The HW is written to disk too: replication-offset-checkpoint

The HW lives in memory, but it is written to a file periodically so a restarted broker can use it. ReplicaManager keeps a replication-offset-checkpoint file in each log directory (ReplicaManager.scala line 89 and lines 217–218), and every replica.high.watermark.checkpoint.interval.ms (default 5000) checkpointHighWatermarks() writes the HW of every partition (lines 263–266 and 2125–2147). The file has the same format as the other checkpoint files in part 7 (a version number, a line count, then topic partition value lines). At the end of example 1 we look at this file on all three brokers.

4. Leaving and rejoining the ISR — the leader decides, the controller confirms

The rule for leaving: not caught up even once in replica.lag.time.max.ms

The check that shrinks the ISR runs on a timer, not when a Fetch arrives.

  def startup(): Unit = {
    // start ISR expiration thread
    // A follower can lag behind leader for up to config.replicaLagTimeMaxMs x 1.5 before it is removed from ISR
    scheduler.schedule("isr-expiration", () => maybeShrinkIsr(), 0L, config.replicaLagTimeMaxMs / 2)

core/src/main/scala/kafka/server/ReplicaManager.scala lines 282–285

Reading the Scala — () => maybeShrinkIsr() is a lambda with no arguments (Java's () -> maybeShrinkIsr()): the work the scheduler should run each period, passed as a value. config.replicaLagTimeMaxMs / 2 is integer division.

Every half of replica.lag.time.max.ms (default 30000, ReplicationConfigs.java lines 54–55), Partition.maybeShrinkIsr() is called for every partition on this broker. As the comment says, the threshold is 30 seconds and the check runs every 15, so a lagging follower is dropped somewhere between 30 and 45 seconds. Who is lagging is decided like this.

  def getOutOfSyncReplicas(maxLagMs: Long): Set[Int] = {
    val current = partitionState
    if (!current.isInflight) {
      val candidateReplicaIds = (current.isr.asScala.map(_.toInt) - localBrokerId).toSet
      val currentTimeMs = time.milliseconds()
      val leaderEndOffset = localLogOrException.logEndOffset
      candidateReplicaIds.filter(replicaId => isFollowerOutOfSync(replicaId, leaderEndOffset, currentTimeMs, maxLagMs))
    } else {
      Set.empty
    }
  }

core/src/main/scala/kafka/cluster/Partition.scala lines 1158–1168

Reading the Scala — if (...) { a } else { b } is an expression that produces a value, so at the end of a method it is the return value. set - x is a new set without x, asScala views a Java collection as a Scala one, and map(_.toInt) is short for map(x -> x.toInt()).

Of the ISR minus the leader itself, it picks the followers for which isFollowerOutOfSync() holds, and that check is !ReplicaState.isCaughtUp(...) (lines 1136–1143).

    public boolean isCaughtUp(
        long leaderEndOffset,
        long currentTimeMs,
        long replicaMaxLagMs) {
        return leaderEndOffset == logEndOffset() || currentTimeMs - lastCaughtUpTimeMs <= replicaMaxLagMs;
    }

server/src/main/java/org/apache/kafka/server/replica/ReplicaState.java lines 62–67

"Caught up" means one of two things: its LEO equals the leader's, or replica.lag.time.max.ms has not yet passed since it last fully caught up (lastCaughtUpTimeMs). How many records it is behind (a count) is never looked at — only time. Very old versions of Kafka are said to have had a count-based setting too (replica.lag.max.messages; I haven't checked this against source). That name doesn't appear anywhere in this commit's repository.

The first condition has an interesting consequence. On a partition that receives no new records, a follower doesn't drop out of the ISR even if it is frozen. The leader's LEO and the frozen follower's LEO (as far as the leader knows) are equal, so it is "caught up". Step 4 of example 1 confirms this.

The rule for joining: caught up to the HW and to the start of the current leader epoch

The other direction is checked by maybeExpandIsr() every time a Fetch arrives. There are two conditions.

  private def isFollowerInSync(followerReplica: Replica): Boolean = {
    leaderLogIfLocal.exists { leaderLog =>
      val followerEndOffset = followerReplica.stateSnapshot.logEndOffset
      followerEndOffset >= leaderLog.highWatermark && leaderEpochStartOffsetOpt.exists(followerEndOffset >= _)
    }
  }

core/src/main/scala/kafka/cluster/Partition.scala lines 909–914

Reading the Scala — An Option is a box that may hold a value (Some) or not (None). opt.exists(f) means "there is a value and f is true for it". The _ in followerEndOffset >= _ is the value in the box.

The follower's LEO must be at or past the leader's HW and at or past the offset where the current leader epoch began. The reason for the second condition is in the comment above the method (lines 864–877): there may be committed records between a new leader's HW and LEO, and a follower that joins the ISR before receiving them could become leader and lose them. On top of that come the checks in isReplicaIsrEligible() — the broker must not be fenced (marked "not alive" by the controller), must not be in controlled shutdown, and the broker epoch in its Fetch must match the one in the metadata (lines 916–934). That last one keeps a stale Fetch from a restarted broker, arriving late, from wrongly putting it into the ISR.

The leader only proposes: PendingShrinkIsr and AlterPartition

Even when the leader decides "let's shrink the ISR to 1,2", it doesn't change the ISR on the spot. The ISR is a value that must be recorded in the controller's metadata log (part 10), and if the leader changes, the new leader must know the same ISR. So the partitionState of a Partition takes one of three shapes.

All three states are Java records in server/.../partition/ (CommittedPartitionState, PendingShrinkIsr, PendingExpandIsr).

The pending states mean two things. First, while one change is in progress (isInflight), no other change starts (the if (!current.isInflight) in getOutOfSyncReplicas()). Second, the maximalIsr used for the HW calculation differs by state. The comments in prepareIsrExpand() and prepareIsrShrink() give the reason.

    // When expanding the ISR, we assume that the new replica will make it into the ISR
    // before we receive confirmation that it has. This ensures that the HW will already
    // reflect the updated ISR even if there is a delay before we receive the confirmation.
    // Alternatively, if the update fails, no harm is done since the expanded ISR puts
    // a stricter requirement for advancement of the HW.

core/src/main/scala/kafka/cluster/Partition.scala lines 1616–1620

    // When shrinking the ISR, we cannot assume that the update will succeed as this could
    // erroneously advance the HW if the `AlterPartition` were to fail. Hence the "maximal ISR"
    // for `PendingShrinkIsr` is the current ISR.

core/src/main/scala/kafka/cluster/Partition.scala lines 1643–1645

The principle is one: the HW must never run ahead because of a change that isn't confirmed. While growing the ISR, the new replica is counted in advance, making the HW more conservative; while shrinking, the departing replica is still counted, so the HW doesn't move until the controller accepts. That is why an acks=all request held up by a departing follower is only released when the AlterPartition response comes back (example 1 times this).

The proposal goes to alterIsrManager.submit() (submitAlterPartition(), lines 1690–1724). On the receiving side, DefaultAlterPartitionManager looks like this.

    @Override
    public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
                                                  LeaderAndIsr leaderAndIsr) {
        CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
        AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
        boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
        if (enqueued) {
            maybePropagateIsrChanges();
        } else {
            future.completeExceptionally(new OperationNotAttemptedException(String.format(
                    "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
        }
        return future;
    }

    void maybePropagateIsrChanges() {
        // Send all pending items if there is not already a request in-flight.
        if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
            // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
            List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
            sendRequest(inflightAlterPartitionItems);
        }
    }

server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java lines 103–125

Each partition can have only one unsent change (putIfAbsent), and only one AlterPartition request at a time for the whole broker flies to the controller (inflightRequest). Changes for other partitions that pile up meanwhile go out together in the next request. This keeps a broker with thousands of partitions from flooding the controller with requests. The request is sent by the broker-1-to-controller-alter-partition-channel-manager thread we saw in example 1-1 (DefaultAlterPartitionManager.java lines 80–88; the thread name format is in NodeToControllerChannelManagerImpl.java line 79).

On the controller, ReplicationControlManager.alterPartition() checks the broker epoch and partition epoch and checks again, for every broker in the new ISR, that it is registered, not in controlled shutdown, not fenced and has a matching broker epoch (ineligibleReplicasForIsr(), ReplicationControlManager.java lines 1418–1438). If that passes, it writes a PartitionChangeRecord to the metadata log, bumps the partition epoch and responds (from line 1093). The leader takes the response in handleAlterPartitionUpdate(), returns to CommittedPartitionState and calls maybeIncrementLeaderHW() once more with the new ISR (Partition.scala lines 1797–1828). The ISR updated to ... and version updated to N line in the server log is printed here.

Easy to get wrong: there are two ways the ISR shrinks

Besides the leader shrinking it with maybeShrinkIsr(), there is a way the controller shrinks it directly. When a broker starts a controlled shutdown (handleBrokerInControlledShutdown()), or is fenced after its heartbeats stop for broker.session.timeout.ms (default 9000, KRaftConfigs.java lines 43–44) (handleBrokerFenced()), the controller produces records that remove that broker from every ISR it is in (ReplicationControlManager.java lines 1449–1460 and 1514–1523). Stopping a broker with kafka-server-stop.sh (SIGTERM) takes this path, so the ISR shrinks almost immediately; only a broker that is alive but slow takes the leader's path. Example 1 raises broker.session.timeout.ms to 30 seconds to see only the leader's path. With the default, the frozen broker would be fenced after 9 seconds and fall into the controller's path.

Example 1-2: a frozen follower, the ISR, the HW and acks=all

The same run.sh starts the cluster with replica.lag.time.max.ms=5000 and broker.session.timeout.ms=30000. The check runs every half of that, 2.5 seconds. The steps are:

1. Write 10 records with acks=all.
2. Freeze broker 3 with kill -STOP (the process is frozen, but its sockets stay open), wait 12 seconds without writing anything, and look at the ISR. Then wake it with kill -CONT.
3. A Java program (IsrWait.java) creates two producers and sends one record with each, then freezes broker 3 again, immediately sends 5 records with acks=1, asks for the HW, then sends 1 record with acks=all and times how long the response takes.
4. Wake broker 3 and see whether the ISR comes back.

Step 3 is done inside Java because starting a JVM takes a few seconds. At first I froze the broker from the shell and then started Java, but when the container was busy the program took 17 seconds to send its first record, and the threshold (5 seconds) had passed by then. The HW is asked with the Admin client's listOffsets(OffsetSpec.latest()). ListOffsets(LATEST) without an isolation level returns the end a consumer can read, i.e. the HW.

        try (KafkaProducer<String, String> acks1 = producer("1");
             KafkaProducer<String, String> acksAll = producer("all")) {
            // Warm-up: each producer sends one record (the follower is alive, so both finish quickly)
            acks1.send(new ProducerRecord<>(topic, "warmup-acks1")).get();
            acksAll.send(new ProducerRecord<>(topic, "warmup-acksall")).get();
            System.out.printf("before SIGSTOP: high watermark = %d%n", highWatermark(tp));

            new ProcessBuilder("kill", "-STOP", followerPid).start().waitFor();
            stoppedAt = System.currentTimeMillis();
            System.out.println("SIGSTOP sent to the follower");

            for (int i = 0; i < 5; i++) {
                RecordMetadata md = acks1.send(new ProducerRecord<>(topic, "acks1-" + i)).get();
                System.out.printf("acks=1   offset=%d  (t+%d ms)%n", md.offset(), since());
            }
            System.out.printf("high watermark = %d  (t+%d ms)%n", highWatermark(tp), since());

            long start = System.currentTimeMillis();
            RecordMetadata md = acksAll.send(new ProducerRecord<>(topic, "acksall-0")).get();
            System.out.printf("acks=all offset=%d  waited %d ms  (t+%d ms)%n",
                    md.offset(), System.currentTimeMillis() - start, since());
            System.out.printf("high watermark = %d  (t+%d ms)%n", highWatermark(tp), since());

[shortened from ex/u09-isr-hw/IsrWait.java]

The output follows. Steps 1 and 2 (creating the topic, and example 1-1's threads) and the final stopped line are left out. The rep-0 Leader: … lines are printed by the example's Isr.java, which asks through the Admin client. The times (t+… ms, measured from the freeze), waited … ms, the lastCaughtUpTimeMs value, and the order in which the server log prints the ISR (2,1,3 or 3,2,1) differ on every run. The timestamps at the start of the server log lines were removed.

=== 3. write 10 records with acks=all
rep:0:10
=== 4. SIGSTOP broker 3, no new records, wait 12s
rep-0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
=== 5. Java: warm up, SIGSTOP broker 3 again, then produce
before SIGSTOP: high watermark = 12
SIGSTOP sent to the follower
acks=1   offset=12  (t+23 ms)
acks=1   offset=13  (t+36 ms)
acks=1   offset=14  (t+51 ms)
acks=1   offset=15  (t+66 ms)
acks=1   offset=16  (t+75 ms)
high watermark = 12  (t+160 ms)
acks=all offset=17  waited 4964 ms  (t+5126 ms)
high watermark = 18  (t+5191 ms)
rep-0 Leader: 1 Replicas: 1,2,3 Isr: 1,2
--- leader (broker 1) server.log
INFO [Partition rep-0 broker=1] Shrinking ISR from 2,1,3 to 1,2. Leader: (highWatermark: 12, endOffset: 18). Out of sync replicas: (brokerId: 3, endOffset: 12, lastCaughtUpTimeMs: 1794027649149). (kafka.cluster.Partition)
INFO [Partition rep-0 broker=1] ISR updated to 1,2  and version updated to 1 (kafka.cluster.Partition)
=== 6. SIGCONT broker 3, wait 5s
rep-0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
--- leader (broker 1) server.log
INFO [Partition rep-0 broker=1] ISR updated to 2,1,3  and version updated to 2 (kafka.cluster.Partition)
=== 7. replication-offset-checkpoint (high watermark on disk)
broker 1: rep 0 18
broker 2: rep 0 18
broker 3: rep 0 18

Let's match it with the source line by line.

• Step 4: broker 3 was frozen for 12 seconds (more than twice the 5-second threshold), yet the ISR is still 1,2,3. With no new records, the leader's LEO (10) and broker 3's LEO (10) are equal, so the first condition of isCaughtUp() is true.
• Step 5, the warm-up: the warm-up records sent by the two producers take offsets 10 and 11, so the HW before the freeze is 12.
• Step 5, acks=1: the responses come back as soon as the leader writes, regardless of the frozen follower (offsets 12–16). But the HW asked right afterwards is still 12, so the five records just written are invisible to consumers. Broker 3 is still in the ISR and its LEO is stuck at 12 (the "smallest LEO" from section 3).
• Step 5, acks=all: the response took 4,964ms — about 5.1 seconds counted from the freeze (reruns gave 5,093ms and 6,008ms; since the check runs every 2.5 seconds, it varies roughly between 5 and 7.5 seconds). Only after 5 seconds passed since lastCaughtUpTimeMs, the check running every 2.5 seconds noticed, printed Shrinking ISR from 2,1,3 to 1,2, and AlterPartition was accepted and ISR updated to 1,2 was printed did the HW move and the DelayedProduce get released. The HW asked right after is 18: even the acks=1 records that had been blocked became visible at once (the "jumping HW" from section 3).
• The Shrinking ISR line: Leader: (highWatermark: …, endOffset: …) and Out of sync replicas: (brokerId: 3, endOffset: …, lastCaughtUpTimeMs: …) are printed as-is by the info(...) in maybeShrinkIsr() (lines 1116–1119). The leader thinks broker 3's endOffset is 12, and its own endOffset is 18. The ISR is a set, so the printed order, like 2,1,3, means nothing. The two spaces in 1,2 and are where (under-min-isr) goes when the ISR is below the minimum (line 1818).
• Step 6: once the awakened broker 3 sent Fetches again and its LEO reached the HW, maybeExpandIsr() proposed growing the ISR and ISR updated to 2,1,3 was printed. The partition epoch (version) went up once on the shrink (0 → 1) and once on the expansion (1 → 2).
• Step 7: all three brokers have the same HW in replication-offset-checkpoint. Followers also write the HW they got from the leader to their own file.

5. What does acks=all wait for — DelayedProduce and min.insync.replicas

In part 6, a Produce request was written to the leader's log in ReplicaManager.appendRecords() and then handed to maybeAddDelayedProduce(). Whether to respond right away or wait is decided by this condition.

  // If all the following conditions are true, we need to put a delayed produce request and wait for replication to complete
  //
  // 1. required acks = -1
  // 2. there is data to append
  // 3. at least one partition append was successful (fewer errors than partitions)
  private def delayedProduceRequestRequired(requiredAcks: Short,
                                            entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
                                            localProduceResults: Map[TopicIdPartition, LogAppendResult]): Boolean = {
    requiredAcks == -1 &&
    entriesPerPartition.nonEmpty &&
    localProduceResults.values.count(_.exception().isPresent) < entriesPerPartition.size
  }

core/src/main/scala/kafka/server/ReplicaManager.scala lines 1356–1367

Reading the Scala — When a method's body is a single expression, its value is the return value. count(_.exception().isPresent) is "the number of results that have an exception".

acks=all travels on the wire as -1 (part 1). acks=0 and 1 are filtered out by this condition and answered as soon as the leader has written (with acks=0 no response is sent at all; the connection is closed only if there was an error, KafkaApis.scala lines 512–530). With -1, a DelayedProduce is created. The target it waits for is, per partition, requiredOffset = the last offset written + 1 (ReplicaManager.scala lines 838–840).

      val delayedProduce = new DelayedProduce(timeoutMs, initialProduceStatus.asJava, delegate, responseCallback.asJava)

      // create a list of (topic, partition) pairs to use as keys for this delayed produce operation
      val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toList

      // try to complete the request immediately, otherwise put it into the purgatory
      // this is because while the delayed produce operation is being created, new
      // requests may arrive and hence make this operation completable.
      delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys.asJava)

core/src/main/scala/kafka/server/ReplicaManager.scala lines 903–911

Reading the Scala — the _ in keys.map(new TopicPartitionOperationKey(_)) stands for each element, so it is the same as keys.map(k -> new TopicPartitionOperationKey(k)). .toList makes a Scala list, and .asJava turns a Scala collection into a Java one so it can be passed to a Java method.

tryCompleteElseWatch() is the same purgatory mechanism as part 8's DelayedFetch: try once, and if that doesn't work, hang it up with the partitions as keys. The time limit timeoutMs is the TimeoutMs field of the Produce request (KafkaApis.scala lines 548–549), and the producer's Sender puts request.timeout.ms (default 30000) into that field (Sender.java line 890 and lines 930–934). In other words, your app's request.timeout.ms is how long the broker will wait on acks=all for you.

When created, a DelayedProduce writes REQUEST_TIMED_OUT into the error code in advance.

        // first update the acks pending variable according to the error code
        produceStatus.forEach((topicPartition, status) -> {
            if (status.responseStatus.error == Errors.NONE) {
                // Timeout error state will be cleared when required acks are received
                status.acksPending = true;
                status.responseStatus.error = Errors.REQUEST_TIMED_OUT;
            } else {
                status.acksPending = false;
            }

server/src/main/java/org/apache/kafka/server/purgatory/DelayedProduce.java lines 110–118

When the condition is met, tryComplete() clears that error (lines 152–155); when time runs out, the pre-written REQUEST_TIMED_OUT goes into the response as it is. The condition is judged by Partition.checkEnoughReplicasReachOffset(), called through a delegate function that ReplicaManager passes in.

        val minIsr = effectiveMinIsr(leaderLog)
        if (leaderLog.highWatermark >= requiredOffset) {
          /*
           * The topic may be configured not to accept messages if there are not enough replicas in ISR
           * in this scenario the request was already appended locally and then added to the purgatory before the ISR was shrunk
           */
          if (minIsr <= curMaximalIsr.size)
            (true, Errors.NONE)
          else
            (true, Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND)
        } else
          (false, Errors.NONE)

core/src/main/scala/kafka/cluster/Partition.scala lines 970–981

Reading the Scala — (true, Errors.NONE) is a two-value tuple. The method returns "is it done, and with which error" as a pair.

What acks=all waits for is not "every replica in the ISR" but "the HW passing my record". They sound the same but aren't. As section 3 showed, the HW is computed from the ISR at that moment, so when a follower drops out of the ISR, the HW moves without waiting for it. The acks=all in example 1 finished not because broker 3 caught up but because broker 3 left the ISR. The guarantee of acks=all is "the ISR at response time has it all", and what sets the lower bound on the size of that ISR is min.insync.replicas.

That lower bound bites in two places. One is the NOT_ENOUGH_REPLICAS_AFTER_APPEND in the code above, for when the ISR shrank below the bound while waiting. In that case the record has already been written to the leader's log (the comment: "the request was already appended locally"). The other is before writing.

          val minIsr = effectiveMinIsr(leaderLog)
          val inSyncSize = partitionState.isr.size

          // Avoid writing to leader if there are not enough insync replicas to make it safe
          if (inSyncSize < minIsr && requiredAcks == -1) {
            throw new NotEnoughReplicasException(s"The size of the current ISR : $inSyncSize " +
              s"is insufficient to satisfy the min.isr requirement of $minIsr for partition $topicPartition, " +
              s"live replica(s) broker.id are : $inSyncReplicaIds")
          }

core/src/main/scala/kafka/cluster/Partition.scala lines 1233–1241

This is the start of Partition.appendRecordsToLeader(). If it's acks=all and the ISR is below the bound, the request is rejected without writing to the log. acks=1 is not subject to this check (the requiredAcks == -1 condition). min.insync.replicas defaults to 1 (ServerLogConfigs.java lines 158–159), so with default settings acks=all succeeds even when the leader is alone in the ISR. The producer's acks defaults to "all" (ProducerConfig.java lines 430–435), so "the defaults are safe" is only true once you raise min.insync.replicas on the broker side to 2 or more.

The red boxes are failures the producer receives; the green boxes are responses. Only the first rejection leaves the record unwritten.

Who makes the purgatory look at a DelayedProduce again is in the source too. The purgatory doesn't watch conditions by itself; somebody has to call checkAndComplete(partition). Those calls are the tryCompleteDelayedRequests() at the end of section 2's updateFollowerFetchState() (→ DelayedOperations.checkAndCompleteAll(), Partition.scala lines 69–76), the one after section 4's handleAlterPartitionUpdate() (lines 1714–1716), and the one when a write to the leader raised the HW (addCompletePurgatoryAction(), ReplicaManager.scala lines 853–876). In other words, what wakes up acks=all is usually the followers' Fetches (when it is released because the ISR shrank, as in example 1, the AlterPartition response wakes it). Timeouts are reaped by the ExpirationReaper-1-Produce thread we saw in part 2.

Example 2: turning off one broker with min.insync.replicas=3

ex/u09-min-isr/run.sh creates a topic strict with replicas 1:2:3 and min.insync.replicas=3, and runs a Java program (MinIsr.java) that sends one record with acks=all and one with acks=1 three times: at the start, after stopping broker 3 with kafka_stop_node 3 (SIGTERM, controlled shutdown), and after lowering the topic setting to min.insync.replicas=2. The producer is set to retries=0. NotEnoughReplicasException is a retriable error, so otherwise the producer would keep resending for delivery.timeout.ms (default 2 minutes) and end with a timeout.

            props.put(ProducerConfig.ACKS_CONFIG, acks);
            // Don't retry — NotEnoughReplicas is a retriable error, so otherwise the producer resends for delivery.timeout.ms
            props.put(ProducerConfig.RETRIES_CONFIG, "0");
            props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");

[shortened from ex/u09-min-isr/MinIsr.java]

=== 1. topic strict, replicas 1,2,3, min.insync.replicas=3
 Topic: strict Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr: 
acks=all -> ok, offset=0
acks=1   -> ok, offset=1
=== 2. stop broker 3 (SIGTERM, controlled shutdown)
 Topic: strict Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2 Elr: 3 LastKnownElr: 
acks=all -> NotEnoughReplicasException: Messages are rejected since there are fewer in-sync replicas than required.
acks=1   -> ok, offset=2
=== 3. min.insync.replicas=2
acks=all -> ok, offset=3
acks=1   -> ok, offset=4
=== 4. offsets on the leader's disk (broker 1)
offset: 0 payload: acks-all
offset: 1 payload: acks-1
offset: 2 payload: acks-1
offset: 3 payload: acks-all
offset: 4 payload: acks-1

• As soon as broker 3 was stopped, the ISR became 1,2. It was a controlled shutdown, so this is section 4's "controller shrinks it" path.
• In that state acks=all was rejected with NotEnoughReplicasException and acks=1 was accepted (offset 2). As the dump at the end shows, the rejected acks=all record is not in the log (offset 1 is followed directly by the acks=1 record at offset 2), because appendRecordsToLeader() threw before writing.
• After lowering the bound to 2, acks=all succeeded with the same ISR (1,2). min.insync.replicas is a topic setting, so it applies immediately without restarting brokers.
• The exception message is not the long sentence built by Partition ("The size of the current ISR : 2 is insufficient …") but the default message attached to Errors.NOT_ENOUGH_REPLICAS (Errors.java line 231). The producer's Sender builds the exception with response.error.exception(response.errorMessage) (Sender.java line 780), and uses this default message when the response carries no error message (Errors.java lines 484–488). This response carried none.
• Note the Elr: 3. The ELR (eligible leader replicas) is the list of replicas that left the ISR but are still eligible to become leader (the comment in the metadata record schema PartitionChangeRecord.json points to KIP-966). In this run, broker 3 went onto that list as the ISR shrank to 2, below the minimum of 3. The rules that fill the list belong to the controller's leader election, which this post doesn't follow.

6. When the leader changes — truncating a diverged log with leader epochs

The new leader records where its epoch begins and re-measures every follower from scratch

When the controller picks a new leader (leader epoch + 1), Partition.makeLeader() runs on the broker that receives that metadata. Here is the core of what it does when the epoch changed.

      if (isNewLeaderEpoch) {
        val leaderEpochStartOffset = leaderLog.logEndOffset
        // ...
        leaderLog.assignEpochStartOffset(partitionRegistration.leaderEpoch, leaderEpochStartOffset)

        // Initialize lastCaughtUpTime of replicas as well as their lastFetchTimeMs and
        // lastFetchLeaderLogEndOffset.
        remoteReplicas.foreach { replica =>
          replica.resetReplicaState(
            currentTimeMs,
            leaderEpochStartOffset,
            isNewLeader,
            isr.contains(replica.brokerId)
          )
        }

core/src/main/scala/kafka/cluster/Partition.scala lines 637–662

Reading the Scala — remoteReplicas.foreach { replica => ... } is Java's forEach(replica -> ...). A lambda in braces can be passed without parentheses.

The new leader writes "epoch N starts at my current LEO" into its epoch cache (assignEpochStartOffset()). That cache is LeaderEpochFileCache, stored as epoch startOffset lines in the leader-epoch-checkpoint file of the partition directory (LeaderEpochCheckpointFile.java line 45). Then it resets each follower's state with resetReplicaState(). When it has just become leader, it sets each follower's LEO to "unknown" and sets lastCaughtUpTimeMs to now only for followers that were in the ISR, so they aren't kicked out of the ISR right away (Replica.java lines 108–146).

On a broker that became a follower, makeFollower() clears the ISR information and records the new leader and epoch (lines 696–745), and ReplicaManager detaches the fetcher and reattaches it toward the new leader. Where to fetch from is decided here.

  protected def initialFetchOffset(log: UnifiedLog): Long = {
    if (log.latestEpoch.isPresent)
      log.logEndOffset
    else
      log.highWatermark
  }

core/src/main/scala/kafka/server/ReplicaManager.scala lines 2102–2107

If it has epoch information, it asks to fetch from its own LEO. The problem is that the tail of its log may differ from the new leader's. If the old leader had records accepted with acks=1 that only it holds, the new leader will be writing different records at those offsets.

The Fetch response tells it where the logs diverged

Section 1 said that a follower puts lastFetchedEpoch in its Fetch. Before reading, the leader uses that value to check whether the follower's log is on the same path as its own.

    lastFetchedEpoch.ifPresent { fetchEpoch =>
      val epochEndOffset = lastOffsetForLeaderEpoch(currentLeaderEpoch, fetchEpoch, fetchOnlyFromLeader = false)
      // ...
      if (epochEndOffset.leaderEpoch < fetchEpoch || epochEndOffset.endOffset < fetchOffset) {
        val divergingEpoch = new FetchResponseData.EpochEndOffset()
          .setEpoch(epochEndOffset.leaderEpoch)
          .setEndOffset(epochEndOffset.endOffset)

        return new LogReadInfo(
          FetchDataInfo.empty(fetchOffset),
          Optional.of(divergingEpoch),
          initialHighWatermark,
          initialLogStartOffset,
          initialLogEndOffset,
          initialLastStableOffset)
      }
    }

core/src/main/scala/kafka/cluster/Partition.scala lines 1386–1417

lastOffsetForLeaderEpoch() asks the leader's epoch cache "where did epoch E end?". The end is "the start offset of the epoch after E" (LeaderEpochFileCache.endOffsetFor(), lines 284–328). If the follower says "I have epoch-0 records up to just before offset 15" but the leader knows epoch 0 ended at 10, the follower's 10–14 are records the leader doesn't have. The leader then returns divergingEpoch(epoch 0, end 10) instead of records. In this case the follower's state isn't updated either (the !logReadInfo.divergingEpoch.isPresent condition in section 2's fetchRecords()).

When the follower's processFetchRequest() sees this response, it doesn't process records (lines 282–290); it decides the offset to truncate to via truncateOnFetchResponse() → truncationOffset().

    val replicaEndOffset = logEndOffset(tp)

    // get (leader epoch, end offset) pair that corresponds to the largest leader epoch
    // less than or equal to the requested epoch.
    val endOffsetForEpochOpt = endOffsetForEpoch(tp, leaderEpochOffset.leaderEpoch)
    if (endOffsetForEpochOpt.isPresent) {
      val offsetAndEpoch = endOffsetForEpochOpt.get
      val followerEpochEndOffset = offsetAndEpoch.offset
      val followerEpoch = offsetAndEpoch.epoch()
      if (followerEpoch != leaderEpochOffset.leaderEpoch) {
        // ...
      } else {
        val offsetToTruncateTo = min(followerEpochEndOffset, leaderEpochOffset.endOffset)
        min(offsetToTruncateTo, replicaEndOffset)
      }

core/src/main/scala/kafka/server/AbstractFetcherThread.scala lines 507–529

Reading the Scala — min(a, b) is the function from scala.math._, imported at the top of the file. if (...) { ... } else { ... } is an expression that yields a value, so the last line, min(offsetToTruncateTo, replicaEndOffset), is directly the offset to truncate to.

It truncates to the smallest of the end the leader gave, the end of that epoch as the follower itself knows it, and the follower's LEO (truncate() → Partition.truncateTo()), and fetches again from that offset. If the follower doesn't know the epoch the leader named (the branch elided here), it truncates to the end of the earlier epoch it does know, and if the leader reports a divergence again on the next Fetch, it truncates once more (the comment on lines 489–504). A log with no epoch information at all (empty, or in a very old format) can't use this method, so partitionFetchState() puts that partition in the TRUNCATING state and section 1's maybeTruncate() truncates it to the HW (lines 144–175 and 427–436).

Drawn with the real numbers from example 3. It starts with broker 1 holding lost-1 and broker 2 holding new-1 at the same offset, 10.

What changed since 4.4.0: the path that asked with OffsetsForLeaderEpoch is gone

In 4.4.0, AbstractFetcherThread had a path that asked the leader for the truncation point with a separate OffsetsForLeaderEpoch request (truncateToEpochEndOffsets()). But the RemoteLeaderEndPoint that followers use already had isTruncationOnFetchSupported = true in 4.4.0, so partitions with epoch information were truncated by the Fetch response's divergingEpoch, as in this post, even then. The separate path remained for the endpoint that didn't support it — LocalLeaderEndPoint (false), which moves logs between directories inside one broker (checked in the 4.4.0 source). Commit b234a3a667 from 2026-10-08, 16 days before our reference commit, "KAFKA-18477 Remove usage of OffsetForLeaderEpochRequest in AbstractFetcherThread", removed that path. So in this commit maybeTruncate() is left with only the truncate-to-HW case, and LeaderEndPoint.fetchEpochEndOffsets() is used only by tiered storage's TierStateMachine. The side where a broker receives and handles OffsetsForLeaderEpoch requests (KafkaApis.handleOffsetForLeaderEpochRequest(), line 2196) is still there; consumers also use that request to check that their offset wasn't truncated after a rebalance (the comment in endOffsetFor(), lines 294–296).

Example 3: acks=1 records that only the leader had disappear

ex/u09-truncation/run.sh writes ok-1–ok-10 with acks=all to a topic div with replicas 1:2:3, then runs the time-sensitive part in about 2 seconds in a Java program, Diverge.java: SIGSTOP the two followers (brokers 2 and 3) → wait 1 second → lost-1–lost-5 with acks=1 → kill -9 the leader (broker 1) → SIGCONT the two followers. Then it waits for a new leader to be elected, writes new-1–new-3 with acks=all, and restarts broker 1. replica.lag.time.max.ms is set to 10 seconds so the leader won't drop the followers from the ISR before it dies.

At first I did this sequence from the shell with the console producer, but while the console producer took more than 10 seconds to start, the leader dropped the frozen followers from the ISR. Then the leader died with itself as the only ISR member, giving Leader: none … Isr: Elr: 1, and the restarted broker 1 came back as leader, so the lost-* records survived. That was the rule "only a replica in the ISR can become leader" doing its job, but it wasn't what this example meant to show, so I switched to a Java program.

Even after switching to Java, one run had lost-1 replicated to the followers. It looks like this: the Fetch the follower sent just before freezing waits in the leader's purgatory for up to replica.fetch.wait.max.ms (500ms); if lost-1 is written meanwhile, that Fetch completes carrying lost-1, the response lands in the frozen follower's socket buffer, and the follower writes it down once it wakes up (this is a guess explained from the source's DelayedFetch; I didn't confirm it at the packet level). So the program now waits 1 second after freezing before writing, and I ran it twice afterwards with the same result both times, shown below.

The batch lines are kafka-dump-log.sh --print-data-log output shortened with sed, and leader-epoch-checkpoint is the file's contents joined onto one line (version 0, the line count, then epoch startOffset pairs). The ok-2–ok-9 record lines are shortened to …. The time since the freeze (ms after SIGSTOP) differs on every run.

=== 1. topic div, replicas 1,2,3 (leader 1), 10 records with acks=all
 Topic: div Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr: 
--- leader-epoch-checkpoint (version, count, then 'epoch startOffset')
broker 1: 0 1 0 0 
broker 2: 0 1 0 0 
broker 3: 0 1 0 0 
=== 2. Java: SIGSTOP followers 2,3 -> 5 records with acks=1 -> kill -9 leader 1 -> SIGCONT 2,3
acks=1 offset=10 (only the leader has it)
acks=1 offset=11 (only the leader has it)
acks=1 offset=12 (only the leader has it)
acks=1 offset=13 (only the leader has it)
acks=1 offset=14 (only the leader has it)
leader killed, followers resumed (1171 ms after SIGSTOP)
--- broker 1 log (process is dead, the file is still there)
  batch offsets 0-9 (count 10) partitionLeaderEpoch=0
    offset 0: ok-1
    …
    offset 9: ok-10
  batch offsets 10-10 (count 1) partitionLeaderEpoch=0
    offset 10: lost-1
  batch offsets 11-11 (count 1) partitionLeaderEpoch=0
    offset 11: lost-2
  batch offsets 12-12 (count 1) partitionLeaderEpoch=0
    offset 12: lost-3
  batch offsets 13-13 (count 1) partitionLeaderEpoch=0
    offset 13: lost-4
  batch offsets 14-14 (count 1) partitionLeaderEpoch=0
    offset 14: lost-5
=== 3. wait for a new leader, then 3 records with acks=all
 Topic: div Partition: 0 Leader: 2 Replicas: 1,2,3 Isr: 2,3 Elr: LastKnownElr: 
--- broker 2 log (new leader)
  batch offsets 0-9 (count 10) partitionLeaderEpoch=0
    offset 0: ok-1
    …
    offset 9: ok-10
  batch offsets 10-12 (count 3) partitionLeaderEpoch=1
    offset 10: new-1
    offset 11: new-2
    offset 12: new-3
broker 1: 0 1 0 0 
broker 2: 0 2 0 0 1 10 
broker 3: 0 2 0 0 1 10 
=== 4. restart broker 1
 Topic: div Partition: 0 Leader: 2 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr: 
--- broker 1 server.log
INFO [ReplicaFetcher replicaId=1, leaderId=2, fetcherId=0] Truncating partition div-0 to offset 10 due to leader epoch and offset EpochEndOffset(errorCode=0, partition=0, leaderEpoch=0, endOffset=10) (kafka.server.ReplicaFetcherThread)
INFO [UnifiedLog partition=div-0, dir=/tmp/kafka/1/data] Truncating to offset 10 (org.apache.kafka.storage.internals.log.UnifiedLog)
--- broker 1 log after truncation
  batch offsets 0-9 (count 10) partitionLeaderEpoch=0
    offset 0: ok-1
    …
    offset 9: ok-10
  batch offsets 10-12 (count 3) partitionLeaderEpoch=1
    offset 10: new-1
    offset 11: new-2
    offset 12: new-3
broker 1: 0 2 0 0 1 10 
broker 2: 0 2 0 0 1 10 
broker 3: 0 2 0 0 1 10 
--- console consumer from the beginning
ok-1 ok-2 ok-3 ok-4 ok-5 ok-6 ok-7 ok-8 ok-9 ok-10 new-1 new-2 new-3 

• Step 2: only leader broker 1's log has lost-1–lost-5 (offsets 10–14, epoch 0). Because it's acks=1, the producer got a success response for all five.
• Step 3: after the controller fenced broker 1 (its heartbeats stopped, and broker.session.timeout.ms is 9 seconds), it picked broker 2, still in the ISR, as the new leader. 1 10 — "epoch 1 starts at offset 10" — was added to the epoch caches of brokers 2 and 3, and new-1–new-3 were written at offsets 10–12 with epoch 1. At the same offset 10, broker 1 has lost-1 and broker 2 has new-1.
• Step 4: the restarted broker 1 logged Truncating partition div-0 to offset 10 due to leader epoch and offset EpochEndOffset(… leaderEpoch=0, endOffset=10). That is the log line in maybeTruncateToEpochEndOffsets() (line 208), and the part in parentheses is the divergingEpoch the new leader sent in the Fetch response. Then UnifiedLog logged Truncating to offset 10. After truncation the log matches the leader's, with new-1–new-3 (epoch 1), and the epoch cache matches too.
• The console consumer: reading from the beginning gives only the 10 ok-* records and the 3 new-* records. The five lost-* records, for which the producer got success responses, are left nowhere in the cluster. This is what acks=1's "written on the leader" guarantee means when the leader dies. Had the same thing been done with acks=all, the HW wouldn't have moved while the followers were frozen, and the producer would never have received a success response (example 1).

What this post leaves out

• Unclean leader election (unclean.leader.election.enable, default false) and the ELR (eligible leader replicas) and LastKnownElr: the controller's metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java. The comment on the follower side's fetchOffsetAndTruncate() (AbstractFetcherThread.scala lines 546–616) explains the aftermath
• Reading from a replica that isn't the leader (replica.selector.class, findPreferredReadReplica()): ReplicaManager.scala lines 1813–1833
• Replication quotas (leader.replication.throttled.rate and friends, shouldFollowerThrottle()): RemoteLeaderEndPoint.scala lines 224–230, server/src/main/java/org/apache/kafka/server/quota/
• ReplicaAlterLogDirsThread and LocalLeaderEndPoint, which move logs between directories inside one broker, and Partition's futureLog: core/src/main/scala/kafka/server/
• Catching up from tiered storage (TierStateMachine, OFFSET_MOVED_TO_TIERED_STORAGE): core/src/main/java/kafka/server/TierStateMachine.java
• Partition reassignment (AddingReplicas/RemovingReplicas, OngoingReassignmentState): Partition.updateAssignmentAndIsr() (lines 837–862)
• Incremental fetch sessions (FetchSessionHandler): followers use the same sessions as consumers. Part 8
• Replication of the metadata log itself (KRaft): the same "copy with Fetch and commit with an HW", but a different implementation. Part 10
• Metrics (UnderReplicated, UnderMinIsr, IsrShrinksPerSec, MaxLag and so on): Partition.scala lines 221–226, AbstractFetcherManager.scala lines 48–65

Summary

What's next

Throughout this post we kept saying "the controller decides" and "it spreads through the metadata". That is the side that picks leaders, receives AlterPartition and writes PartitionChangeRecords, and fences brokers. The next post goes to that controller. Interestingly, the controller's metadata log, __cluster_metadata, is also copied by sending Fetch requests to a leader, and commits by treating the point a majority has received as its high watermark — a Raft majority, not this post's ISR. In Part 10: KRaft and Metadata — How the Controller Remembers the Cluster we look at KafkaRaftClient's role transitions, how QuorumController builds records on a single-threaded event queue, and how a broker takes those records all the way to this post's makeLeader() and makeFollower().

If you'd like to compare replication with another system, the MariaDB series covers how a database keeps a record of its changes and recovers from it.

• Part 1: How I Read It with AI, and the Map — the whole picture of one record's journey
• Part 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Part 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path
• Part 7: Log Storage — Segments, Indexes, Retention and Compaction
• Part 8: The Read Path — Fetch Requests, Long Polling and Zero-Copy
• Part 12: Idempotence and Transactions — How Exactly-Once Is Built — why acks=all is a prerequisite of the idempotent producer