Topic Management — Leaders, Replicas, and the TopicManager

The TopicManager is the per-node application layer that ties everything together. Cluster-wide topic metadata (leader, epoch, replica set, ISR) lives in broker/cluster.ClusterMetadataStore, covered on the previous page — TopicManager is a lean consumer of that state: it only tracks which logs are open on this node, and asks the cluster for everything else through the TopicCoordinator interface. Code lives in broker/topic/.

Step 1: Understand what TopicManager owns

TopicManager struct

// broker/topic/topic.go
package topic

type TopicManager struct {
    mu sync.RWMutex
    // Topics is local runtime state only: this node's open log handles, keyed by
    // topic name. Cluster-wide state (leader, epoch, replica set, ISR) lives behind
    // TopicCoordinator, not here.
    Topics               map[string]*log.LogManager
    BaseDir              string
    Logger               *zap.Logger
    CurrentNodeID        string // Local node ID from config; not persisted.
    coordinator          TopicCoordinator
    stopPeriodic         chan struct{}
    replicationCancel    context.CancelFunc // non-nil while the replication/reconcile loops are running
    replConns            *replicationConnCache
    replicationBatchSize uint32
    ISRLagThreshold      uint64 // max record lag for ISR membership
}

Topics is just map[string]*log.LogManager: a name and an open log, nothing else. TopicManager doesn't implement any Raft-facing interface, doesn't hold JSON-tagged fields, and doesn't do any snapshotting — all of that belongs to ClusterMetadataStore. This package's only job is to keep the right logs open on the right nodes and to serve produce requests against them.

NewTopicManager

func NewTopicManager(baseDir string, coord TopicCoordinator, logger *zap.Logger) (*TopicManager, error) {
    if logger == nil {
        logger = zap.NewNop()
    }
    tm := &TopicManager{
        Topics:          make(map[string]*log.LogManager),
        BaseDir:         baseDir,
        Logger:          logger,
        coordinator:     coord,
        stopPeriodic:    make(chan struct{}),
        replConns:       newReplicationConnCache(),
        ISRLagThreshold: DefaultISRLagThreshold,
    }
    go tm.periodicLog(defaultMetadataLogInterval)
    tm.replicationBatchSize = DefaultReplicationBatchSize
    return tm, nil
}

coord is *cluster.Cluster in production, or an in-memory fake in tests (tests.FakeTopicCoordinator) — NewTopicManager just takes it directly as a constructor argument. periodicLog runs in the background for the lifetime of the manager, logging local LEOs every 30 seconds for debugging.

Step 2: Define the TopicCoordinator interface

Interface for everything TopicManager needs from the cluster

broker/topic/topic_coordinator.go defines exactly what TopicManager needs — metadata queries, metadata mutation, and cluster/Raft state — as one interface, implemented by *cluster.Cluster in production:

// broker/topic/topic_coordinator.go
package topic

type TopicCoordinator interface {
    // — metadata queries (read-only) —
    TopicExists(topic string) bool
    TopicNames() []string
    TopicInfo(topic string) (info protocol.TopicInfo, ok bool)
    TopicLeaderNodeID(topic string) (leaderNodeID string, ok bool)
    TopicHasReplica(topic, nodeID string) bool
    TopicMinISRLeo(topic string, localLEO uint64) uint64
    NodeIDWithLeastTopics(candidateNodeIDs []string) (string, error)

    // — metadata mutation (Raft-applied; only take effect when called on the Raft leader) —
    ApplyCreateTopicEvent(topic string, replicaCount uint32, leaderNodeID string, replicaNodeIds []string) error
    ApplyDeleteTopicEventInternal(topic string) error
    ApplyIsrUpdateEventInternal(topic, replicaNodeID string, isr bool) error
    ApplyLeaderChangeEvent(topic, leaderNodeID string, leaderEpoch int64) error
    RecordReplicaFetch(topic, replicaNodeID string, leo int64, lagThreshold uint64, localLEO uint64) (isr bool, ok bool)

    // — cluster/raft state —
    IsLeader() bool
    GetRaftLeaderNodeID() (string, error)
    AliveNodeIDs() []string
    NodeRPCAddr(nodeID string) (string, bool)
    IsNodeAlive(nodeID string) bool
}

