Pull-Based Replication, ISR, and High Watermark in a Distributed Log

In this section, you'll implement the replication thread (followers pull data from the leader over a long-lived, reused connection), ISR tracking (the leader monitors which replicas are in sync), and high watermark advancement (consumers see data only after it is replicated). All of this lives in broker/topic/replication.go.

This chapter covers replication end to end without repeating what's already covered on the previous page: HandleProduce/advanceHighWatermark/waitForAllFollowersToCatchUp (the producer-side half of ISR/HW handling) live in broker/topic/producer.go and were covered there. This page is about the other half — how a follower actually pulls data, and how the leader tracks and reacts to what it pulled.

#Step 1: Understand pull-based replication model

#Why pull instead of push

This system uses pull-based replication, similar to Kafka:

1. The follower periodically calls FetchBatch on the leader, starting from its current LEO.
2. The leader reads from its log using ReadUncommitted (up to LEO, not HW) and returns records.
3. The leader records the follower's LEO for ISR computation.
4. The follower appends the received records to its local log in a batch.
5. After all ISR replicas have caught up, the leader advances HW.

Why pull, not push? Pull gives each follower control over its own pace. A slow follower does not block the leader's write path. The leader just serves fetch requests when they arrive.

#Step 2: Two independent loops

#Why replication and reconciliation are separate

Every node runs two independent background loops. Keeping them separate means a slow or unreachable replication peer (replicateAllTopics can legitimately block on network I/O) can never delay reconciliation, which the rest of the system depends on staying fast (see the previous page — callers right after CreateTopic rely on the reconcile tick being quick):

// broker/topic/replication.go
package topic

const (
    DefaultReplicationBatchSize = 5000
    replicationTickInterval     = 1 * time.Second
    // reconcileTickInterval drives reconcileLocalTopics (topic.go): opening/closing
    // local logs in reaction to cluster metadata changes. Deliberately much faster
    // than replicationTickInterval — it's a cheap in-memory diff against cluster
    // metadata, only touching disk when something actually changed.
    reconcileTickInterval = 50 * time.Millisecond
)

type ReplicaTopicInfo struct {
    TopicName    string
    LeaderNodeID string
}

#Starting and stopping both loops

Start/stop are now built on a context.CancelFunc instead of a manually-managed stop channel — simpler, and the same pattern used elsewhere in the codebase:

func (tm *TopicManager) StartReplicationThread() {
    tm.mu.Lock()
    if tm.replicationCancel != nil {
        tm.mu.Unlock()
        return
    }
    ctx, cancel := context.WithCancel(context.Background())
    tm.replicationCancel = cancel
    tm.mu.Unlock()
    go tm.runReplicationThread(ctx)
}

func (tm *TopicManager) StopReplicationThread() {
    tm.mu.Lock()
    cancel := tm.replicationCancel
    tm.replicationCancel = nil
    tm.mu.Unlock()
    if cancel != nil {
        cancel()
    }
}

runReplicationThread launches the two loops on independent goroutines and waits for both to exit before cleaning up cached connections:

func (tm *TopicManager) runReplicationThread(ctx context.Context) {
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        tm.runReconcileLoop(ctx)
    }()
    go func() {
        defer wg.Done()
        tm.runReplicateLoop(ctx)
    }()
    wg.Wait()
    tm.replConns.closeAll()
}

func (tm *TopicManager) runReconcileLoop(ctx context.Context) {
    ticker := time.NewTicker(reconcileTickInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            tm.reconcileLocalTopics()
        }
    }
}

func (tm *TopicManager) runReplicateLoop(ctx context.Context) {
    ticker := time.NewTicker(replicationTickInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            tm.replicateAllTopics(ctx)
        }
    }
}

reconcileLocalTopics (opening/closing local logs to track cluster metadata) was covered on the previous page — every node runs it, whether or not it replicates anything, which is why it's on its own 50ms loop here rather than folded into the replication tick. The rest of this page is about runReplicateLoop and everything downstream of it.

#Step 3: Organize replication with per-leader goroutine pools

#Why per-leader goroutines

