Running a Distributed Log Cluster End to End

Running a Distributed Log Cluster End to End

In this section you'll bring everything together and run the distributed log end to end: build the binaries, start a 3-node cluster (with Docker or as plain local processes), create topics, produce and consume messages, and verify replication and fault tolerance against the real, running system.

Step 1: Build the binaries

Compiling everything with make build

make build

The Makefile builds four binaries, each from its own module directory:

build: build-server build-producer build-consumer build-topic

build-server:
    @mkdir -p $(BIN)
    $(GO_CMD) build -o $(BIN)/server ./broker/cmd/server

build-producer:
    @mkdir -p $(BIN)
    $(GO_CMD) build -o $(BIN)/producer ./producer/cmd

build-consumer:
    @mkdir -p $(BIN)
    $(GO_CMD) build -o $(BIN)/consumer ./consumer/cmd

build-topic:
    @mkdir -p $(BIN)
    $(GO_CMD) build -o $(BIN)/topic ./broker/cmd/topic
Binary Source package Purpose
bin/server ./broker/cmd/server The broker itself: log storage, Raft, Serf, RPC server
bin/producer ./producer/cmd CLI producer — sends messages read from stdin
bin/consumer ./consumer/cmd CLI consumer — streams messages to stdout
bin/topic ./broker/cmd/topic Topic administration: create, delete, list

The producer and consumer live in their own top-level modules (producer/, consumer/) so they can be imported and versioned independently of the broker; the server and the topic admin CLI live under broker/cmd/. Each also has its own build target if you only need one:

make build-server
make build-producer
make build-consumer
make build-topic

Step 2: Configure server startup flags

Command-line options for the server

broker/cmd/server/main.go defines the server's flags with cobra, bound through viper so every flag can also be set as an environment variable:

rootCmd.Flags().StringVar(&bindAddr, "bind-addr", "127.0.0.1:9092", "Serf listen address (use 0.0.0.0 in Docker)")
rootCmd.Flags().StringVar(&advertiseAddr, "advertise-addr", "", "Address other nodes use to reach this node (e.g. node1). When set, bind 0.0.0.0 for Serf/Raft/RPC")
rootCmd.Flags().IntVar(&rpcPort, "rpc-port", 9094, "RPC listen port")
rootCmd.Flags().StringVar(&dataDir, "data-dir", "/tmp/mlog", "data directory")
rootCmd.Flags().StringVar(&nodeID, "node-id", "node-1", "node ID")
rootCmd.Flags().StringSliceVar(&peers, "peer", nil, "peer nodes (nodeID=addr) for discovery join, repeatable")
rootCmd.Flags().StringVar(&raftAddr, "raft-addr", "127.0.0.1:9093", "Raft transport address")
rootCmd.Flags().BoolVar(&bootstrap, "bootstrap", false, "Bootstrap the Raft cluster")
rootCmd.Flags().Uint32Var(&replicationBatchSize, "replication-batch-size", 5000, "Max records per topic per replication request")
Flag Default Controls
--bind-addr 127.0.0.1:9092 Serf (gossip) listen address
--advertise-addr (empty) Hostname other nodes use to reach this node; when set, the node binds 0.0.0.0 for Serf/Raft/RPC instead of the literal --bind-addr/--raft-addr host
--rpc-port 9094 RPC listen port (produce, fetch, admin, replication)
--data-dir /tmp/mlog Root directory for log segments and Raft state
--node-id node-1 Unique node identifier
--peer (none) nodeID=host:port entries for Serf join, repeatable
--raft-addr 127.0.0.1:9093 Raft transport address
--bootstrap false Bootstrap a brand-new Raft cluster — set on exactly one node
--replication-batch-size 5000 Max records per topic per replication fetch

Every node needs three distinct ports: Serf (gossip membership), Raft (consensus transport), and RPC (client traffic and replication fetches).

How --advertise-addr rewrites the Raft address