Defining this as one interface means TopicManager never touches cluster-internal storage directly — it's a lean consumer of cluster state rather than an owner of it. That, in turn, means tests can swap in tests.FakeTopicCoordinator (an in-memory implementation that still exercises the real cluster.ClusterMetadataStore.Apply logic) without spinning up Raft or Serf at all.

Step 3: Replica placement policy

PickReplicaNodeIds

Picking which nodes host a new topic's replicas is a small, pure function with no cluster state involved, so it lives in its own file rather than as a TopicManager method or behind TopicCoordinator:

// broker/topic/placement.go
package topic

const DefaultISRLagThreshold = uint64(100)

func PickReplicaNodeIds(leaderNodeID string, replicaCount int, candidateNodeIDs []string) ([]string, error) {
    var others []string
    for _, id := range candidateNodeIDs {
        if id != leaderNodeID {
            others = append(others, id)
        }
    }
    if len(others) < replicaCount {
        return nil, ErrNotEnoughNodesf(replicaCount, len(others))
    }
    return others[:replicaCount], nil
}

Least-loaded leader selection (with lexicographic tie-breaking) lives on cluster.ClusterMetadataStore.NodeIDWithLeastTopics instead, since it needs to inspect every topic's current leader — state that belongs to the cluster package. TopicManager.CreateTopic calls it through the TopicCoordinator interface (tm.coordinator.NodeIDWithLeastTopics(candidates)) rather than owning the logic itself.

Step 4: Create and delete topics

CreateTopic

// broker/topic/topic.go
func (tm *TopicManager) CreateTopic(ctx context.Context, req *protocol.CreateTopicRequest) (*protocol.CreateTopicResponse, error) {
    c := tm.coordinator
    if !c.IsLeader() {
        return nil, fmt.Errorf("create topic must be sent to Raft leader: %w", ErrCannotReachLeader)
    }
    if c.TopicExists(req.Topic) {
        return nil, ErrTopicExistsf(req.Topic)
    }
    candidates := c.AliveNodeIDs()
    leaderNodeID, err := c.NodeIDWithLeastTopics(candidates)
    if err != nil {
        return nil, err
    }
    replicaNodeIds, err := PickReplicaNodeIds(leaderNodeID, int(req.ReplicaCount), candidates)
    if err != nil {
        return nil, ErrCreateTopic(err)
    }
    tm.Logger.Info("create topic via Raft", zap.String("topic", req.Topic), zap.String("leader_node_id", leaderNodeID), zap.Strings("replica_node_ids", replicaNodeIds))
    if err := c.ApplyCreateTopicEvent(req.Topic, req.ReplicaCount, leaderNodeID, replicaNodeIds); err != nil {
        return nil, err
    }
    return &protocol.CreateTopicResponse{Topic: req.Topic, ReplicaNodeIds: replicaNodeIds}, nil
}

The flow is: pick a leader, pick replicas, propose a CreateTopicEvent through Raft, and every step reads through tm.coordinator. Crucially, this method never opens a local log — opening the leader's or a replica's log happens asynchronously, on every node, via a fast reconcile poll (Step 6).

DeleteTopic

func (tm *TopicManager) DeleteTopic(ctx context.Context, req *protocol.DeleteTopicRequest) (*protocol.DeleteTopicResponse, error) {
    c := tm.coordinator
    if !c.IsLeader() {
        return nil, fmt.Errorf("delete topic must be sent to Raft leader: %w", ErrCannotReachLeader)
    }
    if !c.TopicExists(req.Topic) {
        return nil, ErrTopicNotFoundf(req.Topic)
    }
    tm.Logger.Info("delete topic via Raft", zap.String("topic", req.Topic))
    if err := c.ApplyDeleteTopicEventInternal(req.Topic); err != nil {
        return nil, ErrApplyDeleteTopic(err)
    }
    return &protocol.DeleteTopicResponse{Topic: req.Topic}, nil
}