A critical design choice: topics are grouped by leader, and each leader gets its own goroutine. A slow or dead leader blocks only its own goroutine — other leaders continue unaffected.

#Identifying topics to replicate

ListReplicaTopics scans this node's open logs and returns the ones where this node is not the leader — i.e. the topics it needs to replicate, asking the coordinator for each log's current leader:

// broker/topic/topic.go
func (tm *TopicManager) ListReplicaTopics() []ReplicaTopicInfo {
    tm.mu.RLock()
    names := make([]string, 0, len(tm.Topics))
    currentNodeID := tm.CurrentNodeID
    for name, l := range tm.Topics {
        if l != nil {
            names = append(names, name)
        }
    }
    tm.mu.RUnlock()

    var out []ReplicaTopicInfo
    for _, name := range names {
        leaderID, ok := tm.coordinator.TopicLeaderNodeID(name)
        if !ok || leaderID == currentNodeID {
            continue
        }
        out = append(out, ReplicaTopicInfo{TopicName: name, LeaderNodeID: leaderID})
    }
    return out
}

The return type is minimal — just TopicName and LeaderNodeID. The actual LEO is fetched on demand inside ReplicateFromLeader.

#Launching worker goroutines per leader

replicateAllTopics groups topics by leader and launches one goroutine per leader:

// broker/topic/replication.go
func (tm *TopicManager) replicateAllTopics(ctx context.Context) {
    leaderToTopics := make(map[string][]string)
    for _, info := range tm.ListReplicaTopics() {
        leaderToTopics[info.LeaderNodeID] = append(leaderToTopics[info.LeaderNodeID], info.TopicName)
    }
    if len(leaderToTopics) == 0 {
        return
    }

    batchSize := tm.replicationBatchSize
    if batchSize == 0 {
        batchSize = DefaultReplicationBatchSize
    }

    var wg sync.WaitGroup
    for leaderID, topicNames := range leaderToTopics {
        wg.Add(1)
        go func(leaderID string, topicNames []string) {
            defer wg.Done()
            if err := tm.ReplicateFromLeader(ctx, leaderID, topicNames, batchSize); err != nil {
                tm.Logger.Warn("replication from leader failed", zap.String("leader_id", leaderID), zap.Error(err))
            }
        }(leaderID, topicNames)
    }
    wg.Wait()
}

sync.WaitGroup ensures the tick handler waits for all goroutines to finish before the next tick fires. Errors are logged but do not crash the loop — the next tick will retry.

#Step 4: Reuse connections across replication ticks

#replicationConnCache

Opening a fresh TCP connection and handshake for every fetch would be real, avoidable overhead for what's normally a steady-state relationship between a follower and its leader. Instead, one ConsumerClient per leader is kept alive across ticks in a small cache owned by TopicManager:

// broker/topic/replication.go
type replicationConnCache struct {
    mu      sync.Mutex
    clients map[string]*client.ConsumerClient
}

func newReplicationConnCache() *replicationConnCache {
    return &replicationConnCache{clients: make(map[string]*client.ConsumerClient)}
}

// get returns the cached connection for leaderID, dialing and caching one if absent.
func (c *replicationConnCache) get(network transport.Network, leaderID, rpcAddr, currentNodeID string) (*client.ConsumerClient, error) {
    c.mu.Lock()
    defer c.mu.Unlock()
    if cc, ok := c.clients[leaderID]; ok {
        return cc, nil
    }
    cc, err := client.NewConsumerClientOverNetwork(network, rpcAddr)
    if err != nil {
        return nil, err
    }
    cc.SetReplicaNodeID(currentNodeID)
    // This loop already retries at tick granularity (abort this leader, try again next
    // tick); the client's own retry-on-topic-not-ready would just stack another retry
    // budget on top.
    cc.DisableTopicNotReadyRetry()
    c.clients[leaderID] = cc
    return cc, nil
}

// invalidate drops and closes the cached connection for leaderID, so the next get
// redials — used when a call reports the connection/leader is actually gone.
func (c *replicationConnCache) invalidate(leaderID string) {
    c.mu.Lock()
    cc := c.clients[leaderID]
    delete(c.clients, leaderID)
    c.mu.Unlock()
    if cc != nil {
        _ = cc.Close()
    }
}

