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

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:

Leader (Node 1)                     Follower (Node 2)
┌──────────────┐                    ┌──────────────┐
│ Log:         │                    │ Log:         │
│ [0][1][2][3] │  ← FetchBatch ──   │ [0][1]       │
│   LEO=4      │  ── records ──→    │   LEO=2      │
│   HW=2       │                    │              │
└──────────────┘                    └──────────────┘
  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(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.NewConsumerClient(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.

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. Uses batch writes (ApplyRecordBatch) instead of per-record appends.
  3. 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(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()
            }

            leo, ok := tm.GetLEO(topicName)
            if !ok {
                continue
            }

            resp, err := cc.FetchBatch(ctx, &protocol.FetchBatchRequest{
                Topic:    topicName,
                Id:       consumerID,
                Offset:   leo,
                MaxCount: batchSize,
            })
            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 — keep topic for next round.
                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
                }
            }

            // 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. Topics that hit transient errors also stay in pending for a retry.

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

Handling fetch errors gracefully

Three categories of errors during fetch:

Error type 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.
Transient error Keep topic in pending for another round.

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.

Step 7: Trace the complete replication cycle

End-to-end replication flow

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 8: 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.
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.
HW advancement HW = min(LEO across ISR), computed via coordinator.TopicMinISRLeo. Consumers read only up to 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.