Raft Consensus and the Cluster Package — Distributed Metadata with HashiCorp Raft

The cluster needs a single source of truth for metadata — which topics exist, who leads each topic, and which replicas are in sync. In this section, you'll integrate HashiCorp Raft for consensus. The code is split across two packages that mirror a clean separation of concerns: broker/cluster/raft is generic Raft plumbing (node lifecycle, FSM, log store — nothing here knows what a "topic" is), and broker/cluster is where the actual replicated state lives (ClusterMetadataStore) behind a Cluster type that the rest of the broker talks to.

Step 1: Understand the split between RaftNode and Cluster

What each type does

raft.RaftNode (broker/cluster/raft/node.go) is the low-level Raft wrapper. It only knows about Raft concepts — servers, voters, terms, snapshots — never about topics or replicas:

1. Sets up Raft — configures the Raft node with the FSM, log store, stable store, and transport.
2. Joins and leaves — adds or removes servers from the Raft voter configuration.
3. Applies raw bytes — ApplyEvent(data []byte) proposes an already-encoded event to the Raft log; it has no idea what's inside.
4. Queries — "who is the Raft leader?", "what are the current server IDs?", "is Raft ready?"
5. Watches peer changes — an observer-based stream of voter add/remove events (Step 8).

cluster.Cluster (broker/cluster/cluster.go) is what everything else in the broker actually talks to. It owns a *raft.RaftNode and a *ClusterMetadataStore, and is where topic-shaped operations live: encoding a CreateTopicEvent and calling RaftNode.ApplyEvent, answering "who leads topic X?", and reconciling Raft's voter list against Serf membership.

Step 2: Define the MetadataStore interface

Interface for applying events

raft.RaftNode's FSM delegates to a MetadataStore — implemented by cluster.ClusterMetadataStore:

// broker/cluster/raft/metadata.go
package raft

type MetadataStore interface {
    Apply(ev *MetadataEvent) error
    Restore(data []byte) error
    Snapshot() ([]byte, error)
}

• Apply — process a single committed event (e.g., create a topic, update ISR).
• Snapshot — serialize the store's entire current state, for Raft to compact its log.
• Restore — rebuild state from a previously taken snapshot.

Snapshot is part of the interface itself rather than something the FSM does by directly marshaling the store — ClusterMetadataStore.Snapshot() returns already-encoded protobuf bytes (see Step 6), and the FSM just forwards them.

MetadataEvent is the envelope every event travels in — an event type plus its already-encoded payload — and is itself encoded as protobuf, not JSON:

type MetadataEventType uint16

const (
    MetadataEventTypeCreateTopic MetadataEventType = iota
    MetadataEventTypeDeleteTopic
    MetadataEventTypeLeaderChange
    MetadataEventTypeIsrUpdate
    MetadataEventTypeAddNode
    MetadataEventTypeRemoveNode
    MetadataEventTypeUpdateNode
)

type MetadataEvent struct {
    EventType MetadataEventType
    Data      []byte
}

func EncodeMetadataEvent(ev *MetadataEvent) ([]byte, error) {
    return proto.Marshal(&pb.MetadataEvent{EventType: uint32(ev.EventType), Data: ev.Data})
}

func DecodeMetadataEvent(data []byte) (*MetadataEvent, error) {
    var m pb.MetadataEvent
    if err := proto.Unmarshal(data, &m); err != nil {
        return nil, err
    }
    return &MetadataEvent{EventType: MetadataEventType(m.EventType), Data: m.Data}, nil
}

Each event type (CreateTopicEvent, DeleteTopicEvent, LeaderChangeEvent, IsrUpdateEvent, and node-related AddNodeEvent/RemoveNodeEvent/UpdateNodeEvent) has its own small Encode*/Decode* pair in this file, each just wrapping the matching generated pb.* type. As you'll see in Step 8, cluster membership itself is derived from Raft's own voter configuration plus Serf's gossiped tags, so the node-related event types are defined for completeness but nothing in this system actually applies them — worth knowing if you go looking for where AddNodeEvent gets applied and don't find a caller.

Step 3: Create the RaftNode struct

RaftNode structure

// broker/cluster/raft/node.go
package raft

const (
    SnapshotThreshold   = 10000
    SnapshotInterval    = 10
    RetainSnapshotCount = 10
)

type RaftNode struct {
    Logger     *zap.Logger
    raft       *raft.Raft
    raftConfig *raft.Config
    LocalAddr  raft.ServerAddress
    cfg        config.Config
}