// closeAll closes every cached connection — called when replication stops.
func (c *replicationConnCache) closeAll() {
    c.mu.Lock()
    clients := c.clients
    c.clients = make(map[string]*client.ConsumerClient)
    c.mu.Unlock()
    for _, cc := range clients {
        _ = cc.Close()
    }
}

ConsumerClient here is consumer/client.ConsumerClient — the same low-level, single-connection client type the consumer CLI uses (covered in the next chapter), just configured differently: SetReplicaNodeID marks fetches as replica reads (server-side, this means "read via ReadUncommitted, up to LEO, not just HW" — see the transport/RPC chapter), and DisableTopicNotReadyRetry turns off the client's own built-in retry-on-not-ready loop, because the replication tick loop already provides equivalent retry behavior one level up (abandon this leader for the current tick, try again next tick) — stacking a second retry budget underneath would just add latency without adding value.

The connection is dialed through tm.network, a transport.Network (set with TopicManager.SetNetwork, nil meaning the real network). Raft and Serf dial through the same seam (see the Raft and Serf pages), so every node-to-node connection in the broker goes through one interface. That's what lets the chaos tests on the final page partition nodes from each other without touching the operating system's network.

Because the cache is keyed by leader ID rather than by call, concurrent replication goroutines for the same leader (multiple topics being replicated from one leader, in the same tick) safely share one connection — the mutex in get/invalidate is only held for the map operation itself, not for the network calls that use the returned client.

The cache's lifecycle in one place: get dials on first use and reuses on every tick after; invalidate(leaderID) drops and closes a connection so the next tick redials, used when a call reports the connection is actually gone; closeAll() tears everything down once both loops exit.

#Step 5: Fetch records from a leader

#The ReplicateFromLeader engine

ReplicateFromLeader is the core of the replication engine. For each leader, it now reuses (or opens, on first use) this node's cached connection, then:

1. Iterates topics in a loop, removing topics that are caught up.
2. Sends its latest leader epoch with every fetch, and truncates its log when the leader reports that it diverged (Step 7).
3. Uses batch writes (ApplyRecordBatch) instead of per-record appends, then records the leader's epoch boundaries for those records.
4. Handles connection errors (invalidating the cache entry) and offset-not-found errors gracefully.

// broker/topic/replication.go
func (tm *TopicManager) ReplicateFromLeader(ctx context.Context, leaderID string, topicNames []string, batchSize uint32) error {
    rpcAddr, ok := tm.coordinator.NodeRPCAddr(leaderID)
    if !ok {
        return fmt.Errorf("leader node %s not found", leaderID)
    }

    cc, err := tm.replConns.get(tm.network, leaderID, rpcAddr, tm.currentNodeID())
    if err != nil {
        return fmt.Errorf("connect to leader %s at %s: %w", leaderID, rpcAddr, err)
    }

    consumerID := fmt.Sprintf("replicate-%s-%s", tm.currentNodeID(), leaderID)

    // Fetch each topic in a loop. A topic is done when FetchBatch returns fewer than batchSize entries.
    pending := make([]string, len(topicNames))
    copy(pending, topicNames)

    for len(pending) > 0 {
        if ctx.Err() != nil {
            return ctx.Err()
        }

        next := make([]string, 0, len(pending))
        for _, topicName := range pending {
            if ctx.Err() != nil {
                return ctx.Err()
            }

            l := tm.lookupTopic(topicName)
            if l == nil {
                continue
            }
            leo := l.LEO()
            lastEpoch := l.Epochs.LatestEpoch()

            resp, err := cc.FetchBatch(ctx, &protocol.FetchBatchRequest{
                Topic:            topicName,
                Id:               consumerID,
                Offset:           leo,
                MaxCount:         batchSize,
                LastFetchedEpoch: lastEpoch,
            })
            if err != nil {
                var rpcErr *protocol.RPCError
                if errors.As(err, &rpcErr) && rpcErr.Code == protocol.CodeReadOffset {
                    // Caught up — skip this topic.
                    continue
                }
                if protocol.ShouldReconnect(err) {
                    // Connection gone — invalidate it so the next tick redials, and
                    // abort this leader entirely for now; next tick will retry.
                    tm.replConns.invalidate(leaderID)
                    return fmt.Errorf("connection lost to leader %s: %w", leaderID, err)
                }
                // Transient error — leave this topic for the next tick rather than
                // re-fetching immediately, which would spin against the leader for as
                // long as the error lasts.
                continue
            }

            if resp.Diverging != nil {
                // Our log holds records the leader doesn't: cut back to where the two
                // agree, then fetch again from there in the next round.
                if err := tm.truncateDivergentTail(topicName, l, lastEpoch, resp.Diverging); err != nil {
                    tm.Logger.Warn("truncate divergent tail failed", zap.String("topic", topicName), zap.Error(err))
                    continue
                }
                next = append(next, topicName)
                continue
            }

            if len(resp.Entries) > 0 {
                values := make([][]byte, 0, len(resp.Entries))
                for _, entry := range resp.Entries {
                    if entry != nil {
                        values = append(values, entry.Value)
                    }
                }
                if err := tm.ApplyRecordBatch(topicName, values); err != nil {
                    next = append(next, topicName)
                    continue
                }
                tm.recordEpochStarts(topicName, l, resp.EpochStarts)
            }

            // If we got a full batch, there's likely more data — keep fetching.
            if uint32(len(resp.Entries)) >= batchSize {
                next = append(next, topicName)
            }
        }
        pending = next
    }
    return nil
}

