Consensus is the problem of getting several machines to agree on one value, or one ordered sequence of values, even when some of them crash, restart or are cut off by the network. Coordination is the everyday name for the things you build on it: electing a single leader, granting a lock, storing configuration everyone sees the same way, tracking which nodes are alive. Interviewers probe it because it is where distributed systems become hard: "just use a distributed lock" hides lease expiry, process pauses, clock assumptions and split brain, and candidates who cannot explain fencing tokens usually cannot explain why their design is safe. For a Backend or AI engineer it matters because the schedulers, databases and control planes you run on (Kubernetes, etcd, Kafka's controller, sharded vector stores, training-job rendezvous) all depend on it, and because you will be tempted to reuse it for things it is bad at. The protocol-level treatment is in Consensus — Raft & Paxos; this page is the working-engineer view.
Consensus is when a group of computers, each of which might crash, slow down, or lose its network connection at any moment, needs to agree on a single answer, like "who's in charge" or "what happened first." It matters because without an agreed process for reaching that agreement, different machines can end up with different, conflicting answers.
Coordination is the practical, everyday use of that agreement: deciding which single machine gets to be the leader right now, who is allowed to make a change at a time, or making sure everyone is looking at the same shared settings. It's the layer that keeps a group of independent machines acting like one dependable system instead of a crowd of machines working at cross purposes.
Picture a group of siblings sharing one family car, none of whom live in the same house and none of whom can just shout across the room to the others. Consensus is the rule they agree on for deciding who gets the car today, say, whoever texts the family group chat first and gets a "yes" reply from at least two others. Coordination is actually using that rule day to day: checking who has the car, updating everyone when plans change, and making sure two siblings don't both show up expecting to drive it at the same time.
Problem What must be agreed Typical primitive
------------------------- --------------------------------- ---------------------------
Leader election "exactly one primary right now" ephemeral node / lease + term
Mutual exclusion "one holder of lock X" lock with lease + fencing token
Configuration / flags "everyone sees version 42" linearizable key-value + watch
Membership / discovery "who is alive and serving" session heartbeat / ephemeral keys
Ordering / sequencing "a single order of commits" replicated log
Shard / partition owner "which node owns partition 7" compare-and-swap on an ownership key
Why not just "ask a server"? That server is one machine: if it dies, nobody can decide;
if you add a second one, they can disagree (split brain).Without coordination, two replicas can each decide they are the primary and accept conflicting writes; two workers can each believe they own the same shard and double-process it; half a fleet can run with the new model-routing config and half with the old. The general trick is to concentrate the small amount of state that must be agreed on into a replicated system that tolerates failure of a minority, and have everyone else consult it. That system is a consensus group, usually 3 or 5 nodes, and it is a control-plane component: small data, strong consistency, modest write rate. Application data with high throughput lives elsewhere, and coordination only decides who is allowed to touch it. The consistency vocabulary (linearizable, sequential, eventual) is covered in CAP Theorem & Consistency Models; consensus is how a CP system provides the linearizable side.
Roles: follower, candidate, leader. Time is divided into TERMS (a monotonically increasing epoch).
Election
follower hears nothing from a leader for a randomised election timeout
-> becomes candidate, term += 1, votes for itself, sends RequestVote to all
-> a node grants at most ONE vote per term, and only to a candidate whose log is
at least as up to date as its own
-> majority of votes => leader for that term; split vote => timeout, new term, retry
(random timeouts make repeated ties unlikely)
Replication (leader only accepts writes)
client --> leader: append entry to its log
--> AppendEntries to followers (also serves as heartbeat)
--> entry COMMITTED once stored on a majority (n=5: leader + 2 followers)
--> leader applies to state machine, replies to client
A crashed leader's committed entries survive: any new leader must have won a majority
whose logs overlap the old commit majority.
n=5, term 3: L:[1 2 3 4] F1:[1 2 3 4] F2:[1 2 3 4] F3:[1 2 3] F4:[1 2]
entry 4 is on 3 of 5 nodes -> committed (F3 and F4 are merely lagging and will catch up).The two safety ideas to be able to state are majority overlap and terms. Any two majorities of the same cluster share at least one node, so a new leader elected by a majority necessarily talks to at least one node that saw every committed entry, and the voting rule (only vote for a candidate whose log is at least as up to date) makes sure it wins only if it has them. Terms act as a logical clock: a node that sees a higher term steps down, so a deposed leader that wakes up cannot keep writing. The cost is on the write path: every committed write needs a round trip to a majority plus a durable disk write (an fsync) on those nodes, so commit latency is set by the slowest member of the fastest majority and by disk. A cluster spread across regions pays wide-area round-trip time on every write, which is why consensus groups are usually kept inside one region or use carefully chosen quorums.
Paxos solves the same problem with a different presentation: single-decree Paxos agrees on one value through prepare/promise and accept/accepted phases with proposal numbers, and Multi-Paxos strings decisions into a log with a stable leader. In practice, "Raft-like" is what you meet: etcd, Consul and CockroachDB's ranges use Raft; Google's Chubby and Spanner use Paxos variants; ZooKeeper uses its own atomic broadcast protocol, ZAB; newer Kafka versions replace their ZooKeeper dependency with KRaft, a Raft-based controller quorum (check the version you run). Reads need care: a read served by a stale leader (or a follower) may be out of date, so linearizable reads either go through the log, use a read-index round trip, or rely on a leader lease; some systems offer cheaper stale reads explicitly.
ZooKeeper etcd
- znodes in a tree (small data) - flat key-value store with revisions
- EPHEMERAL nodes: deleted when the - LEASES: keys attached to a lease die when the
client's session expires lease is not kept alive
- SEQUENTIAL nodes: auto-incrementing - every write has a global, increasing revision
suffix - transactions: compare (value/revision) then put
- WATCHES: one-shot change notification - WATCH streams from a revision (can replay history)
Leader election recipe (ZooKeeper): each candidate creates /election/n_ (ephemeral+sequential);
the lowest sequence number is leader; each other candidate watches its predecessor only
(avoids a "herd" waking on every change).
Kubernetes: controllers elect a leader via a Lease object stored in etcd; all cluster state
goes through the API server to etcd, which is why etcd health = control-plane health.These services are consensus packaged as a small API, and their value is the primitives: sessions and ephemeral keys turn "process died" into an event; watches turn "config changed" into a push; compare-and-swap and revisions turn "who is first" into a database operation. Keep the data small (they are designed for kilobytes of metadata, not for your embeddings; etcd, for example, enforces a storage quota that is small by default, so check current docs) and keep the cluster small and odd-sized, typically 3 or 5 nodes. More nodes do not add throughput: they add replication work and a bigger quorum. For an AI platform they show up as leader election for schedulers and controllers, model or feature-flag configuration, ownership of shards in a sharded index, service membership for inference replicas and rendezvous for distributed training jobs; see Container Orchestration (Kubernetes Basics) for the control plane view.
Naive lock: acquire(lock, ttl=10s) -> do work -> release
time ----------------------------------------------------------------------->
Client A: acquire (t=0) check ok (t=1) [ GC pause / VM freeze / slow disk ..... ] write (t=15)
Lock svc: A holds lock ................... lease EXPIRES at t=10 ...
Client B: acquire (t=12) write (t=12)
Storage: B's write ---> A's stale write overwrites it
A cannot detect the pause from inside: its own check "do I still hold the lock?" was true
at t=1, and nothing can make check-then-write atomic across a network.
Same effect from: stop-the-world GC, container CPU throttling, live migration, a delayed
network packet, a clock that jumped, a lease shortened by clock drift.
FIX - fencing token: the lock service returns a monotonically increasing number with each
grant; every write to the protected resource carries it; the RESOURCE rejects any token lower
than the highest it has already seen. Safety no longer depends on the client's timing.A lock with a timeout (a lease) is needed because a crashed holder must not block everyone forever, and that is precisely what makes it unsafe: the service can only decide the holder is dead by waiting, and the holder cannot know whether it has been declared dead. Any design that says "I checked I still hold the lock, so I can write" has a time-of-check to time-of-use gap. Longer leases shrink the window but never close it, and they lengthen failover. The only sound fix is for the protected resource to take part: it must be able to compare tokens, so it needs a conditional write, a version column, or a compare-and-swap. If the resource cannot do that (a third-party API, an email), the lock only offers best-effort mutual exclusion and you need idempotency instead (see Idempotency & Exactly-Once Semantics).
The program in Try It Yourself reproduces the pause with a simulated clock: client A gets the lock with a 10 second lease and freezes; the lease expires; B takes over and writes; A wakes and writes. First without fencing, then with a storage that rejects stale tokens. It also prints quorum arithmetic, because "why 3 or 5 nodes, not 4" is a standard follow-up.
Partition splits a 5-node cluster: [ A B C ] | [ D E ]
side with 3 (majority): can elect a leader and commit.
side with 2 (minority): cannot commit; its leader (if any) steps down / rejects writes.
=> at most one side makes progress: no split brain, at the price of availability on the minority side (CP).
Split brain WITHOUT quorum (2 nodes + a heartbeat): link breaks, each side decides the other died,
both promote themselves, both accept writes; after healing you must discard or merge one history.
Mitigations: quorum (odd n or a witness), fencing / STONITH (power off the old primary), leases.
Choose the mechanism by need:
single writer for one DB row / job -> row lock, SELECT ... FOR UPDATE, unique constraint, advisory lock
dedupe of repeated work -> idempotency key, conditional write (compare-and-set)
ordering of events per key -> partitioned log (single consumer per partition)
eventually consistent, mergeable -> CRDTs / last-writer-wins, no coordination on the hot path
truly global agreement, small data -> etcd / ZooKeeper / Raft groupConsensus is a tool for agreeing on small, critical state, and its cost is latency on every decision plus lost availability when a majority is unreachable. So do not put it on a hot path: not per-request, not for bulk data, not for counters that can be approximated, and not where a database you already run can do the job with a unique constraint or a conditional update. A 3-node etcd also becomes a shared dependency, and a slow disk or a noisy neighbour can stall the control plane of everything using it. Leaderless and eventually consistent designs (see Database Replication & Sharding) avoid coordination on the write path and reconcile later, at the price of anomalies the application must accept. Also plan for the coordinator being unavailable: caches of the last known config, a defined behaviour for "cannot reach the lock service" (fail closed for correctness, fail open for efficiency locks), and timeouts and circuit breakers as in Fault Tolerance & Circuit Breakers.
A simulated clock makes the lease-expiry race deterministic (stdlib only). Client A takes a 10 second lease and freezes for 14 seconds; B takes over legitimately; A wakes and writes. Then the same run against a storage that enforces fencing tokens. Experiment: shorten A's pause so it wakes before the lease ends, and see that fencing then changes nothing; then add a "renew" call and observe that renewal still cannot make the check and the write atomic.
class LockService:
"""Lease-based lock on a simulated clock. Every grant gets a higher fencing token."""
def __init__(self):
self.holder, self.expires, self.token = None, 0, 0
def acquire(self, client, now, lease):
if self.holder is None or now >= self.expires: # free, or previous lease expired
self.holder, self.expires = client, now + lease
self.token += 1 # monotonically increasing
return self.token
return None
def still_mine(self, client, now):
return self.holder == client and now < self.expires
class Storage:
"""Shared resource. If fencing=True it rejects any token lower than the highest it has seen."""
def __init__(self, fencing):
self.fencing, self.max_token, self.value, self.log = fencing, 0, None, []
def write(self, client, token, value, now):
if self.fencing and token < self.max_token:
self.log.append(f"t={now:>2} {client} write REJECTED (token {token} < {self.max_token})")
return False
self.max_token = max(self.max_token, token)
self.value = value
self.log.append(f"t={now:>2} {client} wrote {value!r} (token {token})")
return True
def scenario(fencing):
lock, store = LockService(), Storage(fencing)
LEASE = 10
tok_a = lock.acquire("A", now=0, lease=LEASE) # A gets the lock at t=0, lease until t=10
check_a = lock.still_mine("A", now=1) # A checks "am I still the owner?" at t=1: yes
# ... A now stalls (GC pause / VM freeze / slow network) for 14 seconds ...
tok_b = lock.acquire("B", now=12, lease=LEASE) # t=12: A's lease expired, B legitimately takes over
store.write("B", tok_b, "B's result", now=12)
store.write("A", tok_a, "A's stale result", now=15) # t=15: A wakes up, still believes it holds the lock
print(f"fencing={fencing} A token={tok_a} B token={tok_b} A's check at t=1 said: {check_a}")
for line in store.log: print(" ", line)
print(" final value:", repr(store.value))
scenario(fencing=False)
print()
scenario(fencing=True)
print("\nQuorum sizes (majority = n // 2 + 1):")
for n in (3, 4, 5, 7):
print(f" n={n} majority={n // 2 + 1} tolerates {n - (n // 2 + 1)} failure(s)")
# Output:
# fencing=False A token=1 B token=2 A's check at t=1 said: True
# t=12 B wrote "B's result" (token 2)
# t=15 A wrote "A's stale result" (token 1)
# final value: "A's stale result"
#
# fencing=True A token=1 B token=2 A's check at t=1 said: True
# t=12 B wrote "B's result" (token 2)
# t=15 A write REJECTED (token 1 < 2)
# final value: "B's result"
#
# Quorum sizes (majority = n // 2 + 1):
# n=3 majority=2 tolerates 1 failure(s)
# n=4 majority=3 tolerates 1 failure(s)
# n=5 majority=3 tolerates 2 failure(s)
# n=7 majority=4 tolerates 3 failure(s)Read the output: without fencing the final value is the stale one, B's newer result was silently overwritten by a client that no longer owned the lock, and A's own liveness check had said yes. With fencing the storage rejects token 1 after seeing token 2, so correctness no longer depends on A noticing the pause. Practical token sources are etcd's revision numbers, ZooKeeper's zxid or sequential node number, or a counter incremented in a database row. Redis-based locks (single instance or the Redlock algorithm) are commonly discussed here: Martin Kleppmann's critique argues they are not safe as a correctness mechanism because they lack a fencing token and rely on timing assumptions, while they are reasonable for an efficiency lock (avoiding duplicate work that would be harmless anyway). Know both sides of that debate and state which purpose the lock serves.
The same idea against real infrastructure, as labelled pseudocode (no output is claimed):
# etcd-style sketch (pseudocode, not executed here)
lease = etcd.grant_lease(ttl=10)
ok, rev = etcd.txn(
compare=[create_revision("/locks/job-7") == 0], # only if nobody holds it
success=[put("/locks/job-7", client_id, lease=lease)],
)
token = rev # the revision of the successful write is the fencing token
storage.write(data, fencing_token=token) # storage: UPDATE ... WHERE last_token < :tokenQ: Why do distributed systems need consensus?
A: Whenever several nodes must agree on one decision (who is leader, who holds a lock, what the current config is) despite failures, a single coordinator becomes a single point of failure and two coordinators can disagree. Consensus replicates the decision across a group so that any majority can continue and no two majorities can decide differently. The trade-off is added latency and reduced availability when a majority is lost.
Q: Explain how Raft elects a leader and why it is safe.
A: A follower that misses heartbeats becomes a candidate, increments the term and asks for votes; each node grants one vote per term and only to a candidate with an up-to-date log. A majority wins, and randomised timeouts make repeated split votes unlikely. Safety comes from majority overlap: the winner's supporters include at least one node that holds every committed entry, so committed entries are never lost.
Q: When is a write "committed" in Raft, and what does that cost?
A: When the leader has stored it on a majority of nodes (its own log included), at which point it can be applied and acknowledged. Latency is therefore about one round trip to the nearest majority plus an fsync on those nodes, illustratively a few milliseconds inside a data centre and tens of milliseconds or more across regions, depending on the network and disks. That is why consensus groups handle metadata rather than high-volume data.
Q: Why 3 or 5 nodes and not 4 or 6?
A: Majority is n/2 + 1 rounded down, so 4 nodes need 3 and tolerate 1 failure, exactly like 3 nodes, but they cost more and have a slower quorum. Odd sizes give the most fault tolerance per node; 5 nodes tolerate 2 failures, and going beyond that rarely helps because every write must reach a larger majority.
Q: Why is a distributed lock with a TTL not safe by itself?
A: The holder can stall (GC pause, VM freeze, network delay) past the lease, the lock service hands the lock to someone else, and the first client then resumes believing it still owns it. Checking "do I hold the lock?" right before writing does not help because the check and the write are not atomic across the network. A fencing token verified by the resource makes the stale write fail, at the cost of requiring the resource to support conditional writes.
Q: What is split brain and how do you prevent it?
A: Two nodes or partitions both believe they are the leader and accept writes, producing divergent state. Prevent it with quorum (only the side with a majority may act, so a minority partition stops), leases so leaders step down if they cannot renew, and fencing (tokens or STONITH) to neutralise a deposed leader. Two-node clusters cannot do this safely without a third vote or witness.
Q: When would you not use ZooKeeper or etcd?
A: For high-throughput or large data, for per-request decisions, and whenever a unique constraint, a conditional database update, a row lock or an idempotency key already gives the guarantee you need. They are also unsuited to being your only copy of business data. Adding one means running and monitoring another stateful cluster that everything now depends on.
Follow-up: Is a Redis SET key value NX PX 10000 lock fine for guarding a payment?
A: Not as the only protection. It provides mutual exclusion under normal timing but has no fencing token, and a failover of an asynchronously replicated Redis can lose the key so two clients hold it. Use it to reduce duplicate work, and make the payment itself safe with an idempotency key or a conditional write in the database.
Follow-up: A leader is partitioned from the majority but still thinks it leads. What stops it doing damage?
A: It cannot commit anything because commits need a majority, and when the partition heals it sees a higher term and steps down. The remaining risk is stale reads and side effects outside the consensus group (external calls it makes while it still believes it leads), so leaders should stop serving reads when their lease or heartbeat contact lapses, and external effects need fencing or idempotency.
Follow-up: Do fencing tokens require the lock service to be perfectly available?
A: No, but they require the lock service itself to be linearizable, otherwise two clients could receive the same or non-monotonic tokens; that is why it is backed by consensus. If the lock service is down, no new tokens are issued, so new work stops, which is the safe failure mode for a correctness lock.
Consensus appears in designs as the control plane, not the data plane: a 3 or 5 node etcd or ZooKeeper ensemble holding leader leases, shard-to-node assignments, feature flags and model-routing configuration for an inference fleet; Kubernetes controllers electing a leader through a Lease object; a database using Raft per shard for its replicated log (see Database Replication & Sharding); a streaming broker keeping its metadata in a Raft quorum (see Message Queues & Event Streaming). Quote assumptions when estimating: with roughly 1 ms round trip inside one availability zone and an SSD fsync in the low milliseconds (ILLUSTRATIVE, hardware dependent), a coordination write costs a few milliseconds, so a service doing 5,000 requests per second must not consult the coordinator per request; it should cache the lease holder or config and refresh it every few seconds, with a watch for changes. Explain the failure story as well: what happens to the workers when the coordinator loses quorum (they keep running on cached state for a bounded time and stop taking new leadership), how long failover takes (bounded by the lease TTL plus an election, typically seconds), and how a stale leader is fenced. State clearly whether each lock is for efficiency or for correctness.
Adopting consensus buys a single, agreed-upon truth that survives the failure of a minority of nodes; it costs write latency (a majority round trip and a disk sync), lost availability during partitions on the minority side, and an operationally sensitive cluster whose health everything else inherits. Leases trade failover speed against safety: short leases fail over fast but expire during a GC pause or slow disk, long leases are safer against false expiry but leave the system leaderless for longer after a real crash; fencing tokens remove the dependency on timing for correctness but require resources that support conditional writes. It is the wrong choice for high-volume data, latency-critical per-request paths, geographically spread writes where every commit pays a wide-area round trip, and for problems that a database constraint, an idempotent operation, or a partitioned log already solves more simply. Weaker designs that tolerate temporary duplicates or conflicts (idempotent handlers, CRDTs, last-writer-wins with reconciliation) are frequently the better engineering answer, so justify any coordination you add.