Consensus & Leader Electionlesson 4 of 4 · 3 min read

Using It Without Building It

Do not implement consensus

The practical advice here is short. Do not implement consensus.

Use one of the established coordination stores. Each is years of work with fault-injection test suites behind it, and hand-rolled leader election is a recurring cause of outages in published postmortems. Kubernetes keeps all of its cluster state in one. Kafka used another for years before building its own.

Care more about what you put in it than which one you pick. Consensus is for small critical state that everybody must agree on: who leads, which machine owns which shard, who is in the cluster, flags that must flip atomically.

Do not treat it as a database. Every write costs a round trip to a majority and every member stores a full copy, so throughput is thousands of writes a second, not millions. Putting application data in there is the classic misuse.

Elect your leaders with a lease, the standard pattern. A machine acquires a key with an expiry and keeps refreshing it while it works, and if it dies the key expires and somebody else acquires it.

Read carefully what that does and does not give you. It guarantees at most one holder of the lease. It does not stop a holder that paused for a long garbage collection from waking up after its lease expired, still believing it leads.

Fencing tokens are not optional

Add fencing tokens, then, because they are not optional. Each lease carries a number that only ever increases, the holder includes it in every write to shared storage, and the storage refuses any number lower than the highest it has seen.

Notice what that buys you. A stale leader's writes are refused by the resource itself, so your correctness no longer depends on assumptions about timing. This is the detail that separates a working answer from a plausible one in an interview.

Be honest about what depending on consensus costs. If your coordination store is down, nothing that depends on it can elect a leader or take a lock.

You concentrated availability into one component, which is the right trade when the alternative is split brain, and it needs to be a decision rather than an accident.

the shape of it

step 1 of 3

The lease alone cannot stop a paused holder, so the resource enforces order using the fencing token.

a lease is not enough on its own

Java

// Acquire a lease and refresh it while you work. This guarantees
// at most one holder, and it does not stop a holder that paused
// for a long garbage collection from waking up still believing it leads.
Lease lease = etcd.acquire("leader/indexer", Duration.ofSeconds(10));

// So every write carries the lease's number, and the number only
// ever increases.
storage.write(data, lease.fencingToken());

// The storage refuses anything below the highest token it has seen.
void write(byte[] data, long token) {
  if (token < highestSeen) throw new StaleLeader(token, highestSeen);
  highestSeen = token;
  ...
}

// Now a stale leader's writes are refused by the resource itself,
// and correctness stops depending on assumptions about timing.

Worked example

A team runs a batch job that must have exactly one active instance across 12 workers. The first version uses a Postgres row as a lock with a 60 second timeout, and it works until a worker hits a 90 second garbage collection pause: its lock expires, a second worker takes over, and then the first wakes up and writes results for a job someone else is now running, producing duplicate charges. The rewrite uses an etcd lease with a fencing token. Each acquisition returns an increasing revision number, workers stamp every output file with it, and the output store rejects any write carrying a revision below the highest seen. The same 90 second pause happens again two months later, the stale worker's write is rejected outright, and the incident is a log line instead of a refund run.

Consensus & Leader Election: wrapping up

In the real world

01Kubernetes keeps every object in etcd, so the control plane's correctness rests on Raft. A cluster that loses etcd quorum keeps running existing pods but cannot schedule or change anything.
02Kafka ran on ZooKeeper for controller election and metadata until KRaft replaced it, removing a whole external dependency and cutting the time to recover from a controller failure.
03CockroachDB runs a Raft group per data range rather than one for the whole cluster, so consensus scales horizontally instead of funnelling every write through one group.
04Google's Chubby is the original of this pattern: a small, highly available lock service that other systems use for leader election rather than each implementing it.
05Martin Kleppmann's 'How to do distributed locking' is the standard reference for why fencing tokens are required, written as a critique of using Redis locks without them.

Questions people ask

Why 3 or 5 nodes and never 4?

Fault tolerance is floor((N-1)/2). Three tolerates 1 failure and four also tolerates only 1, because a majority of 4 is 3. The fourth node adds cost and write latency for no extra safety, and it makes an even split possible. Go 3, then 5, and only past that if you genuinely need to survive 3 simultaneous failures.

Is a Redis lock good enough for leader election?

Usually not on its own. A TTL lock guarantees at most one holder only if nobody pauses longer than the TTL, and garbage collection pauses break that assumption. Either use a system built for it, etcd or ZooKeeper, or add fencing tokens so the resource rejects stale writes rather than trusting the timing.

What actually happens when a cluster loses quorum?

It stops accepting writes. A 3-node cluster that loses 2 nodes has 1 survivor, which is not a majority, so it refuses rather than risk diverging from the nodes it cannot see. Reads may still be served depending on configuration. This is the availability you traded away for consistency, and it is why quorum members belong in separate failure domains.

Quick review

The problem it solves:: failover has to pick exactly one new leader. Two nodes each believing they are primary is split-brain, and both accept writes that later conflict
Quorum:: any decision needs a majority, N/2 + 1. Two majorities cannot exist at once, which is what makes agreement safe across a partition
Run odd numbers:: 3 nodes tolerate 1 failure, 5 tolerate 2. Going from 3 to 4 still tolerates only 1, so the extra node adds cost and latency but no safety
Raft:: elects a leader per term, and all writes flow through it. A follower that hears nothing for its election timeout becomes a candidate and asks for votes
Randomised timeouts:: each node waits 150 to 300 ms before standing for election, so two candidates rarely tie. A split vote just retries in the next term
Log replication:: the leader appends an entry, ships it to followers, and commits once a majority acknowledge. That majority overlap is why a committed entry survives any leader change
Fencing tokens:: every leadership term carries a monotonically increasing number. Storage rejects writes stamped with an old term, which stops a paused ex-leader corrupting data
Do not implement it:: use etcd, ZooKeeper or Consul. Hand-rolled election is a classic source of outages, and Paxos is famously hard to get right in code

the trade-off

Every write costs a network round trip to a majority, so consensus is slow compared with a single node and gets slower as you add members. It also cannot survive losing the majority: a 3-node cluster that loses 2 nodes stops accepting writes rather than risking divergence, which is the correct behaviour and still an outage. Keep consensus for small, critical state like who is leader, and keep bulk data out of it.

in the room

Electing a database primary, assigning shard ownership, distributed locks, cluster membership, and any config that every node must agree on.

the short version

Put the decision in etcd, keep your data out of it.