The leader's RPC address comes from tm.coordinator.NodeRPCAddr(leaderID) — Serf-backed, reached through Cluster — consistent with every other cluster-state lookup in this package going through TopicCoordinator.

#The pending-slice pattern for termination

Instead of a fixed round limit, a pending slice shrinks as topics catch up:

1. Start with all topic names in pending.
2. For each topic, fetch a batch. If fewer than batchSize entries are returned, the topic is caught up — drop it from pending.
3. If a full batch was returned, there might be more data — keep the topic in pending for another round.
4. If the leader reported a divergence, the follower truncates and keeps the topic in pending, so it fetches again from its new LEO in the same tick.
5. A topic that hits a transient error is dropped until the next tick, so a leader that's briefly unable to serve it isn't hammered with back-to-back requests.

This naturally terminates when all topics are caught up, and handles the common case (small lag) in a single pass.

#Handling fetch errors gracefully

A fetch can end four ways besides returning records:

Outcome	Handling
CodeReadOffset (offset out of range)	Topic is caught up. Remove from pending.
ShouldReconnect (connection lost)	replConns.invalidate(leaderID), then abort the entire leader goroutine. Next tick redials and retries.
Diverging set in the response	Truncate the local log (Step 7), keep the topic in pending.
Transient error	Drop the topic until the next tick.

#Writing fetched records in batches

The fetched entries are applied to the local log in a single batch write:

// broker/topic/topic.go
func (tm *TopicManager) ApplyRecordBatch(topicName string, values [][]byte) error {
    if len(values) == 0 {
        return nil
    }
    l := tm.lookupTopic(topicName)
    if l == nil {
        return nil
    }
    _, err := l.AppendBatch(values)
    return err
}

AppendBatch writes all records to the segment in a single call, which is significantly faster than calling Append N times (fewer syscalls, one lock acquisition).

#Step 6: Track and update in-sync replica status

#Recording replica LEO and ISR membership

When a follower fetches data, the leader records its LEO. This happens in the fetch handler (RPC side calls RecordReplicaLEOFromFetch). The ISR math itself lives on cluster.TopicMetadata (previous page), so this method is mostly plumbing between the local log and the coordinator:

// broker/topic/topic.go
func (tm *TopicManager) RecordReplicaLEOFromFetch(ctx context.Context, topicName, replicaNodeID string, leo int64) error {
    l := tm.lookupTopic(topicName)
    var localLEO uint64
    if l != nil {
        localLEO = l.LEO()
    }
    lagThreshold := tm.ISRLagThreshold
    if lagThreshold == 0 {
        lagThreshold = DefaultISRLagThreshold
    }
    isr, ok := tm.coordinator.RecordReplicaFetch(topicName, replicaNodeID, leo, lagThreshold, localLEO)
    if !ok {
        return nil
    }
    if l != nil {
        l.SetHighWatermark(tm.coordinator.TopicMinISRLeo(topicName, localLEO))
    }
    return tm.coordinator.ApplyIsrUpdateEventInternal(topicName, replicaNodeID, isr)
}