buildConfig in main.go does one piece of real work beyond copying flags into a config.Config: when --advertise-addr is set, it splits the port back out of --raft-addr and rebuilds the address peers should dial, while keeping the literal bind address for the local listener:

if advertiseAddr != "" {
    _, raftPort, err := net.SplitHostPort(raftAddr)
    if err != nil {
        return config.Config{}, fmt.Errorf("invalid raft-addr %q: %w", raftAddr, err)
    }
    raftConfig.Address = net.JoinHostPort(advertiseAddr, raftPort)
    raftConfig.BindAddress = raftAddr
}

That's exactly the pattern the Docker Compose file below relies on: a container binds Raft to 0.0.0.0:9093 (BindAddress) but tells the rest of the cluster to reach it at node1:9093 (Address) — a hostname only resolvable on the Docker network. RPCListenAddr() (broker/config) follows the same idea for the RPC port: if --advertise-addr is set it listens on 0.0.0.0:<rpc-port>; otherwise it binds directly using the configured host, which is what the plain-process local cluster relies on with 127.0.0.1.

All flags can also be set via environment variables with the MLOG_ prefix — MLOG_NODE_ID, MLOG_DATA_DIR, and MLOG_RAFT_ADDR are bound directly to their flags and work as plain shell environment variables.

Step 3: Understand the startup wiring order

NewCommandHelper in broker/cmd/server/helper.go

CommandHelper wires the broker's pieces together in a fixed order:

func NewCommandHelper(config config.Config) (*CommandHelper, error) {
    cmdHelper := &CommandHelper{Config: config, shutdowns: make(chan struct{})}
    if err := cmdHelper.setupCoordinator(); err != nil {
        return nil, err
    }
    if err := cmdHelper.setupTopicManager(); err != nil {
        return nil, err
    }
    if err := cmdHelper.setupRpcServer(); err != nil {
        return nil, err
    }
    if err := cmdHelper.setupMembership(); err != nil {
        return nil, err
    }
    return cmdHelper, nil
}

1. setupCoordinator — cluster.NewCluster(...) constructs the Raft node. Raft's transport is started and, on a --bootstrap node, the single-member configuration is bootstrapped inside this constructor call — there's no separate step later. Right after, topic.NewTopicManager(dataDir, coord, logger) is created, passing the coordinator in, because the topic manager needs to query cluster metadata (topic leaders, ISR) from the moment it exists. coord.SetOnNodeRemoved(topicMgr.ReassignLeadersForDeadNode) wires a callback so that when Raft/Serf agree a node is gone, topic leadership is reassigned. If this node is bootstrapping, setupCoordinator then blocks up to 30 seconds (WaitforRaftReady) for a Raft leader to actually be elected before moving on.

2. setupTopicManager — if Raft isn't ready yet (a non-bootstrap node still catching up), waits up to RaftReadyTimeout (15 seconds), logging a warning rather than failing outright if it times out. RestoreFromMetadata() replays this node's share of the current Raft-committed topic metadata, opening local logs for any topic it hosts. StartReplicationThread() then launches the reconcile and replicate loops described in the replication chapter.

3. setupRpcServer — computes the RPC listen address, creates the ConsumerManager (which recovers committed consumer offsets from disk), and constructs rpc.NewRpcServer(listenAddr, topicMgr, consumerMgr). The server object exists at this point but is not yet accepting connections — that happens later, in Start().

4. setupMembership — discovery.New(coord, config) constructs the Serf agent and, synchronously, joins the cluster through the configured --peer addresses before returning. This is the moment this node's presence becomes visible to the rest of the cluster over gossip. coord.SetMemberLister(membership) gives the coordinator a way to consult current Serf membership.

func (cmdHelper *CommandHelper) Start() error {
    if err := cmdHelper.coord.Start(); err != nil {
        return fmt.Errorf("start coordinator: %w", err)
    }
    if err := cmdHelper.rpcServer.Start(); err != nil {
        return err
    }
    return nil
}