RaftNode doesn't assert it implements discovery.Handler and doesn't hold a MemberLister — those concerns belong to Cluster, which is the type that actually gets wired into discovery.Membership. RaftNode stays pure Raft, with no notion of Serf or topics.

Step 4: Set up Raft

Raft initialization

NewRaftNode builds the FSM, sets up the underlying hashicorp/raft.Raft, and returns the wrapper — but does not bootstrap yet:

func NewRaftNode(cfg config.Config, metadataStore MetadataStore, logger *zap.Logger) (*RaftNode, error) {
    if logger == nil {
        logger = zap.NewNop()
    }
    fsm, err := NewFSM(cfg.RaftConfig.Dir, metadataStore)
    if err != nil {
        return nil, err
    }
    raftNode, raftConfig, localAddr, err := setupRaft(fsm, cfg.RaftConfig)
    if err != nil {
        return nil, err
    }

    c := &RaftNode{
        Logger:     logger,
        raft:       raftNode,
        raftConfig: raftConfig,
        LocalAddr:  localAddr,
        cfg:        cfg,
    }
    rpcAddr, err := cfg.RPCAddr()
    if err != nil {
        return nil, err
    }
    c.Logger.Info("coordinator started", zap.String("raft_addr", cfg.RaftConfig.Address), zap.String("rpc_addr", rpcAddr))
    return c, nil
}

Raft configuration and storage setup

setupRaft is a standalone function — it only needs the FSM and Raft config, and now returns the raft config and local transport address too (so NewRaftNode can stash them for later use, e.g. by Start):

func setupRaft(fsm raft.FSM, cfg config.RaftConfig) (*raft.Raft, *raft.Config, raft.ServerAddress, error) {
    raftBindAddr := cfg.Address
    if cfg.BindAddress != "" {
        raftBindAddr = cfg.BindAddress
    }
    raftAdvertiseAddr := cfg.Address
    raftConfig := raft.DefaultConfig()
    raftConfig.SnapshotThreshold = uint64(SnapshotThreshold)
    raftConfig.SnapshotInterval = time.Duration(SnapshotInterval) * time.Second
    raftConfig.LocalID = raft.ServerID(cfg.ID)
    raftConfig.LogLevel = cfg.LogLevel

    advertiseAddr, err := net.ResolveTCPAddr("tcp", raftAdvertiseAddr)
    if err != nil {
        return nil, nil, "", fmt.Errorf("failed to resolve Raft advertise address %s: %w", raftAdvertiseAddr, err)
    }
    transport, err := raft.NewTCPTransport(raftBindAddr, advertiseAddr, 3, 10*time.Second, os.Stderr)
    if err != nil {
        return nil, nil, "", fmt.Errorf("failed to make TCP transport bind %s advertise %s: %w", raftBindAddr, raftAdvertiseAddr, err)
    }
    snapshots, err := raft.NewFileSnapshotStore(cfg.Dir, RetainSnapshotCount, os.Stderr)
    if err != nil {
        return nil, nil, "", fmt.Errorf("failed to create snapshot store at %s: %w", cfg.Dir, err)
    }
    boltDB, err := raftboltdb.NewBoltStore(filepath.Join(cfg.Dir, "raft.db"))
    if err != nil {
        return nil, nil, "", fmt.Errorf("failed to create bolt store: %w", err)
    }
    logStore, err := NewLogStore(cfg.Dir)
    if err != nil {
        return nil, nil, "", fmt.Errorf("failed to create log store: %w", err)
    }
    ra, err := raft.NewRaft(raftConfig, fsm, logStore, boltDB, snapshots, transport)
    if err != nil {
        return nil, nil, "", ErrNewRaft(err)
    }
    return ra, raftConfig, transport.LocalAddr(), nil
}

Walk through the setup:

1. Bind vs. Advertise address. In Docker, the Raft node binds to 0.0.0.0:9093 (so connections from any interface are accepted) but advertises node1:9093 (so other nodes can reach it by hostname). cfg.BindAddress is the listen address; cfg.Address is what others use.

2. Raft config. Snapshots are taken every 10 seconds if at least 10,000 entries have been committed. LogLevel comes from the config (typically "ERROR" to reduce noise).

3. Transport. Raft uses its own TCP transport, separate from the application's api/transport. The advertiseAddr is resolved to a *net.TCPAddr and passed to NewTCPTransport. The 3 is the max connection pool size; 10*time.Second is the connection timeout.

4. Stores. Five components Raft needs:

Component	Implementation
FSM	raft.FSM — deserializes metadata events (protobuf), calls MetadataStore.Apply()
LogStore	raft.logStore — adapter over broker/log.Log (converts 1-based Raft indices to 0-based offsets)
StableStore	BoltDB (raft.db) — stores Raft's current term and voted-for
SnapshotStore	File-based — retains up to 10 snapshots
Transport	TCP — Raft's own transport for AppendEntries, RequestVote, etc.

5. Bootstrap is a separate step. setupRaft only builds the Raft instance — bootstrapping a brand-new cluster happens in RaftNode.Start(), called explicitly by Cluster.NewCluster right after construction:

func (c *RaftNode) Start() error {
    cfg := c.cfg.RaftConfig
    raftConfig := c.raftConfig
    if cfg.Boostatrap {
        configuration := raft.Configuration{
            Servers: []raft.Server{
                {ID: raftConfig.LocalID, Address: c.LocalAddr},
            },
        }
        if err := c.raft.BootstrapCluster(configuration).Error(); err != nil {
            return ErrBootstrapCluster(err)
        }
    }
    return nil
}

(The Boostatrap field name typo is preserved from the original config struct — it's broker/config.RaftConfig.Boostatrap.) Only one node should set --bootstrap; otherwise you get split-brain.

Step 5: Implement the FSM

FSM struct and Apply method

When Raft commits an entry, the FSM's Apply method is called. It decodes the protobuf metadata event and delegates to the MetadataStore:

// broker/cluster/raft/fsm.go
var _ raft.FSM = (*FSM)(nil)

type FSM struct {
    mu            sync.RWMutex
    metadataStore MetadataStore
    BaseDir       string
}

func NewFSM(baseDir string, metadataStore MetadataStore) (*FSM, error) {
    return &FSM{metadataStore: metadataStore, BaseDir: baseDir}, nil
}

func (c *FSM) Apply(l *raft.Log) interface{} {
    metadataEvent, err := DecodeMetadataEvent(l.Data)
    if err != nil {
        return err
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.metadataStore.Apply(metadataEvent)
}

This is the core of the Raft integration. Every node in the cluster receives the same committed entries in the same order and applies them to its MetadataStore. This guarantees all nodes have a consistent view of cluster metadata.

The FSM holds its own sync.RWMutex to protect the metadata store during Apply (write lock) and Snapshot (read lock). This ensures snapshots see a consistent state even while new entries are being applied.

FSM snapshotting and restoration

Snapshots compact the Raft log. Instead of replaying thousands of entries, a new node (or a node that fell behind) can load a snapshot and then replay only recent entries. The FSM doesn't marshal anything itself — it just asks the store for its own serialized bytes:

func (c *FSM) Snapshot() (raft.FSMSnapshot, error) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    data, err := c.metadataStore.Snapshot()
    if err != nil {
        return nil, err
    }
    return &metadataSnapshot{data: data}, nil
}

func (c *FSM) Restore(r io.ReadCloser) error {
    defer r.Close()
    data, err := io.ReadAll(r)
    if err != nil {
        return err
    }
    return c.metadataStore.Restore(data)
}

The snapshot struct is simple — it holds the serialized bytes and writes them to the Raft sink:

var _ raft.FSMSnapshot = (*metadataSnapshot)(nil)

type metadataSnapshot struct {
    data []byte
}

func (s *metadataSnapshot) Persist(sink raft.SnapshotSink) error {
    if _, err := sink.Write(s.data); err != nil {
        sink.Cancel()
        return err
    }
    return sink.Close()
}

func (s *metadataSnapshot) Release() {}

Step 6: Implement the LogStore adapter

LogStore overview

Raft needs a persistent log store. Instead of using another library, this adapts the project's own segment-based log. The key challenge is unchanged: Raft uses 1-based indices while the log uses 0-based offsets. What did change is the on-disk encoding of each Raft log entry — it's now a protobuf pb.RaftLog message instead of a JSON-marshaled raft.Log.

LogStore struct with embedded Log

The logStore embeds *log.Log (not wraps it in a field), so it inherits all of the log's methods directly:

// broker/cluster/raft/logstore.go
var _ raft.LogStore = (*logStore)(nil)

type logStore struct {
    *log.Log
}

func NewLogStore(dir string) (*logStore, error) {
    log, err := log.NewLog(filepath.Join(dir, "__cluster_metadata__.log"))
    if err != nil {
        return nil, err
    }
    return &logStore{Log: log}, nil
}

The log directory is filepath.Join(dir, "__cluster_metadata__.log") — a subdirectory under the Raft data directory, keeping Raft's own log entries separate from topic data.

Implementing LogStore interface methods

func (l *logStore) FirstIndex() (uint64, error) {
    if l.IsEmpty() {
        return 0, nil
    }
    return l.LowestOffset() + 1, nil
}

func (l *logStore) LastIndex() (uint64, error) {
    if l.IsEmpty() {
        return 0, nil
    }
    return l.HighestOffset() + 1, nil
}

The conversion is straightforward: Raft index = log offset + 1.

const segmentOffsetPrefix = 8

func (l *logStore) GetLog(index uint64, out *raft.Log) error {
    if index < 1 {
        return ErrRaftLogIndex(index)
    }
    offset := index - 1
    in, err := l.Read(offset)
    if err != nil {
        return err
    }
    if len(in) < segmentOffsetPrefix {
        return ErrLogRecordTooShort(offset)
    }
    payload := in[segmentOffsetPrefix:]
    var m pb.RaftLog
    if err := proto.Unmarshal(payload, &m); err != nil {
        return err
    }
    out.Index = index
    out.Term = m.Term
    out.Type = raft.LogType(m.Type)
    out.Data = m.Data
    out.Extensions = m.Extensions
    out.AppendedAt = time.Unix(0, m.AppendedAtUnixNano)
    return nil
}

GetLog has the same important detail as before: the segment Read returns the full record including the 8-byte offset prefix ([Offset 8 bytes][Value]). The segmentOffsetPrefix constant skips those 8 bytes to get the actual protobuf payload, which is then unmarshaled into a pb.RaftLog and copied field-by-field into the raft.Log Raft expects — including converting the stored Unix nanosecond timestamp back into a time.Time.

func (l *logStore) StoreLog(log *raft.Log) error {
    data, err := proto.Marshal(&pb.RaftLog{
        Index:              log.Index,
        Term:               log.Term,
        Type:               uint32(log.Type),
        Data:               log.Data,
        Extensions:         log.Extensions,
        AppendedAtUnixNano: log.AppendedAt.UnixNano(),
    })
    if err != nil {
        return err
    }
    _, err = l.Append(data)
    return err
}

func (l *logStore) StoreLogs(logs []*raft.Log) error {
    for _, log := range logs {
        if err := l.StoreLog(log); err != nil {
            return err
        }
    }
    return nil
}

func (l *logStore) DeleteRange(min, max uint64) error {
    if max < 1 {
        return nil
    }
    return l.Truncate(max - 1)
}

DeleteRange truncates at max - 1 (converting the 1-based Raft index to a 0-based offset). The Truncate method on the log removes all segments up to and including the given offset, which is what Raft needs when compacting old entries after a snapshot.

Why use the project's own log? It works and avoids adding another storage dependency. The segment-based log handles append, read by offset, and truncation — exactly what Raft needs.

Step 7: Build ClusterMetadataStore — the actual replicated state

TopicMetadata: the Raft-replicated view of one topic

This is the piece that moved out of TopicManager entirely. broker/cluster/cluster_metadata.go defines the data Raft actually keeps consistent across the cluster — one topic's leader, epoch, and replica set:

// broker/cluster/cluster_metadata.go
package cluster

var _ raft.MetadataStore = (*ClusterMetadataStore)(nil)

type ReplicaState struct {
    ReplicaNodeID string `json:"replica_id"`
    LEO           int64  `json:"leo"`
    IsISR         bool   `json:"is_isr"`
}

// TopicMetadata is the cluster-wide (Raft-replicated) view of one topic: its
// current leader, epoch, and replica set.
type TopicMetadata struct {
    mu                  sync.RWMutex
    Name                string
    LeaderNodeID        string
    LeaderEpoch         int64
    DesiredReplicaCount int // from CreateTopic; used to re-add replicas when nodes rejoin
    Replicas            map[string]*ReplicaState
}

Each TopicMetadata guards its own fields with its own mutex — reads like LeaderID() and mutations like SetLeader(), SetReplicaISR(), and RecordReplicaFetch() (all called with the store's coarser lock already released) are safe to call independently.

RecordReplicaFetch is worth calling out: it's invoked directly by whichever node currently leads the topic when it serves a Fetch from a replica — not through Raft — so it recomputes ISR membership locally and returns the new status for the caller to propagate via a Raft-applied IsrUpdateEvent if it changed:

func (t *TopicMetadata) RecordReplicaFetch(nodeID string, leo int64, lagThreshold uint64, leaderLEO uint64) (isr bool) {
    t.mu.Lock()
    defer t.mu.Unlock()
    rs := t.Replicas[nodeID]
    if rs == nil {
        rs = &ReplicaState{ReplicaNodeID: nodeID, LEO: leo}
        t.Replicas[nodeID] = rs
    } else {
        rs.LEO = leo
    }
    if leaderLEO > lagThreshold {
        isr = uint64(leo) >= leaderLEO-lagThreshold
    } else {
        isr = leo >= 0 // all replicas are in-sync for small topics
    }
    rs.IsISR = isr
    return isr
}

ClusterMetadataStore: Apply, Snapshot, Restore

type ClusterMetadataStore struct {
    mu     sync.RWMutex
    Topics map[string]*TopicMetadata
}

func (s *ClusterMetadataStore) Apply(ev *raft.MetadataEvent) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    switch ev.EventType {
    case raft.MetadataEventTypeCreateTopic:
        e, err := raft.DecodeCreateTopicEvent(ev.Data)
        if err != nil {
            return err
        }
        s.createTopicLocked(e.Topic, e.LeaderNodeID, e.LeaderEpoch, e.ReplicaNodeIds)
    case raft.MetadataEventTypeDeleteTopic:
        e, err := raft.DecodeDeleteTopicEvent(ev.Data)
        if err != nil {
            return err
        }
        delete(s.Topics, e.Topic)
    case raft.MetadataEventTypeLeaderChange:
        e, err := raft.DecodeLeaderChangeEvent(ev.Data)
        if err != nil {
            return err
        }
        if t := s.Topics[e.Topic]; t != nil {
            oldLeaderID := t.SetLeader(e.LeaderNodeID, e.LeaderEpoch)
            if oldLeaderID != e.LeaderNodeID && oldLeaderID != "" {
                t.AddReplicaIfAbsent(oldLeaderID, false)
            }
        }
    case raft.MetadataEventTypeIsrUpdate:
        e, err := raft.DecodeIsrUpdateEvent(ev.Data)
        if err != nil {
            return err
        }
        if t := s.Topics[e.Topic]; t != nil {
            t.SetReplicaISR(e.ReplicaNodeID, e.Isr)
        }
    default:
        return fmt.Errorf("unknown or unsupported event type for cluster metadata: %d", ev.EventType)
    }
    return nil
}

Notice the switch only handles four event types — CreateTopic, DeleteTopic, LeaderChange, IsrUpdate. The node-related event types defined in broker/cluster/raft/metadata.go (AddNodeEvent, RemoveNodeEvent, UpdateNodeEvent) fall through to the default case if they ever showed up here — but as covered in Step 8, nothing ever applies them anymore, so this switch never actually sees them in practice.

LeaderChange has a subtle but important detail: when a topic's leader changes, the old leader is added back as a (non-ISR) replica via AddReplicaIfAbsent — a demoted leader still holds a full copy of the log and should be tracked as a replica candidate for future re-election, not dropped from TopicMetadata entirely.

Snapshot/Restore serialize the whole Topics map as protobuf (pb.MetadataSnapshot, with nested pb.TopicState/pb.ReplicaState messages) instead of JSON-marshaling the struct directly:

func (s *ClusterMetadataStore) Snapshot() ([]byte, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    return proto.Marshal(snapshotToPB(s.Topics))
}

func (s *ClusterMetadataStore) Restore(data []byte) error {
    var m pb.MetadataSnapshot
    if err := proto.Unmarshal(data, &m); err != nil {
        return err
    }
    s.mu.Lock()
    defer s.mu.Unlock()
    s.Topics = pbToSnapshot(&m)
    return nil
}

snapshotToPB/pbToSnapshot are the conversion helpers between the in-memory map[string]*TopicMetadata and the generated pb.MetadataSnapshot — the same shape of conversion you'll see again in api/protocol/convert.go for client-facing messages.

Two more ClusterMetadataStore methods matter for the rest of the system: NodeIDWithLeastTopics(candidateNodeIDs []string) (deterministic tie-break by lexicographically smallest node ID, used for CreateTopic leader placement) and TopicMetadata.MinISRLeo(localLEO) (returns min(localLEO, all ISR replicas' LEO) — the value the high watermark gets set to).

Step 8: Build Cluster — the object the rest of the broker talks to

Cluster struct

cluster.Cluster wraps a *raft.RaftNode and a *ClusterMetadataStore together, plus a MemberLister for reconciling against Serf:

// broker/cluster/cluster.go
package cluster

// MemberLister returns information about cluster members currently alive (as seen
// by Serf). Used to reconcile Raft voters with Serf membership, and as the source
// of node addresses (Raft's own configuration only knows Raft addresses, not RPC
// addresses).
type MemberLister interface {
    AliveMembers() []string
    AliveNodeDetails() []discovery.NodeInfo
}

type Cluster struct {
    Logger        *zap.Logger
    node          *raft.RaftNode
    cfg           config.Config
    metadataStore *ClusterMetadataStore

    mu            sync.RWMutex
    memberLister  MemberLister
    onNodeRemoved func(nodeID string)
}

func NewCluster(cfg config.Config, logger *zap.Logger) (*Cluster, error) {
    if logger == nil {
        logger = zap.NewNop()
    }
    c := &Cluster{
        Logger:        logger,
        cfg:           cfg,
        metadataStore: NewClusterMetadataStore(),
    }
    node, err := raft.NewRaftNode(cfg, c.metadataStore, logger)
    if err != nil {
        return nil, err
    }
    c.node = node
    c.node.Start()
    go c.watchPeerChanges()
    return c, nil
}

NewCluster is where bootstrap actually happens (via node.Start()), and where the peer-change watcher goroutine gets launched.

Topic metadata queries — Cluster as the read path

Every topic-metadata query TopicManager needs is a thin method on Cluster that reaches into metadataStore:

func (c *Cluster) TopicInfo(topic string) (info protocol.TopicInfo, ok bool) {
    t := c.metadataStore.GetTopic(topic)
    if t == nil {
        return protocol.TopicInfo{}, false
    }
    leaderID, epoch, replicaSnaps := t.Snapshot()
    replicas := make([]protocol.ReplicaInfo, 0, len(replicaSnaps))
    for _, rs := range replicaSnaps {
        replicas = append(replicas, protocol.ReplicaInfo{NodeID: rs.ReplicaNodeID, IsISR: rs.IsISR, LEO: rs.LEO})
    }
    return protocol.TopicInfo{Name: topic, LeaderNodeID: leaderID, LeaderEpoch: epoch, Replicas: replicas}, true
}

func (c *Cluster) RecordReplicaFetch(topic, replicaNodeID string, leo int64, lagThreshold uint64, localLEO uint64) (isr bool, ok bool) {
    t := c.metadataStore.GetTopic(topic)
    if t == nil {
        return false, false
    }
    return t.RecordReplicaFetch(replicaNodeID, leo, lagThreshold, localLEO), true
}

TopicExists, TopicNames, TopicLeaderNodeID, TopicHasReplica, TopicMinISRLeo, and NodeIDWithLeastTopics follow the same shape — Cluster never holds this state itself, it only forwards to metadataStore.

Applying events

The Apply*Event methods are how mutations happen — encode the typed event, wrap it in a MetadataEvent, and hand raw bytes to RaftNode.ApplyEvent:

func (c *Cluster) ApplyCreateTopicEvent(topic string, replicaCount uint32, leaderNodeID string, replicaNodeIds []string) error {
    if !c.node.IsLeader() {
        c.Logger.Debug("not leader, skipping create topic event", zap.String("topic", topic))
        return nil
    }
    eventData, err := raft.EncodeCreateTopicEvent(raft.CreateTopicEvent{
        Topic:          topic,
        ReplicaCount:   replicaCount,
        LeaderNodeID:   leaderNodeID,
        LeaderEpoch:    1,
        ReplicaNodeIds: replicaNodeIds,
    })
    if err != nil {
        return err
    }
    data, err := raft.EncodeMetadataEvent(&raft.MetadataEvent{
        EventType: raft.MetadataEventTypeCreateTopic,
        Data:      eventData,
    })
    if err != nil {
        return err
    }
    c.Logger.Info("apply create topic event", zap.String("topic", topic), zap.String("leader_node_id", leaderNodeID))
    if err := c.node.ApplyEvent(data); err != nil {
        c.Logger.Error("raft apply create topic failed", zap.Error(err), zap.String("topic", topic))
        return err
    }
    return nil
}

ApplyDeleteTopicEventInternal and ApplyLeaderChangeEvent follow the identical pattern. ApplyIsrUpdateEventInternal adds special-cased logging: ISR updates happen every replication cycle and are expected to fail during leadership transitions, so "shutdown"/"leadership lost" errors are logged at Debug rather than Error:

func (c *Cluster) ApplyIsrUpdateEventInternal(topic, replicaNodeID string, isr bool) error {
    if !c.node.IsLeader() {
        return nil
    }
    eventData, err := raft.EncodeIsrUpdateEvent(raft.IsrUpdateEvent{Topic: topic, ReplicaNodeID: replicaNodeID, Isr: isr})
    if err != nil {
        return err
    }
    data, err := raft.EncodeMetadataEvent(&raft.MetadataEvent{EventType: raft.MetadataEventTypeIsrUpdate, Data: eventData})
    if err != nil {
        return err
    }
    if err := c.node.ApplyEvent(data); err != nil {
        msg := err.Error()
        if strings.Contains(msg, "shutdown") || strings.Contains(msg, "leadership lost") {
            c.Logger.Debug("raft apply ISR update failed (shutdown or leadership change)", zap.Error(err))
        } else {
            c.Logger.Error("raft apply ISR update failed", zap.Error(err))
        }
        return err
    }
    return nil
}

The flow: Cluster.ApplyCreateTopicEvent() → encodes as protobuf → RaftNode.ApplyEvent() → Raft replicates to majority → Raft calls FSM.Apply() on all nodes → ClusterMetadataStore.Apply() updates in-memory state on every node.

Join and Leave

// Join adds id as a Raft voter. Cluster membership itself is purely Raft's own
// configuration — no separate metadata event is applied for it.
func (c *Cluster) Join(id, raftAddr, rpcAddr string) error {
    return c.node.Join(id, raftAddr, rpcAddr)
}

// Leave removes id as a Raft voter. watchPeerChanges observes the resulting Raft
// PeerObservation and fires onNodeRemoved — that single path covers removals
// regardless of what triggered them.
func (c *Cluster) Leave(id string) error {
    return c.node.Leave(id)
}

Node identity and address information come from two sources that are already authoritative and already replicated by construction — Raft's own voter configuration (who's a member) and Serf's gossiped tags (what their addresses are) — so there's no separate node-metadata event to keep in sync on join or leave. That's why AliveNodeIDs, IsNodeAlive, and NodeRPCAddr (shown below) go through RaftServerIDs() and the MemberLister instead of a Raft-replicated node map — there's nothing else to consult.

func (c *Cluster) AliveNodeIDs() []string {
    raftIDs, err := c.node.RaftServerIDs()
    if err != nil {
        return nil
    }
    c.mu.RLock()
    ml := c.memberLister
    c.mu.RUnlock()
    if ml == nil {
        return raftIDs
    }
    alive := make(map[string]struct{})
    for _, name := range ml.AliveMembers() {
        alive[name] = struct{}{}
    }
    out := make([]string, 0, len(raftIDs))
    for _, id := range raftIDs {
        if _, ok := alive[id]; ok {
            out = append(out, id)
        }
    }
    return out
}

func (c *Cluster) NodeRPCAddr(nodeID string) (string, bool) {
    c.mu.RLock()
    ml := c.memberLister
    c.mu.RUnlock()
    if ml == nil {
        return "", false
    }
    for _, n := range ml.AliveNodeDetails() {
        if n.Name == nodeID && n.RpcAddr != "" {
            return n.RpcAddr, true
        }
    }
    return "", false
}

AliveNodeIDs intersects Raft's voter configuration (authoritative — what consensus actually agreed to) with Serf's alive set, so a voter Serf hasn't caught up to marking dead yet doesn't get treated as a live placement candidate. discovery.Membership (see the previous chapter) calls Cluster.SetMemberLister(m) after it's constructed, wiring the two packages together — Cluster never imports discovery for anything but the NodeInfo type used in the MemberLister interface.

watchPeerChanges — reacting to voter changes as they happen

Cluster subscribes to Raft's own configuration-change observer so it reacts the instant a voter is actually added or removed, rather than checking on a timer:

func (c *Cluster) watchPeerChanges() {
    events, stop := c.node.WatchPeerChanges()
    defer stop()
    for ev := range events {
        if ev.Removed {
            c.Logger.Info("raft peer removed", zap.String("node_id", ev.NodeID))
            c.mu.RLock()
            fn := c.onNodeRemoved
            c.mu.RUnlock()
            if fn != nil {
                fn(ev.NodeID)
            }
        } else {
            c.Logger.Info("raft peer added", zap.String("node_id", ev.NodeID))
        }
    }
}

RaftNode.WatchPeerChanges registers a hashicorp/raft observer filtered to raft.PeerObservation events and translates them into a simple channel:

// broker/cluster/raft/node.go
type PeerChangeEvent struct {
    NodeID  string
    Removed bool
}

func (c *RaftNode) WatchPeerChanges() (<-chan PeerChangeEvent, func()) {
    raw := make(chan raft.Observation, 16)
    observer := raft.NewObserver(raw, false, func(o *raft.Observation) bool {
        _, ok := o.Data.(raft.PeerObservation)
        return ok
    })
    c.raft.RegisterObserver(observer)

    out := make(chan PeerChangeEvent, 16)
    stop := make(chan struct{})
    go func() {
        defer close(out)
        for {
            select {
            case <-stop:
                return
            case obs, ok := <-raw:
                if !ok {
                    return
                }
                po, ok := obs.Data.(raft.PeerObservation)
                if !ok {
                    continue
                }
                select {
                case out <- PeerChangeEvent{NodeID: string(po.Peer.ID), Removed: po.Removed}:
                case <-stop:
                    return
                }
            }
        }
    }()

    return out, func() {
        c.raft.DeregisterObserver(observer)
        close(stop)
    }
}

The decision to add or remove a voter comes from Serf, in real time — discovery.Membership calls Cluster.Join/Cluster.Leave directly off Serf's own join/leave/failed events (see the previous chapter). watchPeerChanges is the reliable way to react to the result of a removal, however it was triggered. Cluster.SetOnNodeRemoved is where broker/cmd/server/helper.go wires this up to TopicManager.ReassignLeadersForDeadNode, so a dropped voter triggers leader failover for every topic it led — covered in the next chapter, topic management.

Step 9: Understand the complete integration

How it all connects

Serf event: member join
    │
    ▼
discovery.Membership.handleJoin
    │
    ▼
Cluster.Join(id, raftAddr, rpcAddr)
    │
    ▼
RaftNode.Join → raft.AddVoter(id, raftAddr)
    │
    ▼
watchPeerChanges observes the voter-added event, logs it
    (no separate metadata event — membership IS the Raft voter config)

For topic creation:

Client: CreateTopicRequest → RPC → TopicManager.CreateTopic()
    │
    ▼
TopicManager picks leader node and replicas (via Cluster.AliveNodeIDs, Cluster.NodeIDWithLeastTopics)
    │
    ▼
Cluster.ApplyCreateTopicEvent(name, replicaCount, leader, replicas)
    │
    ▼
Raft replicates to majority
    │
    ▼
FSM.Apply() on ALL nodes
    │
    ▼
ClusterMetadataStore.Apply(CreateTopicEvent)
    │
    ▼
Every node's TopicManager reconcile loop (broker/topic/replication.go)
notices the new topic in cluster metadata and opens a local log
if it's the leader or a replica

Opening the local log happens on a fast (50ms) poll inside TopicManager, deliberately decoupled from the Raft apply path — the next two chapters cover why.

Step 10: Bootstrap the cluster

Cluster bootstrap sequence

When starting a fresh cluster:

1. Node 1 starts with --bootstrap. RaftNode.Start() bootstraps a single-node Raft cluster and elects node 1 leader.
2. Node 2 starts with --start-join-addr node1:gossip_port. Serf gossips the join; discovery.Membership on node 1 calls Cluster.Join(node2, ...), which adds node 2 as a Raft voter.
3. Node 3 joins similarly.
4. Now all three nodes are Raft voters, and Cluster.AliveNodeIDs()/Cluster.NodeRPCAddr() on every node agree (Raft's voter list intersected with Serf's alive set; addresses from Serf tags).

Only one node should use --bootstrap. If multiple nodes bootstrap, you get split-brain.

Summary

Component	Purpose
RaftNode (broker/cluster/raft)	Pure Raft plumbing: setup, Join/Leave (add/remove voter only), ApplyEvent, leader/config queries, WatchPeerChanges observer stream. Knows nothing about topics.
FSM (broker/cluster/raft)	Apply() decodes a protobuf MetadataEvent and calls MetadataStore.Apply(). Snapshot()/Restore() forward to the store's own protobuf (de)serialization.
logStore (broker/cluster/raft)	Embeds *log.Log, stores entries in __cluster_metadata__.log as protobuf pb.RaftLog. Converts 1-based Raft indices to 0-based offsets; GetLog strips the 8-byte segment offset prefix.
ClusterMetadataStore (broker/cluster)	The actual replicated state: map[string]*TopicMetadata (leader, epoch, replicas). Implements raft.MetadataStore. Snapshots/restores as protobuf.
Cluster (broker/cluster)	What the rest of the broker talks to: topic-metadata queries, Apply*Event methods, Join/Leave (voter-only, no metadata event), AliveNodeIDs/NodeRPCAddr via Raft config + Serf MemberLister, event-driven watchPeerChanges → onNodeRemoved callback.

With Raft and cluster metadata in place, the next page builds TopicManager — the per-node application layer that consumes Cluster's replicated state through the TopicCoordinator interface and manages the produce/consume workflow.