The replica map, the LEO comparison, and the threshold logic all live behind tm.coordinator.RecordReplicaFetch, which calls straight through to cluster.TopicMetadata.RecordReplicaFetch (shown on the previous page):

// broker/cluster/cluster_metadata.go — for reference, this is what RecordReplicaFetch actually does
func (t *TopicMetadata) RecordReplicaFetch(nodeID string, leo int64, lagThreshold uint64, leaderLEO uint64) (isr bool) {
    // ... update rs.LEO, compare against lagThreshold, set rs.IsISR ...
}

A few things worth noting about this method:

1. New replicas start in ISR — a first-seen replica's ReplicaState starts with IsISR: true.
2. ISR threshold is configurable — defaults to DefaultISRLagThreshold (100 records), tunable via TopicManager.ISRLagThreshold.
3. Small-topic optimization — when the leader's LEO is smaller than the lag threshold, all replicas with non-negative LEO are considered in-sync, preventing false ISR demotions on lightly-loaded topics.
4. ISR changes propagated via Raft — ApplyIsrUpdateEventInternal submits a Raft log entry so all nodes agree on the ISR set.
5. HW advanced after every fetch — via TopicMinISRLeo.

#ISR membership computation logic

A replica is in the ISR if its LEO is within the threshold of the leader's LEO:

Leader LEO = 1500
ISR Threshold = 100

Replica A: LEO = 1480 → in ISR  (1480 >= 1500-100=1400)
Replica B: LEO = 1300 → NOT ISR (1300 < 1500-100=1400)

When a replica falls out of ISR, a Raft event updates the cluster state so all nodes agree on the ISR set.

#Expiring replicas that stop fetching

RecordReplicaLEOFromFetch only re-evaluates a replica when that replica fetches. A replica that has died, or been partitioned away, never fetches again. Its last-known IsISR: true and LEO would then freeze, pinning the high watermark at that LEO forever. So on every replication tick, the leader also demotes any ISR replica that hasn't fetched within ISRLagTime (30 seconds by default, matching Kafka's replica.lag.time.max.ms):

// broker/topic/topic.go
func (tm *TopicManager) expireStaleISR() {
    // Shrinking the ISR is a metadata decision; only act on one with the metadata
    // quorum behind it.
    if !tm.coordinator.IsRaftReady() {
        return
    }
    lagTime := tm.ISRLagTime
    if lagTime <= 0 {
        lagTime = DefaultISRLagTime
    }
    for _, topicName := range tm.coordinator.TopicNames() {
        isLeader, err := tm.IsLeader(topicName)
        if err != nil || !isLeader {
            continue
        }
        expired := tm.coordinator.ExpireStaleISR(topicName, lagTime)
        if len(expired) == 0 {
            continue
        }
        if l := tm.lookupTopic(topicName); l != nil {
            l.SetHighWatermark(tm.coordinator.TopicMinISRLeo(topicName, l.LEO()))
        }
        for _, nodeID := range expired {
            _ = tm.coordinator.ApplyIsrUpdateEventInternal(topicName, nodeID, false)
        }
    }
}

The quorum check at the top matters more than it looks. Picture a leader partitioned away from both of its followers. From where it sits, the followers have simply stopped fetching, exactly as if they had died. If it demoted them, its ISR would be just itself, and its high watermark would jump to its own LEO. Consumers reading from it would then see records that the rest of the cluster never received, and those records are discarded as soon as the partition heals (Step 7). A partitioned node loses its Raft leader within a heartbeat timeout, so with the check in place it holds its ISR steady instead, and its high watermark stays at what the followers actually have. Kafka enforces the same rule by requiring the controller to accept an ISR shrink before the leader acts on it. The check relies on ISRLagTime being much longer than Raft's heartbeat timeout (about one second), which the 30-second default comfortably is.