cmdHelper.Start() runs after NewCommandHelper returns. coord.Start() is a no-op by this point — Raft was already started inside NewCluster back in step 1, and the method is kept only so startup/shutdown stay symmetric. rpcServer.Start() is what actually opens the listening socket.

Put together, the ordering guarantees that matter are: the coordinator (and Raft) exist before the topic manager is built, because the topic manager takes the coordinator as a constructor argument; topic metadata is restored and replication is running before the RPC server is even constructed, so once RPC traffic is accepted the local topic state is already consistent; and the RPC server object (with its handlers) exists before Serf join happens, since rpc.NewRpcServer doesn't depend on Serf at all. The one genuine gap is that Serf join (setupMembership, still inside NewCommandHelper) completes before rpcServer.Start() actually opens the RPC listener (Start(), called by main.go afterward) — so there's a brief window where a peer that just learned about this node over gossip could get connection-refused on its RPC port. That's a benign race: every client and the replication engine already treat connection failures as reconnect-worthy (protocol.ShouldReconnect, covered in the protocol chapter) and simply retry.

Shutdown reverses the order: Shutdown() leaves Serf membership, stops the RPC server, then shuts down the coordinator (Raft).

Step 4: Deploy with Docker Compose

Dockerfile

# infra/Dockerfile

# Build stage
FROM golang:1.25-alpine AS builder
WORKDIR /app

COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 go build -o /server ./broker/cmd/server

# Run stage
FROM alpine:3.19
RUN apk --no-cache add ca-certificates netcat-openbsd
WORKDIR /app
COPY --from=builder /server .

# Default: run server (args overridden by docker-compose)
ENTRYPOINT ["/app/server"]
CMD ["--bind-addr", "0.0.0.0:9092", "--raft-addr", "0.0.0.0:9093", "--rpc-port", "9094", "--data-dir", "/data", "--node-id", "node-1", "--bootstrap", "false"]

netcat-openbsd is installed purely for the healthcheck (nc -z localhost 9094). The build stage compiles just the server binary from ./broker/cmd/server — the producer, consumer, and topic CLIs aren't needed inside the cluster containers, since they're meant to be run from the host against the exposed RPC ports.

docker-compose.yml

# infra/docker-compose.yml

services:
  node1:
    build:
      context: ..
      dockerfile: infra/Dockerfile
    container_name: mlog-node1
    hostname: node1
    ports:
      - "9092:9092"   # Serf
      - "9093:9093"   # Raft
      - "9094:9094"   # RPC
    volumes:
      - node1-data:/data
    command: [
      "server",
      "--bind-addr", "0.0.0.0:9092",
      "--raft-addr", "0.0.0.0:9093",
      "--rpc-port", "9094",
      "--advertise-addr", "node1",
      "--data-dir", "/data",
      "--node-id", "node-1",
      "--bootstrap", "true",
      "--peer", "node-2=node2:9095",
      "--peer", "node-3=node3:9098"
    ]
    networks:
      - mlog-net
    healthcheck:
      test: ["CMD", "nc", "-z", "localhost", "9094"]
      interval: 3s
      timeout: 3s
      retries: 10
      start_period: 20s

  node2:
    build:
      context: ..
      dockerfile: infra/Dockerfile
    container_name: mlog-node2
    hostname: node2
    ports:
      - "9095:9095"   # Serf
      - "9096:9096"   # Raft
      - "9097:9097"   # RPC
    volumes:
      - node2-data:/data
    command: [
      "server",
      "--bind-addr", "0.0.0.0:9095",
      "--raft-addr", "0.0.0.0:9096",
      "--rpc-port", "9097",
      "--advertise-addr", "node2",
      "--data-dir", "/data",
      "--node-id", "node-2",
      "--peer", "node-1=node1:9092",
      "--peer", "node-3=node3:9098"
    ]
    depends_on:
      node1:
        condition: service_healthy
    networks:
      - mlog-net

  node3:
    build:
      context: ..
      dockerfile: infra/Dockerfile
    container_name: mlog-node3
    hostname: node3
    ports:
      - "9098:9098"   # Serf
      - "9099:9099"   # Raft
      - "9100:9100"   # RPC
    volumes:
      - node3-data:/data
    command: [
      "server",
      "--bind-addr", "0.0.0.0:9098",
      "--raft-addr", "0.0.0.0:9099",
      "--rpc-port", "9100",
      "--advertise-addr", "node3",
      "--data-dir", "/data",
      "--node-id", "node-3",
      "--peer", "node-1=node1:9092",
      "--peer", "node-2=node2:9095"
    ]
    depends_on:
      node1:
        condition: service_healthy
    networks:
      - mlog-net