Same shape: check leadership and existence through the coordinator, propose the event, return. The actual removal of each node's local log directory happens in the reconcile loop once the topic disappears from cluster metadata (removeLocalTopic, Step 6) — this is what backs log.Log.Delete() from the storage-layer page.

Step 5: Look up logs and leaders

GetLog, GetLeader, IsLeader, RPC address lookups

func (tm *TopicManager) lookupTopic(name string) *log.LogManager {
    tm.mu.RLock()
    defer tm.mu.RUnlock()
    return tm.Topics[name]
}

// GetLog returns the locally-open log for topic. Returns ErrTopicNotFound both when
// topic isn't known locally at all and when its log hasn't finished opening yet
// (reconcileLocalTopics is mid-flight) — either way there's nothing usable to hand
// back yet, and callers (e.g. Produce/Fetch) already treat ErrTopicNotFound as
// retriable (see client.RetryTopicNotReady).
func (tm *TopicManager) GetLog(topic string) (*log.LogManager, error) {
    l := tm.lookupTopic(topic)
    if l == nil {
        return nil, ErrTopicNotFoundf(topic)
    }
    return l, nil
}

func (tm *TopicManager) IsLeader(topic string) (bool, error) {
    leaderID, ok := tm.coordinator.TopicLeaderNodeID(topic)
    if !ok {
        return false, ErrTopicNotFoundf(topic)
    }
    return leaderID == tm.currentNodeID(), nil
}

func (tm *TopicManager) GetTopicLeaderRPCAddr(topic string) (string, error) {
    leaderID, ok := tm.coordinator.TopicLeaderNodeID(topic)
    if !ok {
        return "", ErrTopicNotFoundf(topic)
    }
    addr, ok := tm.coordinator.NodeRPCAddr(leaderID)
    if !ok {
        return "", ErrTopicNotFoundf(topic)
    }
    return addr, nil
}

GetTopicLeaderRPCAddr is a two-step lookup: ask the cluster who leads the topic, then separately ask Serf for that node's RPC address (Cluster.NodeRPCAddr, backed by MemberLister.AliveNodeDetails() from the previous page). GetRaftLeaderRPCAddr and ListTopics follow the same pattern, delegating to tm.coordinator.GetRaftLeaderNodeID()/tm.coordinator.TopicNames()/tm.coordinator.TopicInfo() respectively — there's no local iteration over tm.Topics for anything but the log handles themselves.

Step 6: Reconcile local logs against cluster metadata

Why a reconcile loop

Cluster metadata (who leads a topic, who its replicas are) is Raft-replicated and consistent the instant it commits — but a node still needs to physically open a log file on disk for any topic it now leads or replicates. Every node runs a fast reconcile loop (50ms — see the next page on replication) that diffs cluster metadata against its own open logs and opens or closes them to match:

// broker/topic/topic.go
func (tm *TopicManager) reconcileLocalTopics() {
    names := tm.coordinator.TopicNames()
    present := make(map[string]struct{}, len(names))
    for _, name := range names {
        present[name] = struct{}{}
        tm.reconcileLocalTopic(name)
    }

    tm.mu.RLock()
    localNames := make([]string, 0, len(tm.Topics))
    for name := range tm.Topics {
        localNames = append(localNames, name)
    }
    tm.mu.RUnlock()

    for _, name := range localNames {
        if _, ok := present[name]; !ok {
            tm.removeLocalTopic(name)
        }
    }
}

// reconcileLocalTopic opens the local log for topicName if this node is its leader or a
// replica and no log is open yet. No-op if topicName isn't leader/replica-local here.
func (tm *TopicManager) reconcileLocalTopic(topicName string) {
    // Cheap check first: once a topic's log is open, steady state (the overwhelming
    // majority of calls — this runs 20x/second per topic) never needs to touch cluster
    // metadata at all.
    if tm.lookupTopic(topicName) != nil {
        return
    }
    info, ok := tm.coordinator.TopicInfo(topicName)
    if !ok {
        return
    }
    currentNodeID := tm.currentNodeID()
    isLocal := info.LeaderNodeID == currentNodeID
    if !isLocal {
        for _, r := range info.Replicas {
            if r.NodeID == currentNodeID {
                isLocal = true
                break
            }
        }
    }
    if !isLocal {
        return
    }
    logManager, err := log.NewLogManager(filepath.Join(tm.BaseDir, topicName))
    if err != nil {
        tm.Logger.Warn("open log failed", zap.String("topic", topicName), zap.Error(err))
        return
    }
    tm.publishLocalLog(topicName, logManager)
}