#Step 7: Detect and truncate divergent logs with leader epochs

#How a follower's log diverges

Fetching from your own LEO assumes your log is a prefix of the leader's. Usually it is, but one failure breaks that assumption: a leader that appends records its followers never receive, and then loses leadership.

Walk through it. Node A leads a topic in epoch 1. All three replicas hold offsets 0–9. A accepts five AckLeader writes at offsets 10–14 (or five AckAll writes still waiting on followers), and crashes before anyone fetches them. B is elected leader for epoch 2 and assigns offsets 10, 11, 12... to new records. When A restarts as a follower, its LEO is 15. Its next fetch asks B for offset 15. B's log has grown past 15, so it returns records from there on, and A appends them.

A's log now has the right length but the wrong records at offsets 10–14, and nothing will ever correct them. Worse, A reports a LEO that looks caught up, so B counts it as in sync. If A ever leads again, consumers read records that no one else has.

Offsets alone can't reveal the problem. The follower needs to know which leader wrote each range of its log.

#The leader epoch cache

Every leader change already carries an epoch number, bumped by one each time (ReassignLeadersForDeadNode, on the topic management page). Each epoch has exactly one leader, and every replica copies that leader's records. So if two replicas agree that epoch e started at the same offset, they hold the same records for as far as both have epoch e. Each log keeps its history of leadership as a short list of (epoch, start offset) pairs:

// broker/log/epoch.go
type EpochEntry struct {
    Epoch       int64
    StartOffset uint64
}

type LeaderEpochCache struct {
    mu      sync.RWMutex
    path    string // <topic dir>/leader-epoch-checkpoint
    entries []EpochEntry
}

// Assign records that epoch's records start at startOffset. Epochs only move forward:
// anything at or below the latest recorded epoch is ignored.
func (c *LeaderEpochCache) Assign(epoch int64, startOffset uint64) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    if epoch <= 0 {
        return nil
    }
    if n := len(c.entries); n > 0 && epoch <= c.entries[n-1].Epoch {
        return nil
    }
    c.entries = append(c.entries, EpochEntry{Epoch: epoch, StartOffset: startOffset})
    return c.persistLocked()
}

LogManager opens the cache alongside its segments (lm.Epochs), and persistLocked rewrites the checkpoint file atomically: write a temp file, fsync it, rename it over the old one. A restarted node therefore still knows which epochs its records came from.

The cache answers one question, the same one Kafka's OffsetForLeaderEpoch request asks: where does epoch e end in this log?

// EndOffsetFor finds the largest recorded epoch at or below epoch and returns it with
// the offset just past its last record: the next entry's start offset, or leo if it
// is the latest epoch.
func (c *LeaderEpochCache) EndOffsetFor(epoch int64, leo uint64) (int64, uint64) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    if len(c.entries) == 0 {
        return epoch, leo
    }
    i := sort.Search(len(c.entries), func(i int) bool { return c.entries[i].Epoch > epoch }) - 1
    if i < 0 {
        return 0, c.entries[0].StartOffset
    }
    if i == len(c.entries)-1 {
        return c.entries[i].Epoch, leo
    }
    return c.entries[i].Epoch, c.entries[i+1].StartOffset
}

#The leader records its epoch before writing

A leader adds its own entry the first time it acts as leader in a new epoch. Rather than hooking every leadership change, it checks lazily before each append and each replication fetch it serves:

// broker/topic/epoch.go
func (tm *TopicManager) ensureLeaderEpoch(topicName string, l *log.LogManager) {
    info, ok := tm.coordinator.TopicInfo(topicName)
    if !ok || info.LeaderNodeID != tm.currentNodeID() {
        return
    }
    if info.LeaderEpoch <= l.Epochs.LatestEpoch() {
        return
    }
    if err := l.Epochs.Assign(info.LeaderEpoch, l.LEO()); err != nil {
        tm.Logger.Warn("persist leader epoch failed", ...)
    }
}

HandleProduce and HandleProduceBatch call it before appending, so no record is ever written under an epoch the cache doesn't know about.

#Followers learn epoch boundaries from fetches