volumes:
  node1-data:
  node2-data:
  node3-data:

networks:
  mlog-net:
    driver: bridge

Key points:

  • node1 is the only node passed --bootstrap true — it's the one that forms the initial single-member Raft configuration. Every other node joins that existing cluster via --peer.
  • node2 and node3 declare depends_on: node1: condition: service_healthy, so Compose won't start them until node1's healthcheck passes — giving Raft time to elect a leader before anyone tries to join.
  • Every container binds its Serf/Raft/RPC listeners to 0.0.0.0 (so the Docker network can reach them) but advertises its own hostname (node1, node2, node3) via --advertise-addr, which is exactly what other containers resolve on the mlog-net bridge network.
  • Each node exposes all three of its ports to the host, at different host-side port numbers per node, so you can reach any node directly from outside Docker for topic admin, producing, or consuming.
  • node1's healthcheck probes the RPC port with nc -z localhost 9094, with a 20-second start_period to allow for Raft leader election before failures start counting.

Start the cluster

make cluster-up

This builds the images and starts all three containers in the background. Tail the logs to watch Raft elect a leader and the other two nodes join:

make cluster-logs

Stop the cluster

make cluster-down

To wipe all data volumes and start completely fresh:

make cluster-restart

Step 5: Run a 3-node cluster locally without Docker

scripts/start-local-cluster.sh

The script builds the server if needed, creates a data directory per node under /tmp/mlog-local, and starts three server processes on 127.0.0.1 with staggered delays so Raft has time to bootstrap before the other nodes try to join:

./scripts/start-local-cluster.sh

It starts node1 with --bootstrap true, sleeps 3 seconds, starts node2, sleeps 1 second, then starts node3 — each backgrounded with its PID appended to a PID file ($DATA_ROOT/pids, default /tmp/mlog-local/pids):

Node Serf Raft RPC Data dir
node-1 127.0.0.1:9092 127.0.0.1:9093 127.0.0.1:9094 /tmp/mlog-local/node1
node-2 127.0.0.1:9095 127.0.0.1:9096 127.0.0.1:9097 /tmp/mlog-local/node2
node-3 127.0.0.1:9098 127.0.0.1:9099 127.0.0.1:9100 /tmp/mlog-local/node3

Since every node runs on 127.0.0.1, none of them pass --advertise-addr — the literal --bind-addr/--raft-addr hosts are already reachable by the other two processes on the same machine.

scripts/stop-local-cluster.sh

./scripts/stop-local-cluster.sh

It reads the PID file, sends SIGTERM to every process, waits up to ~10 seconds (polling every 0.5s) for them to exit, then SIGKILLs anything still alive. As a safety net, it also scans the nine cluster ports (9092–9100) with lsof and kills any listener still bound to them — useful if the PID file is stale or missing.

Step 6: Create, list, and delete topics

The topic CLI

broker/cmd/topic is a small cobra program with three subcommands, all sharing a --addrs flag (default 127.0.0.1:9094) — a comma-separated list of RPC addresses to try:

make create-topic topic=orders replicas=3

which is a thin wrapper around:

go run ./broker/cmd/topic create --addrs 127.0.0.1:9094 --topic orders --replicas 3

or, once built:

./bin/topic create --addrs 127.0.0.1:9094 --topic orders --replicas 3