// removeLocalTopic closes the topic log, removes it from local bookkeeping, and deletes the topic dir.
func (tm *TopicManager) removeLocalTopic(topicName string) {
    tm.mu.Lock()
    l, ok := tm.Topics[topicName]
    if ok {
        delete(tm.Topics, topicName)
    }
    tm.mu.Unlock()
    if l != nil {
        l.Close()
        l.Delete()
    }
    _ = os.RemoveAll(filepath.Join(tm.BaseDir, topicName))
    tm.Logger.Info("topic removed", zap.String("topic", topicName))
}

This single loop covers topic creation, deletion, and leader/replica role changes uniformly: a leader-change event just changes what TopicInfo reports, and the next reconcile tick opens or closes whatever's needed to match. A node rejoining and being reassigned as a replica for an under-replicated topic is handled the same way — it's just a consequence of ClusterMetadataStore's state changing, picked up on the next tick like everything else.

One consequence worth knowing about: there's a real, bounded window — bounded by the 50ms reconcile tick, not by anything unbounded — between CreateTopic returning and a given node's log actually being open. Callers producing or consuming right after creating a topic should expect a transient "topic not found" and retry (this is exactly what client.RetryTopicNotReady, covered in the producer/consumer chapter, exists for).

RestoreFromMetadata

The one place a Raft snapshot restore needs a TopicManager-side action is opening local logs for whatever cluster metadata a Restore() populated:

func (tm *TopicManager) RestoreFromMetadata() error {
    currentNodeID := tm.currentNodeID()
    topicNames := tm.coordinator.TopicNames()
    for _, topicName := range topicNames {
        leaderID, ok := tm.coordinator.TopicLeaderNodeID(topicName)
        if !ok || leaderID == "" {
            continue
        }
        if leaderID == currentNodeID {
            if err := tm.restoreLeaderTopic(topicName); err != nil {
                tm.Logger.Warn("restore leader topic failed", zap.String("topic", topicName), zap.Error(err))
                continue
            }
            // Initialize HW from local state so consumers can read immediately if no
            // replicas exist (otherwise HW stays 0 until replicas report in).
            if l := tm.lookupTopic(topicName); l != nil {
                l.SetHighWatermark(tm.coordinator.TopicMinISRLeo(topicName, l.LEO()))
            }
        } else if tm.coordinator.TopicHasReplica(topicName, currentNodeID) {
            if err := tm.restoreReplicaTopic(topicName, leaderID); err != nil {
                tm.Logger.Warn("restore replica topic failed", zap.String("topic", topicName), zap.Error(err))
                continue
            }
        }
    }
    return nil
}

In practice this and reconcileLocalTopics overlap in purpose (both open logs for topics this node should host) — RestoreFromMetadata just runs once, synchronously, right after a snapshot restore, so a restarted or newly-joined node doesn't have to wait a full reconcile tick to become useful.

Step 7: Handle produce requests

HandleProduce and HandleProduceBatch

Produce handling moved into its own file, broker/topic/producer.go, and now takes the already-resolved *log.LogManager directly instead of a *Topic wrapper:

// broker/topic/producer.go
package topic

