Cluster Manager: Replication and Failover

In one line: replication here happens at partition granularity, not broker granularity — and that one choice is what makes every broker useful for both leading and following.

Three responsibilities

Replication at partition granularity

Each partition has one leader and two replicas, and the replicas sit on different brokers from the leader.

Per-partition leadership is why no broker is idle

Leadership is assigned per partition, not per broker. In the diagram, broker 1 leads partition 0 while simultaneously carrying a replica of partition 1, and broker 2 does the reverse.

Two things follow, and both are worth naming:

Write load spreads evenly. If leadership were per broker, one machine would take every write for a topic while its peers idled. Per-partition leadership means write traffic distributes across all brokers holding leaders.

Failover is granular. Losing broker 2 costs you the leadership of partition 1 only — partitions 0 and 2 keep serving from their own leaders, unaffected. The blast radius is one partition, not one topic.

Contrast this with message-queue design's primary-secondary model, where a node was primary for a set of queues and losing it stalled all of them. Splitting leadership at the partition level is a strictly finer-grained version of the same idea.

Why three replicas, specifically

Three is the standard answer across distributed systems — the Key-Value Store and Distributed Cache chapters both landed on it — and the reasoning is the same here.

Two replicas survive one failure but leave you with no redundancy during the repair window: a second failure loses data, and repairs take hours at terabyte scale. Three survives one failure and still has a spare while you rebuild.

Three also permits quorum protocols — a majority of three is two, so a write acknowledged by two survives any single loss. That gives a middle position between fully synchronous and fully asynchronous replication, which is exactly the trade the messaging queue chapter framed.

Beyond three, you pay linearly in storage for diminishing returns, which is why three is the near-universal default rather than a coincidence.

'Across different brokers' is the constraint doing the work

Three replicas on one broker is three copies of nothing — the machine dies and all three go with it. The requirement is that they are on different brokers, which is what makes them independent failure domains.

Production systems push this further with rack awareness and availability-zone awareness: three replicas on three brokers in the same rack still share a power supply and a top-of-rack switch. Spreading across racks or zones makes the failures genuinely independent.

That is the same reasoning as distributed caching's warning that replicas in one data center do not survive losing the data center — replication only buys availability against failures your placement actually separates.

Leader failure

If a leader fails, the manager elects a new one.

What actually matters here is what the new leader is allowed to serve

Which is why "acknowledge only once the in-sync replicas have it" and "elect only from the in-sync set" are the same guarantee stated at two ends: the first defines what durable means, the second is what makes the promise survive a failure.

Election is the visible part. The subtle part is that a promoted replica must not serve data it does not have.

If replication is asynchronous, a follower may be behind the failed leader. Promoting it means messages the old leader had acknowledged are gone — the producer was told the write succeeded, and no consumer will ever see it. This is exactly the silent-loss failure mode from message-queue design, and it applies unchanged.

There is a second, offset-specific hazard here. Consumers track offsets. If a promoted leader has a shorter log than the old one, an offset a consumer already holds may now point past the end or, worse, at a different message. Real systems solve this with a high-water mark — consumers may only read up to the offset replicated to all in-sync replicas, so a failover can never invalidate an offset a consumer was allowed to see.

The general rule from the cache chapter holds: a node must not serve until it can prove it is current.

Authorization, including across clusters

Controls access to brokers and topics, as well as message replication across clusters.

Cross-cluster replication is the multi-region story, mentioned in passing

"Message replication across clusters" is a single clause, but it is how these systems span regions — Kafka's MirrorMaker does exactly this.

The reason it is an authorization concern rather than just a plumbing one: replicating a topic to another cluster copies data across a trust boundary, potentially a jurisdictional one. Which topics may leave a region is a governance question before it is an engineering question.

And the same trade the cache chapter drew applies: cross-cluster replication is asynchronous — the latency makes anything else untenable — so a remote cluster is always somewhat behind. Fine for disaster recovery or feeding a regional analytics stack; not something to build strong-consistency guarantees on.

The cluster manager is itself a dependency worth defending

It holds the registry, elects leaders, and gates authorization. If it is slow, leader elections drag and partitions stay unavailable longer. If it is down, no failover happens at all and the next broker failure becomes an outage.

Which is the same conclusion distributed caching reached about its configuration service and the messaging queue chapter reached about its cluster manager: the coordinator needs its own replication and consensus, or you have moved the single point of failure rather than removed it. In practice this role is played by a consensus system — ZooKeeper historically, Raft-based controllers now.

Interview signal by level

Next: the component that tracks where every consumer is.