create and delete both call client.TryAddrs to send FindRaftLeaderRequest to each address in turn until one answers with a Raft leader address, then connect directly to that leader and send the actual CreateTopicRequest/DeleteTopicRequest — topic administration is always handled by the Raft leader, since it's the only node allowed to propose metadata changes. The leader picks a leader node and replica set for the topic and proposes it through Raft; once the change is committed, every node reconciles its local logs to match.

List topics:

make list-topics

Unlike create/delete, list doesn't need the Raft leader specifically — it just tries each --addrs entry in order until one responds, since every node's local metadata reflects the same Raft-committed state:

topic=orders leader=node-1 epoch=1 replicas=node-2(isr=true,leo=0),node-3(isr=true,leo=0)

Delete a topic:

make delete-topic topic=orders

Step 7: Produce messages to a topic

Interactive producer from stdin

make producer-connect topic=orders

or directly:

./bin/producer connect --addrs 127.0.0.1:9094 --topic orders

producer/cmd/main.go's connect subcommand calls producerclient.NewClient(ctx, addrs, topic), which discovers the topic's current leader among --addrs and connects — from then on the client itself handles rediscovery and reconnecting on failover, so the CLI never has to. Type messages and press Enter; Ctrl-D exits:

connected to topic "orders" leader at node1:9094
enter messages, each line will be produced to the topic (Ctrl-D to exit)
hello world
offset=0
this is a test
offset=1
distributed log works!
offset=2

Ack modes

./bin/producer connect --addrs 127.0.0.1:9094 --topic orders --acks 2

--acks accepts 0 (none), 1 (leader — the default), or 2 (all ISR replicas). With --acks 2 the client waits until every in-sync replica has the record before the CLI prints its offset — the strongest durability guarantee the system offers, at the cost of the extra round trip(s).

Step 8: Consume messages from a topic

Streaming from the beginning

In a separate terminal:

./bin/consumer connect --addrs 127.0.0.1:9094 --topic orders --id my-consumer --from-beginning

connected to topic "orders" leader at node1:9094
Starting from beginning (offset 0)
0	hello world
1	this is a test
2	distributed log works!

consumer/cmd/main.go prints each record as offset\tvalue, commits its offset back to the server after every message, and polls every 500ms once it catches up to the head of the log.

Note: the consumer CLI's built-in --addrs default is 127.0.0.1:9092, which is node1's Serf port, not its RPC port — pointing a consumer at it won't work. Always pass --addrs explicitly at an RPC port (9094, 9097, or 9100 in the layouts above). The producer and topic CLIs don't have this issue; both default --addrs to 127.0.0.1:9094.

Resuming from the last committed offset

Run the same command again without --from-beginning and without --offset:

./bin/consumer connect --addrs 127.0.0.1:9094 --topic orders --id my-consumer

Resuming from offset 3 (last committed)

The CLI calls FetchCommittedOffset and resumes from there; only if no committed offset exists does it fall back to offset 0. Passing --offset N explicitly starts from exactly N regardless of what's committed.

Step 9: Verify replication across nodes

Reading the same topic through different nodes

Any node can serve a fetch for a topic it doesn't lead — the request is transparently routed to whichever node the client's NewClient discovers as the current leader via cluster metadata:

./bin/consumer connect --addrs 127.0.0.1:9097 --topic orders --id test --from-beginning
./bin/consumer connect --addrs 127.0.0.1:9100 --topic orders --id test --from-beginning

Both return the identical stream, because every node's Raft-replicated metadata agrees on who the leader is, and the client dials that address directly.

Inspecting segment files on disk

On the local (non-Docker) cluster, each node's data directory holds its own copy of the topic's segments:

ls /tmp/mlog-local/node1/orders/
ls /tmp/mlog-local/node2/orders/
ls /tmp/mlog-local/node3/orders/

00000000000000000000.idx
00000000000000000000.log