// HandleProduce appends to the topic log (leader only). For ACK_ALL, waits for replicas to catch up.
func (tm *TopicManager) HandleProduce(ctx context.Context, topicName string, l *log.LogManager, logEntry *protocol.LogEntry, acks protocol.AckMode) (uint64, error) {
    offset, err := l.Append(logEntry.Value)
    if err != nil {
        return 0, err
    }
    tm.advanceHighWatermark(topicName, l)
    switch acks {
    case protocol.AckLeader:
        return offset, nil
    case protocol.AckAll:
        if err := tm.waitForAllFollowersToCatchUp(ctx, topicName, offset); err != nil {
            return 0, ErrWaitFollowersCatchUp(err)
        }
        return offset, nil
    default:
        return 0, ErrInvalidAckModef(int32(acks))
    }
}

HandleProduceBatch is the same shape for AppendBatch, returning both the base and last offsets written. The RPC handler (broker/rpc/producer.go, covered in the transport/RPC page) is what resolves the topic name to a *log.LogManager via TopicManager.GetLog before calling either of these.

advanceHighWatermark

Right after appending, the topic's high watermark is recomputed. This runs on every produce, not only when a replica's ISR status changes — a topic with zero replicas still needs its HW to track its own LEO, since nothing else would advance it:

func (tm *TopicManager) advanceHighWatermark(topicName string, l *log.LogManager) {
    l.SetHighWatermark(tm.coordinator.TopicMinISRLeo(topicName, l.LEO()))
}