A follower's cache has to match the leader's for every record it holds, so the leader sends its epoch boundaries along with the records. The replication fetch carries two new fields:

// api/protocol/types.go
type FetchBatchRequest struct {
    // ...Topic, Id, Offset, MaxCount, ReplicaNodeID...
    LastFetchedEpoch int64 // latest epoch in the follower's cache (0 = none)
}

type FetchBatchResponse struct {
    Entries     []*LogEntry
    EpochStarts []EpochStart    // leader epoch boundaries covering Entries
    Diverging   *DivergingEpoch // set, with no Entries, when the follower must truncate first
}

After appending a batch, the follower Assigns each EpochStart it received (recordEpochStarts).

#The leader checks for divergence before serving a fetch

Before reading any records for a replica, the leader compares the follower's history with its own:

// broker/topic/epoch.go
func (tm *TopicManager) CheckFollowerDivergence(topicName string, l *log.LogManager, lastFetchedEpoch int64, fetchOffset uint64) *protocol.DivergingEpoch {
    tm.ensureLeaderEpoch(topicName, l)
    epoch, endOffset := l.Epochs.EndOffsetFor(lastFetchedEpoch, l.LEO())
    if epoch == lastFetchedEpoch && endOffset >= fetchOffset {
        return nil
    }
    return &protocol.DivergingEpoch{Epoch: epoch, EndOffset: endOffset}
}

The follower says "my latest epoch is e and my log ends at offset". If the leader knows epoch e and that epoch runs at least to offset in its own log, the follower's log is a prefix of the leader's and the fetch proceeds. Otherwise the follower holds records the leader doesn't. The FetchBatch handler then returns the DivergingEpoch instead of records, and skips recording the replica's LEO, since that position isn't valid in the leader's log:

// broker/rpc/consumer.go — FetchBatch
if req.ReplicaNodeID != "" {
    if div := s.topicManager.CheckFollowerDivergence(req.Topic, leaderLog, req.LastFetchedEpoch, req.Offset); div != nil {
        return &protocol.FetchBatchResponse{Diverging: div}, nil
    }
}

The divergence check runs on every replication fetch, not only after a leader change, because it costs one binary search over a list with one entry per leader change. A follower that's caught up still gets CodeReadOffset, so the check happens before the "nothing new" answer. That ordering matters for the scenario above: the old leader's LEO is ahead of the new leader's, and it must hear "you diverged", not "you're caught up".

#The follower truncates and fetches again

// broker/topic/epoch.go
func (tm *TopicManager) truncateDivergentTail(topicName string, l *log.LogManager, requestedEpoch int64, div *protocol.DivergingEpoch) error {
    leo := l.LEO()
    end := min(div.EndOffset, leo)
    if div.Epoch != requestedEpoch {
        _, ourEnd := l.Epochs.EndOffsetFor(div.Epoch, leo)
        end = min(end, ourEnd)
    }
    return l.TruncateTo(end)
}

If the leader knows the follower's epoch, the answer is exact: truncate to where that epoch ends on the leader. If it doesn't, it answered with an older epoch it does know. Then the follower's records from any newer epoch are suspect too, so it also cuts back to where that older epoch ends in its own log. LogManager.TruncateTo (low-level log page) removes the records, pulls LEO and the high watermark back, and drops epoch entries that start past the cut. The next fetch therefore asks about an older epoch. Each round either shortens the log or removes an epoch, so the exchange converges.

Here's a case that takes more than one round. The follower once led an epoch 3 that the current leader never heard of:

	Epoch history	LEO
Leader	(1, 0), (2, 8), (4, 12)	15
Follower	(1, 0), (3, 10)	15

1. The follower fetches with epoch 3, offset 15. The leader's largest epoch at or below 3 is 2, which ends at 12, so it answers Diverging{Epoch: 2, EndOffset: 12}. The follower's own epoch 2 doesn't exist: its largest at or below 2 is epoch 1, which ends at 10. It truncates to min(12, 10) = 10 and drops (3, 10).
2. The follower fetches with epoch 1, offset 10. Epoch 1 ends at 8 on the leader: Diverging{Epoch: 1, EndOffset: 8}. It truncates to 8.
3. The follower fetches with epoch 1, offset 8. Epoch 1 runs to 8 on the leader, so there's no divergence, and the leader returns offsets 8–14 with epoch starts (2, 8) and (4, 12).