The leader's log can be momentarily ahead of a follower's (records between the high watermark and the leader's LEO) until the follower's next replication fetch catches it up — within a second or so under replicationTickInterval.

Step 10: Test fault tolerance and recovery

Kill the leader

Check who's currently leading the topic:

make list-topics

Kill that node:

# Docker:
docker stop mlog-node1

# Local process: find its PID (e.g. from the PID file or `lsof -ti:9094`) and kill it

Serf detects the failure through gossip, Raft elects a new metadata leader among the surviving nodes, and the topic manager reassigns the topic's leader from its remaining ISR replicas.

Produce again, pointing at the surviving nodes — the producer client rediscovers and reconnects to the new leader on its own:

./bin/producer connect --addrs 127.0.0.1:9097,127.0.0.1:9100 --topic orders

reconnected to topic "orders" leader at node2:9097
after failover
offset=3

Consume — the consumer reconnects the same way, and every previously committed record (up to the high watermark before the crash) is still there on the new leader:

./bin/consumer connect --addrs 127.0.0.1:9097 --topic orders --id my-consumer

Restart the failed node

# Docker:
docker start mlog-node1

The restarted node rejoins the cluster through Serf, catches up on the Raft metadata log, and comes back as a follower rather than reclaiming leadership. Its replication thread fetches whatever records it missed while it was down, and the topic manager's reconciliation logic detects the returning node and adds it back into the replica set for any topic it used to serve.

Step 11: Makefile quick reference

build                    # Build all four binaries to bin/
build-server             # Build only the server (./broker/cmd/server)
build-producer           # Build only the producer (./producer/cmd)
build-consumer           # Build only the consumer (./consumer/cmd)
build-topic              # Build only the topic CLI (./broker/cmd/topic)

cluster-up               # docker compose up -d --build (3-node cluster)
cluster-down             # docker compose down
cluster-restart          # docker compose down -v, then up -d --build (fresh volumes)
cluster-logs             # docker compose logs -f

create-topic topic=X replicas=N [addrs=ADDRS]   # go run ./broker/cmd/topic create
delete-topic topic=X [addrs=ADDRS]               # go run ./broker/cmd/topic delete
list-topics [addrs=ADDRS]                        # go run ./broker/cmd/topic list

producer-connect topic=X [addrs=ADDRS]           # go run ./producer/cmd connect
consumer-connect topic=X [addrs=ADDRS] [id=ID]   # go run ./consumer/cmd connect

local-cluster-start      # scripts/start-local-cluster.sh
local-cluster-stop       # scripts/stop-local-cluster.sh

test                     # go test -v ./...

create-topic, delete-topic, and list-topics default addrs to 127.0.0.1:9094; producer-connect defaults to 127.0.0.1:9094 too. consumer-connect defaults addrs to 127.0.0.1:9092 (see the note in Step 8) — pass addrs=127.0.0.1:9094 (or another node's RPC port) explicitly to be safe.

What you've built

Across this book you've built a distributed log with:

  • Append-only storage — segments, sparse indexes, and memory-mapped reads.
  • A wire protocol — length-prefixed frames, a protobuf codec, structured RPC error codes, and a compact binary format for replication batches.
  • A TCP transport and RPC layer — connection handling, idle timeouts, and handler dispatch by message type.
  • Serf-based discovery — nodes find each other through gossip, with no static membership list to maintain.
  • Raft consensus — topic metadata, leader assignment, and ISR membership are replicated consistently across the cluster.
  • Topic management — create, delete, and list topics, with leader placement and reconciliation driven off Raft-committed state.
  • Pull-based replication — per-leader goroutine pools, a reused connection per leader, ISR tracking, and a high watermark that only advances once the ISR has caught up.
  • Producer and consumer client libraries — configurable ack modes, committed-offset tracking, and automatic reconnection on leader change.
  • A real, runnable cluster — built with make build, deployed with Docker Compose or plain local processes, observable through the topic CLI, and resilient to a node going down and coming back.

The result behaves like a small, honest version of Apache Kafka: the same vocabulary of offsets, segments, replication, ISR, high watermarks, and gossip-based discovery that production event-streaming systems are built on, implemented from the ground up in Go.