TopicMinISRLeo — min(localLEO, all in-sync replicas' LEO) — now lives on cluster.TopicMetadata (see the previous page); TopicManager just calls through the coordinator and applies the result locally.

waitForAllFollowersToCatchUp

func (tm *TopicManager) waitForAllFollowersToCatchUp(ctx context.Context, topicName string, offset uint64) error {
    timeout := time.After(5 * time.Second)
    ticker := time.NewTicker(10 * time.Millisecond)
    defer ticker.Stop()

    requiredLEO := offset + 1

    for {
        var replicas []protocol.ReplicaInfo
        if info, ok := tm.coordinator.TopicInfo(topicName); ok {
            replicas = info.Replicas
        }

        useISR := false
        for _, r := range replicas {
            if r.IsISR {
                useISR = true
                break
            }
        }

        allCaughtUp := true
        candidates := 0
        for _, replica := range replicas {
            if useISR && !replica.IsISR {
                continue
            }
            candidates++
            if uint64(replica.LEO) < requiredLEO {
                allCaughtUp = false
                break
            }
        }

        if candidates == 0 || allCaughtUp {
            return nil
        }

        select {
        case <-ticker.C:
            continue
        case <-ctx.Done():
            return ctx.Err()
        case <-timeout:
            tm.Logger.Warn("followers catch-up timeout", zap.String("topic", topicName), zap.Uint64("required_offset", offset))
            return ErrTimeoutCatchUp
        }
    }
}

The algorithm polls every 10ms, only waits on ISR replicas if any exist, and gives up after a 5-second timeout. The replica list comes from tm.coordinator.TopicInfo(topicName) — a point-in-time snapshot from cluster.TopicMetadata.Snapshot() — on every loop iteration.

Step 8: Handle node failure

ReassignLeadersForDeadNode

broker/cmd/server/helper.go registers this method as Cluster's on-node-removed callback (Cluster.SetOnNodeRemoved, from the previous page), so it fires whenever the Raft peer-change watcher observes a voter drop out:

// broker/topic/topic.go
func (tm *TopicManager) ReassignLeadersForDeadNode(nodeID string) {
    if !tm.coordinator.IsLeader() {
        return
    }
    type leaderChange struct {
        topic     string
        newLeader string
        epoch     int64
    }
    var changes []leaderChange
    for _, topicName := range tm.coordinator.TopicNames() {
        info, ok := tm.coordinator.TopicInfo(topicName)
        if !ok || info.LeaderNodeID != nodeID {
            continue
        }
        var newLeader string
        for _, rs := range info.Replicas {
            if rs.NodeID == nodeID || !rs.IsISR {
                continue
            }
            if !tm.coordinator.IsNodeAlive(rs.NodeID) {
                continue
            }
            newLeader = rs.NodeID
            break
        }
        if newLeader == "" {
            tm.Logger.Warn("no ISR replica for leadership", zap.String("topic", topicName), zap.String("old_leader_node_id", nodeID))
            continue
        }
        changes = append(changes, leaderChange{topic: topicName, newLeader: newLeader, epoch: info.LeaderEpoch + 1})
    }
    if len(changes) == 0 {
        return
    }
    // Applied asynchronously so the caller (e.g. the reconciliation loop) isn't
    // blocked for the duration of the resulting Raft round-trips.
    go func() {
        for _, ch := range changes {
            if err := tm.coordinator.ApplyLeaderChangeEvent(ch.topic, ch.newLeader, ch.epoch); err != nil {
                tm.Logger.Warn("leader change apply failed", zap.String("topic", ch.topic), zap.Error(err))
            }
        }
    }()
}

Only ISR replicas are eligible for the new leadership, the new leader's epoch is oldEpoch + 1, and only the current Raft leader actually proposes anything — every other node's callback fires too but no-ops, since only the Raft leader can successfully commit the resulting event anyway. The go func() exists purely so the caller (the peer-change watcher goroutine) isn't blocked for the duration of the resulting Raft round-trips.

Step 9: Understand the end-to-end integration

Complete workflow from request to open log

Client: CreateTopicRequest
    │
    ▼
RPC handler → TopicManager.CreateTopic()
    │
    ├── coordinator.AliveNodeIDs() / NodeIDWithLeastTopics() → picks leader
    ├── PickReplicaNodeIds() → picks replicas
    │
    ▼
Cluster.ApplyCreateTopicEvent(topic, replicaCount, leader, replicas)
    │
    ▼
Raft replicates to majority; ClusterMetadataStore.Apply() updates state on ALL nodes
    │
    ▼
Each node's reconcile loop (next tick, ≤50ms later)
    │
    └── reconcileLocalTopic() opens a local log if this node is leader or replica

For produce:

Client: ProduceRequest (topic, value, acks=AckAll)
    │
    ▼
RPC handler → TopicManager.GetLog(topic) → TopicManager.HandleProduce(...)
    │
    ├── l.Append(value) → writes to disk
    ├── advanceHighWatermark(topic, l)
    │
    └── waitForAllFollowersToCatchUp(topic, offset)
            │
            └── polls coordinator.TopicInfo(topic) every 10ms until:
                - all ISR replicas have LEO > offset
                - or 5s timeout

For node failure:

Node 2 crashes
    │
    ▼
Serf detects failure → discovery.Membership.handleLeave → Cluster.Leave(node2)
    │
    ▼
RaftNode.Leave → raft.RemoveServer(node2)
    │
    ▼
watchPeerChanges observes the voter removal → onNodeRemoved(node2)
    │
    ▼
TopicManager.ReassignLeadersForDeadNode(node2)
    │
    └── go func() { Cluster.ApplyLeaderChangeEvent(topic, newLeader, epoch+1) }

Summary

Component	Purpose
TopicManager (broker/topic/topic.go)	Local-only state: map[string]*log.LogManager of open logs on this node.
TopicCoordinator (topic_coordinator.go)	Interface for every cluster-metadata query/mutation and cluster-state query TopicManager needs. Implemented by *cluster.Cluster; faked in tests.
PickReplicaNodeIds (placement.go)	Pure replica-placement policy — no cluster state involved.
CreateTopic / DeleteTopic	Pick leader/replicas (or validate existence), propose the event through the coordinator. Never open/close a log directly.
reconcileLocalTopics	Runs on a fast poll (see the next page), diffing cluster metadata against locally-open logs — opens logs for newly-assigned leader/replica roles, closes and deletes logs for removed topics.
HandleProduce / HandleProduceBatch (producer.go)	Append to the log, advance HW, and for AckAll block on waitForAllFollowersToCatchUp — reading replica state from coordinator.TopicInfo.
ReassignLeadersForDeadNode	Wired as Cluster's on-node-removed callback. Only ISR replicas are eligible for the new leadership; only the Raft leader's proposal actually takes effect.

With topic management in place, the next page covers replication — how the reconcile loop and the replicate loop run side by side, and how follower replicas pull data from the leader.