The follower ends with exactly the leader's records and epoch history. This is the mechanism Kafka uses (KIP-101 and KIP-279), and KRaft uses the same DivergingEpoch shape in its fetch protocol.

#Step 8: Trace the complete replication cycle

#End-to-end replication flow

The same cycle, step by step:

1. Producer appends to leader
   Leader: LEO goes from 100 → 101

2. Follower's replicate loop wakes (every 1s)
   Follower → cc.FetchBatch(topic, offset=100, maxCount=5000)
   (cc is the cached connection for this leader — no new dial)

3. Leader's fetch handler:
   - Reads records starting at offset 100 using ReadUncommitted
   - Returns entries to follower
   - Calls RecordReplicaLEOFromFetch(topic, followerID, 100)

4. Leader's RecordReplicaLEOFromFetch:
   - coordinator.RecordReplicaFetch: follower's LEO=100, within threshold → ISR
   - SetHighWatermark(coordinator.TopicMinISRLeo(...)) → HW stays at 100
     (follower LEO 100 < leader LEO 101)
   - Propagates ISR status via Raft (ApplyIsrUpdateEventInternal)

5. Follower applies records via ApplyRecordBatch
   Follower: LEO goes from 100 → 101

6. Next fetch cycle (same cached connection):
   Follower → cc.FetchBatch(topic, offset=101)

7. Leader's RecordReplicaLEOFromFetch:
   - follower's LEO=101 → HW = min(101, 101) = 101

8. Consumer can now read offset 100
   (HW=101 means offsets 0..100 are committed)

#Step 9: Configure replication parameters

#Important timing and sizing constants

Constant	Value	Purpose
DefaultReplicationBatchSize	5,000	Max records per fetch in replication
DefaultISRLagThreshold (broker/topic/placement.go)	100	Max LEO lag to stay in ISR
replicationTickInterval	1 second	How often runReplicateLoop wakes
reconcileTickInterval	50 milliseconds	How often runReconcileLoop wakes (new — see Step 2)
AckAll timeout	5 seconds	Max wait for replicas when acks=all (see the previous page)
AckAll poll interval	10 ms	How often waitForAllFollowersToCatchUp checks replica LEO

#Summary

Component	Purpose
Two independent loops	runReconcileLoop (50ms, opens/closes local logs) and runReplicateLoop (1s, pulls data) run on separate goroutines so one never blocks the other.
Per-leader goroutines	Each leader gets its own goroutine within a replication tick. A slow leader does not block others.
replicationConnCache	One ConsumerClient per leader, kept alive across ticks instead of dialed and torn down every second. Invalidated only on reconnect-worthy errors.
Pending-slice pattern	Topics are removed from pending once caught up. No fixed round limit.
Batch writes	ApplyRecordBatch writes all fetched records in one AppendBatch call.
Leader epoch cache	Each log records (epoch, start offset) for every leader that wrote to it, persisted in leader-epoch-checkpoint. Followers copy the leader's boundaries from EpochStarts.
Divergence check	Every replication fetch carries LastFetchedEpoch; the leader answers Diverging when the follower holds records it doesn't, and the follower truncates until its log is a prefix of the leader's.
RecordReplicaLEOFromFetch	Leader records each follower's LEO on every fetch; ISR math now lives on cluster.TopicMetadata, reached through TopicCoordinator.
ISR tracking	Replica within 100 records of leader = in ISR. Changes propagated via Raft.
ISR expiry	A replica that stops fetching for ISRLagTime is demoted, but only while the leader is in contact with the Raft quorum, so a partitioned leader can't shrink its ISR and expose uncommitted records.
HW advancement	HW = min(LEO across ISR), computed via coordinator.TopicMinISRLeo. Consumers read offsets below HW.

With replication complete, the next page builds the producer and consumer client libraries — the interface users interact with to write and read data, including the same connection-reuse and reconnection philosophy you just saw applied to replication.