Consensus and Coordination Algorithms Questions
How independent nodes agree on shared state: Paxos and Raft, leader election, quorum reads and writes, distributed locks, coordination services such as ZooKeeper or etcd, and Byzantine fault tolerance for replicas that cannot be trusted. Covers split-brain avoidance, fencing tokens, and the cost of coordination on throughput and latency. Frames when consensus is required versus when it can be designed away.
Design a follower-reads optimization for a strongly-consistent key-value system that allows slightly stale reads from followers while guaranteeing clients can request fresher data when needed. Describe API semantics, staleness bounds, and how you would ensure monotonic reads and session guarantees.
Sample Answer
Direct answer
Let clients choose the freshness they need per read, and make every follower able to prove how fresh it is before it answers. The API offers three modes: linearizable (as fresh as a leader read), bounded staleness (no older than a stated interval), and at least this version (a session token). Followers track a closed timestamp (a time the leader has promised no new write will ever be committed at or below) and their applied log index; a follower serves a read only if that evidence satisfies the request, otherwise it waits briefly or redirects. Monotonic reads and read-your-writes come from a session token the client carries: the highest log index it has written or observed.
Background in plain terms
In a strongly consistent key-value store built on Raft or Paxos (consensus algorithms that commit each write once a majority of replicas has stored it), one leader orders all writes into a log and followers apply the log a little later. Reading from the leader is always fresh but concentrates load and, across regions, adds latency. Reading from a follower is cheap and local but might miss the last few writes. The design problem is to make follower reads safe by construction: the system, not luck, decides whether a follower is fresh enough.
API semantics
Get(key, consistency)
consistency =
LINEARIZABLE -- reflects every write acknowledged before the call
BOUNDED_STALENESS(max_age) -- reflects every write committed at or before now - max_age
AT_LEAST(session_token) -- reflects every write up to the index in the token
Response: value, served_index, served_by, freshness_timestamp, new_session_token
Put(key, value) -> commit_index, new_session_token
- Default is LINEARIZABLE, so nobody gets stale data by accident.
- Every response returns the index it reflects, so clients and debugging tools can see exactly how fresh an answer was.
- A request that cannot be satisfied locally within a small wait budget is forwarded to the leader, never answered stale.
Real systems expose the same menu: CockroachDB has exact-staleness reads (reading as of one specific past timestamp you name, rather than "as fresh as possible" or "within some window"; follower_read_timestamp()) and bounded-staleness reads (with_max_staleness(...), with_min_timestamp(...)); Consul has stale, default and consistent read modes and returns X-Consul-LastContact so a client can see how long ago the answering server heard from the leader.
How each mode is served from a follower
Linearizable follower reads (ReadIndex)
ReadIndex is Raft's technique for letting a follower serve a linearizable read without itself becoming leader: it asks the current leader for the leader's latest commit index, confirms that leader is genuinely still in charge, then simply waits until its own state has caught up to that index before answering.
sequenceDiagram
participant C as Client
participant F as Follower
participant L as Leader
C->>F: Get(k, LINEARIZABLE)
F->>L: What is your commit index?
L->>L: Confirm leadership with a majority heartbeat
L-->>F: commit index 1042
F->>F: Wait until applied index is at least 1042
F-->>C: value at index 1042
Here, commit index is how far the leader has confirmed a majority of replicas have durably stored the log, while applied index is how far this particular follower has actually executed those log entries into its own copy of the data, which can lag the commit index slightly. A read is only safe to answer once the follower's applied index has caught up to the commit index the leader just reported, which is exactly what the wait step checks. The follower pays one round trip to the leader (small message, no data), but the data read and its bandwidth stay local. Confirming leadership matters because the process answering "what is your commit index" might already have been replaced: the heartbeat check makes it confirm, via a fresh exchange with a majority of the cluster's members, that it is still recognised as leader right now, and Raft guarantees only one leader can win a majority in a given term. A leader that has actually been deposed cannot pass that check, so it cannot hand out a commit index at all, rather than handing out a stale one.
Bounded staleness
The leader periodically publishes a closed timestamp T with the log index where it takes effect. A follower that has applied up to that index can serve any read at a time at or below T. It serves BOUNDED_STALENESS(max_age) if
where epsilon is the maximum clock uncertainty between nodes. Worked example: the leader closes timestamps about 3 s behind real time, clock uncertainty is 250 ms, and a client asks for data at most 10 s old. A follower whose closed timestamp is 3.2 s behind satisfies 3.2 + 0.25 = 3.45 s, which is within 10 s, so it answers locally. A follower cut off from the leader for 12 s has a closed timestamp at least 12 s behind; it must redirect. The staleness bound is only as good as the clock-uncertainty bound, so the system must track clock offset and refuse bounded reads on a node whose clock is out of tolerance.
Session guarantees
A session is one client's sequence of operations. Two guarantees matter:
- Monotonic reads: once you have seen version N, you never later see anything older.
- Read-your-writes: after your write commits at index N, your reads reflect it.
Both reduce to one rule: the client carries a token equal to the highest index it has written or read, and every read is AT_LEAST(token). This simulation shows the failure without the token and the fix with it:
class Replica:
def __init__(self, name):
self.name, self.applied, self.data = name, 0, {}
def apply(self, index, key, value):
self.data[key], self.applied = value, index
class Client:
'''Carries a session token: the highest log index it has written or read.'''
def __init__(self):
self.token = 0
def read(self, replica, key, use_token=True):
need = self.token if use_token else 0
if replica.applied < need:
return f"{replica.name} applied={replica.applied} < token={need}: wait or redirect"
self.token = max(self.token, replica.applied)
return f"{replica.name} applied={replica.applied} -> {key}={replica.data.get(key)}"
log = [(1, "bal", 100), (2, "bal", 80), (3, "bal", 50)]
fast, slow = Replica("f1"), Replica("f2")
for i, k, v in log:
fast.apply(i, k, v)
for i, k, v in log[:1]:
slow.apply(i, k, v) # f2 is two entries behind
for label, use_token in [("without session token", False), ("with session token", True)]:
c = Client()
print(label)
print(" read 1:", c.read(fast, "bal", use_token))
print(" read 2:", c.read(slow, "bal", use_token))
Output:
without session token
read 1: f1 applied=3 -> bal=50
read 2: f2 applied=1 -> bal=100
with session token
read 1: f1 applied=3 -> bal=50
read 2: f2 applied=1 < token=3: wait or redirect
Without the token, a load balancer that sends the second read to a lagging follower makes the balance jump from 50 back to 100. With it, the lagging follower knows it is not allowed to answer.
Practical details for the token:
- Return it in every response and accept it on every request; store it in the client library, or in a cookie for browser sessions that span servers.
- Because a Raft log index of a committed entry never changes meaning across leader changes, the index alone is a valid token; include the term too (the number that increases every time a new leader is elected) if you want to detect a client talking to a cluster that was restored from backup.
- The wait budget is short (tens of milliseconds); after that, redirect to the leader rather than queue indefinitely.
Staleness bounds you can promise
- Bounded staleness: bounded by max_age, provided clocks are within epsilon. Without clock bounds, only index-based guarantees are safe.
- Session reads: no wall-clock bound at all; the guarantee is ordering relative to what the client has seen.
- Unbounded "any replica" reads should not be offered on this system, or only behind an explicit name like
STALE_OKfor diagnostics.
Trade-offs and pitfalls
- Serving follower reads for authorisation decisions (can this user withdraw?) reintroduces the anomalies consensus was bought to prevent; keep those on LINEARIZABLE.
- Closed timestamps lagging under write load push bounded reads to the leader exactly when it is busiest; alert on closed-timestamp lag and size max_age against it.
- Forgetting the token on retries or across devices silently drops the guarantee; make the client library carry it automatically.
- Using a wall-clock timestamp as the session token breaks under clock skew; use the log index.
When would you recommend using strong-consensus protocols such as Raft or Paxos versus eventual-consistency approaches? As a Solutions Architect, compare latency, throughput, failure modes, operational overhead, and appropriate use cases like metadata management versus user-facing high-volume data.
Sample Answer
Direct answer
Use a strong-consensus protocol (Raft or Paxos: replicas commit each write only once a majority has stored it, so everyone sees one agreed order) for small, high-value state where a conflict or a lost write is a correctness failure: cluster membership, leader leases, shard maps, schema and configuration, locks. Use eventual consistency (replicas accept writes independently and converge later) for large, high-volume, user-facing data where a briefly stale read is harmless and conflicts can be merged: timelines, activity feeds, counters, carts, telemetry. Before reaching for either, check whether a plain relational database already gives you the consistency you need, because for a lot of "metadata" it does, and you already operate it.
The comparison
Throughout this table, a quorum is the minimum number of replicas that must respond before an operation counts as done; a write quorum of 1 means one replica's acknowledgement is enough, while consensus protocols require a majority.
| Axis | Strong consensus (Raft/Paxos) | Eventual consistency (leaderless, meaning any replica can accept a write directly with no elected coordinator, or async replication) |
|---|---|---|
| Write latency | At least one round trip from leader to the nearest majority, plus a disk sync; cross-region this means tens to hundreds of ms | Can acknowledge after one local replica (write quorum of 1), so local-network latency |
| Read latency | Linearizable reads (every read reflects all writes acknowledged before it started) need the leader to confirm it still leads; stale reads can come from any replica | Any replica, local latency, possibly stale |
| Throughput | Bounded by one leader per consensus group; scale by splitting data into many groups | Scales roughly with replica count; any node accepts writes |
| Behaviour during a network partition | Minority side stops accepting writes (chooses consistency over availability) | Both sides keep working; conflicts appear and must be resolved |
| Failure modes | Unavailability: quorum loss, leader-election gaps, slow disk causing leader churn | Anomalies: stale reads, lost updates under last-writer-wins (when two writes conflict, whichever carries the later timestamp wins and the other silently vanishes), resurrected deletes (a deleted item reappearing because a replica that missed the delete, or an old un-deleted copy, gets merged back in before the delete marker has fully propagated), divergence needing repair |
| Operational overhead | Odd-sized voter sets across failure domains, careful membership changes (adding or removing a voting replica has to go through a protocol step, not just editing a config file, or it can split one majority into two disjoint ones), disk and network latency sensitivity, snapshots and compaction (periodically writing the full state to disk, then discarding the older log entries it now makes redundant, so the log does not grow forever) | Anti-entropy repair (a background job where replicas compare their data, often via hashes, and reconcile anything that never fully propagated), tombstone management (keeping delete markers around long enough for the delete to reach every replica before safely discarding them), conflict-resolution code, harder correctness testing |
| Typical data size | Small: etcd's documented default storage quota is 2 GiB with 8 GiB suggested as a maximum, and requests default to at most 1.5 MiB | Terabytes to petabytes |
Why metadata and user data fall on opposite sides
Metadata (which node owns shard 17, which instance is the active scheduler, what the current config version is) is small, read often, written rarely, and every component acts on it. If two nodes disagree about who owns shard 17, both may write to it and corrupt it. The cost of consensus (a few milliseconds per rare write) is trivial next to that risk.
User-facing high-volume data (likes, views, a social feed, sensor readings) is large and written constantly. A "like" counter off by one for two seconds harms nobody; making every like wait for a cross-region majority would add latency to the hottest path and cap throughput at what one leader per group can order.
Worked example: a feed service takes 50,000 writes per second. With consensus sharded into groups whose leaders each sustain, say, 5,000 ordered writes per second (an assumed figure; measure your implementation), you need at least 50,000 / 5,000 = 10 groups before any headroom, each with 3 or 5 replicas, and every write pays a majority round trip. An eventually consistent store absorbs the same load by adding nodes, with no per-write coordination. Meanwhile the same service's shard map changes a few times a day; putting that in etcd costs nothing measurable.
The alternative that often wins for metadata: a relational database
Many teams reach for etcd or ZooKeeper for metadata that fits comfortably in a relational database such as Postgres with a synchronous standby (a replica that must confirm it has durably stored a write before the primary reports success, so no acknowledged write ever exists on only the primary) and managed failover (an automated system that detects the primary is dead and promotes the standby, without a human doing it by hand). That choice is often better when:
- The metadata is relational and queried in rich ways (joins, filters, reporting), which a key-value store makes painful.
- It needs multi-row transactions (move a tenant: update its shard, quota and routing rows atomically).
- Failover in tens of seconds is acceptable, and you already run and back up the database.
Note that the database's own automated failover usually relies on a consensus-backed arbiter underneath (a managed service's control plane, or a tool like Patroni using etcd, a small, strongly-consistent key-value coordination store built on Raft, or Consul or ZooKeeper to pick the primary), so you are still using consensus, just not operating it yourself.
A dedicated consensus store still wins when you need:
- Watches: clients notified within milliseconds when a key changes, rather than polling.
- Leases and ephemeral keys: a key that disappears automatically when its owner stops renewing, which is the core of leader election and service membership.
- Fast failover (about a second with etcd's default 1000 ms election timeout) and independence from the database whose failover you might be coordinating.
Decision rule I would give a team
- Would a stale read or a merged conflict cause a correctness failure (double allocation, two leaders, lost money)? If no, prefer eventual consistency.
- If yes, is the data small and coordination-shaped (leases, membership, watches)? Use etcd, ZooKeeper or Consul (HashiCorp's service-discovery and coordination store, also Raft-based).
- If yes and it is relational or transactional, a relational database with synchronous replication is usually the simpler answer.
- If yes and it is large and high-volume (a ledger, inventory), use a database with consensus built in and sharded (distributed SQL databases such as Google Spanner, CockroachDB, TiDB or YugabyteDB), not a coordination store.
What would flip the recommendation: a hard requirement to keep writing in a region cut off from the others (only eventual consistency can do that), or a need for strict ordering on high-volume data (then sharded consensus, accepting the latency).
Trade-offs and pitfalls
- Putting user data in etcd or ZooKeeper because it is already there: the size limits and single-leader write path will bite.
- Using eventual consistency for leadership or locks: an asynchronously replicated lock can be held by two clients after a failover.
- Assuming consensus means exactly-once side effects: a deposed leader can still act on stale belief unless downstream systems check a fencing token (a number that increases with each new leader, used to reject the old one).
- Mixing both without saying which data is which: document, per dataset, which consistency level applies and why.
Tell me about a time you had to decide whether to adopt a consensus-backed component (e.g., etcd) or a simpler eventual-consistency store for a project. Describe the decision process, stakeholders involved, risks considered, and the outcome. Use the STAR structure (Situation, Task, Action, Result).
Sample Answer
Direct answer
A strong answer shows a decision driven by the data's correctness requirements, not by which technology is fashionable. Interviewers want to hear that you identified what would break under eventual consistency, quantified the cost of consensus (latency, operations, on-call load), brought the people who would run it into the decision, and could say afterwards whether it was right. Below is a realistic story skeleton to adapt with your own details; the numbers are illustrative placeholders showing the kind of evidence to cite, not results to claim.
Two terms the story relies on:
- Consensus-backed component: a store like etcd (a key-value store where a group of nodes agree on every write using the Raft algorithm: the specific consensus protocol behind that agreement, where nodes elect a leader and a write only counts once a majority has stored it, so all clients see one agreed order and no acknowledged write is lost while a majority survives).
- Eventually consistent store: one where replicas accept writes independently and converge later; reads can briefly be stale and concurrent writes can conflict (for example DynamoDB with default reads, Cassandra at low consistency levels, or Redis with asynchronous replicas).
Sample story (STAR)
Situation
Our platform team ran a job scheduler across three instances for resilience. Each instance picked up scheduled jobs (billing exports, partner file drops) from a shared table. We had two incidents in a quarter where the same export ran twice: after a network blip, two instances each believed the other was down. Duplicate billing exports reached a partner and had to be retracted manually.
Task
I was asked to make "exactly one active scheduler" reliable. The question the team split on: adopt etcd for leader election, or keep using our existing Redis (asynchronously replicated) with a lock key and a timeout, which was cheaper and already operated.
Action
- Wrote down the failure we could not accept. Two schedulers running the same job was a correctness failure with external consequences. A short pause (seconds) in scheduling was acceptable. That asymmetry, correctness over availability, pointed toward consensus.
- Tested the simpler option against that failure. I showed, with a failover test in staging, that a Redis lock acquired on the primary can be lost if the primary fails before the key replicates, so a second instance can acquire the "same" lock. I also showed the deeper problem any lock has: an instance paused in garbage collection (the runtime's automatic memory cleanup, which for that instant stops every one of the process's own threads, application code included) past its lock timeout wakes up still believing it holds the lock.
- Designed the etcd option end to end, including the part etcd does not solve. Leadership through an etcd lease (a time-limited grant that expires unless renewed) with the election API (etcd's client-library helper that lets a process campaign for a key and get notified when it wins or loses), plus a fencing token: the revision number etcd assigned to the winning leader key (it only ever increases) was written with each job run, and the export service rejected any write carrying an older token. Without fencing, even etcd would not have stopped the paused-instance case.
- Brought in the stakeholders who would carry the cost. SRE owned on-call and asked who would run a 3-node etcd cluster; we agreed that reusing the etcd behind our Kubernetes control plane for application data was not acceptable, so we budgeted a small dedicated cluster with backups and an upgrade runbook. The product owner of the partner integration agreed that a scheduling pause of a few seconds during failover was fine. Security reviewed client authentication with mutual TLS (mTLS, where both client and server present certificates).
- Wrote the risks down with owners: etcd quorum loss (losing enough of the cluster's nodes that no majority remains, so etcd itself stops accepting writes; mitigated by three availability zones), operational unfamiliarity (runbook plus a game day: a scheduled rehearsal where the team deliberately breaks the system in a controlled way to prove the failure and recovery behave as designed), and a rollback path (feature flag, a runtime switch that turns a code path on or off without a redeploy, to fall back to the old scheduler with manual approval).
Result
We shipped behind a flag, ran a game day that killed and paused the active scheduler repeatedly, and saw failover complete within the configured lease time with no duplicate runs in the export service's log. Over the following quarters there were no further duplicate exports. The honest cost: one more stateful system for SRE, and a runbook we had to exercise. If I did it again I would push the fencing-token check into the downstream service first, because that alone removed most of the risk and is cheap.
Why this structure works
| Element | What it demonstrates |
|---|---|
| Naming the unacceptable failure first | You reason from requirements, not tools |
| Testing the cheap option honestly | You did not pick consensus by reflex |
| Fencing token | You know consensus elects a leader but does not stop a stale one from acting |
| Stakeholders by cost they carry | Seniority: SRE, product and security each had a real stake |
| A result with its costs | Credibility; interviewers distrust stories with no downside |
The opposite outcome is just as strong
A story where you chose the eventually consistent store can be better, if the reasoning is visible. For example: a team wanted etcd for user preference flags, you showed that a stale read for a few seconds was harmless, conflicts could be resolved by last-writer-wins (when two writes conflict, keep whichever carries the later timestamp and silently drop the other), and the write volume was far beyond what a small etcd cluster is designed for (etcd's default storage quota is 2 GiB), so you kept the data in the existing replicated database. The skill being tested is matching consistency to consequence.
Pitfalls
- Presenting the decision as "etcd is better" with no failure scenario behind it.
- Fabricated precision ("cut incidents by 87.3%"). Use counts you can defend, or describe the outcome qualitatively.
- Forgetting the operating cost. A consensus cluster is a stateful system with quorum, backup and upgrade concerns.
- Claiming consensus alone guaranteed exactly-once behaviour (each operation's effect happens precisely one time, never zero and never more than one, even across retries and failovers); without fencing or idempotent downstream writes (writes safe to apply more than once because a repeat produces the same end state, not a duplicate effect) it does not.
Why does a distributed database need leader election? Describe common algorithms and systems used for it (Raft, Paxos, ZooKeeper, etcd), how leader election affects availability, and what an architect must consider during network partitions and leader churn.
Sample Answer
Direct answer
A replicated database keeps several copies of each piece of data, and those copies must agree on the order of writes. The simplest way to get one order is to let exactly one node, the leader, accept writes and tell the others (the followers) what to apply. Leader election is the protocol that picks that node and replaces it when it fails, and it must never let two nodes both believe they are leader at the same time. Raft and Paxos are the algorithms; etcd and ZooKeeper are systems that package them. The price is a short write outage every time the leader dies or gets cut off, plus the rule that only the side of a partition holding a majority may keep working.
Why a leader at all
Without a single writer, two replicas can accept conflicting writes to the same row at the same moment ("set balance to 80" on one node, "set balance to 50" on another), and the cluster ends up with copies that disagree. Resolving that afterwards is either impossible (money) or needs merge rules the application must supply. A leader serialises writes: it assigns each write a position in a log (entry 41, 42, 43...), and every follower applies the same log in the same order, so every copy ends in the same state. This pattern is called state machine replication.
Two terms appear in every leader-election protocol:
- Quorum (majority): more than half the voting members, for example 3 of 5. Any two majorities of the same group share at least one member, and that shared member is what stops two leaders from being elected for the same period.
- Term (Raft) / epoch (ZooKeeper) / ballot (Paxos): a number that increases with every election. Messages carry it, and a node that sees a higher number steps down. It is how the cluster tells the current leader from an old one.
The common algorithms and systems
| Name | What it is | How it elects a leader |
|---|---|---|
| Raft | A consensus algorithm (consensus: getting a group of nodes to agree on one value, or one sequence of values, even if some fail) designed to be easy to understand | A follower that hears no heartbeat (a small periodic message a leader sends just to prove it is still alive, carrying no new log entries of its own) for a randomised election timeout increments the term, votes for itself and asks the others for votes. It wins with a majority. Voters refuse a candidate whose log is less up to date than their own, so a new leader always holds every committed write. |
| Paxos (Multi-Paxos) | The older, more general consensus algorithm | A node runs a "prepare" round with a higher ballot number; once a majority promises to ignore lower ballots it acts as the stable leader and skips that round for later writes. Leader election is not specified as a separate piece, which is why real implementations differ a lot. |
| ZooKeeper | A coordination service (a small, strongly consistent key-value store for configuration, locks and membership) | Internally it uses Zab (ZooKeeper Atomic Broadcast), a Paxos-like protocol with its own leader election and epochs. Applications use it for their own elections by creating ephemeral sequential nodes (a znode, ZooKeeper's basic object, like a file in a tiny filesystem; "ephemeral" means it is deleted automatically when the client that created it disconnects, "sequential" means ZooKeeper appends an ever-increasing number to its name): the lowest sequence number is leader, and the node disappears when that client's session (its live connection to ZooKeeper) dies. |
| etcd | A coordination store built on Raft (the store behind Kubernetes) | Its own members elect a Raft leader. Applications elect their leader with a key attached to a lease (a time-limited grant that expires unless the holder keeps renewing it); the client library offers an election API that does this for you. |
Note the two layers. A database such as CockroachDB or TiKV (distributed SQL/key-value databases that split their data into many small pieces) embeds Raft and elects a leader per data range (one such piece: a contiguous slice of the keyspace, replicated and led independently of the other pieces). A service that is not itself replicated (say, one active scheduler among three instances) borrows an election from etcd or ZooKeeper instead of implementing consensus.
How leader election affects availability
While there is no leader, the cluster cannot accept writes, and strongly consistent reads (reads guaranteed to reflect every write acknowledged before them, rather than possibly returning slightly old data) also stall. The outage is roughly:
failure detection time + election time + client retry time.
Worked example with etcd's documented defaults (heartbeat 100 ms, election timeout 1000 ms): a leader crashes. Followers wait up to about 1 s without a heartbeat, one starts an election, and with a network round trip of a few milliseconds it wins almost immediately. Clients that were mid-request see an error or timeout and retry. So a leader crash costs on the order of one to two seconds of write unavailability, not zero. A 5-member cluster tolerates 2 simultaneous failures (3 of 5 is still a majority); a 3-member cluster tolerates 1.
Reads that can accept slightly old data may be served by any follower during the gap, which is a deliberate trade of freshness for availability.
What an architect must consider
Network partitions. Suppose 5 nodes split into groups of 3 and 2. Only the group of 3 can elect a leader and commit writes; the group of 2 must refuse writes. If the old leader was in the group of 2, it keeps thinking it is leader until it notices it cannot reach a majority. That is the split brain risk (two nodes both acting as leader). Raft prevents two committed histories because the minority leader cannot get majority acknowledgements, but the stale leader can still serve stale reads or keep doing side effects outside the database unless:
- reads go through the leader with a majority check or a leader lease (a period during which the leader knows no one else can be elected), and
- external side effects carry a fencing token (the term number, checked by the receiving system so it rejects anything from an older term).
Placement. Put members in at least three failure domains (availability zones or regions). With two zones, losing the zone holding the majority takes the cluster down.
Leader churn. Frequent re-elections cost more than one clean failover, because each election is a write outage. Common causes are an election timeout set too close to the real network round trip, slow disks (a leader stuck in a disk sync misses heartbeats), long garbage-collection pauses and overloaded CPUs. The fixes are tuning the timeout to at least 10 times the round trip time (etcd's own guidance), fast dedicated disks, and alerting on the rate of leader changes rather than just on "is there a leader".
Cluster size. Use an odd number of voters, usually 3 or 5. Going from 5 to 7 tolerates one more failure but every write now waits for 4 acknowledgements instead of 3, so writes get slower.
Trade-offs and pitfalls
- Consensus is not free availability. It trades availability during partitions for correctness. If the data can tolerate conflicts (a "likes" counter, a shopping cart), a leaderless eventually consistent design (every replica can accept writes independently, and copies converge to the same value over time rather than agreeing instantly) may be the better fit.
- Rolling your own election with a heartbeat and a timeout (whoever has not heard from the leader for 5 s takes over) produces two leaders the first time a network is slow rather than dead. Borrow a proven implementation.
- Believing the process that holds leadership is the only one acting. A leader paused by garbage collection can wake up and act after it has been replaced. Fencing tokens, not timeouts, make that safe.
- Watching only "leader exists". A cluster that re-elects every few minutes looks healthy on that metric while clients see repeated error spikes.
Given N replicas, explain how read quorum size R and write quorum size W interact to provide strong consistency. Provide the general condition and walk through a concrete example for N=5: list at least two (R, W) pairs that guarantee linearizability and show why the intersection property holds.
Sample Answer
Short answer
With N replicas, a write is sent to at least W of them and a read asks at least R of them. If
R+W>Nthen every read quorum must share at least one replica with every write quorum, so a read always reaches at least one replica that holds the latest completed write. For N = 5, pairs such as (R, W) = (3, 3) and (2, 4) satisfy this. Overlap is the necessary condition; turning it into full linearizability (every read returns the most recent completed write, as if there were one copy) also needs version numbers on values, a read that picks the highest version and writes it back to a quorum before returning, and, with more than one writer, write quorums that overlap each other as well.
Why the overlap holds
A quorum is simply "enough replicas that any two such groups must overlap". The argument is the pigeonhole principle: a read set A of size R and a write set B of size W are both chosen from only N replicas, so if their sizes add up to more than N they cannot be disjoint.
∣A∣+∣B∣>N⟹A∩B=∅The shared replica has seen the write, so the reader sees it too, as long as it can tell which of the values it received is newest. That is why each value carries a version (a timestamp or counter) and the reader takes the highest.
A second condition matters when several clients can write the same key. Two write quorums must overlap each other, otherwise two concurrent writes can land on disjoint replica sets and nobody can order them:
2W>NWorked example: N = 5
Here is a brute-force check over every possible subset of 5 replicas, confirming both formulas and listing every pair that satisfies read/write overlap.
from itertools import combinations
N = 5
nodes = range(N)
def always_intersect(a, b):
"""True if EVERY a-sized subset of nodes shares a node with EVERY b-sized subset."""
return all(set(x) & set(y) for x in combinations(nodes, a) for y in combinations(nodes, b))
print(" R W read-write overlap write-write overlap")
for R in range(1, N + 1):
for W in range(1, N + 1):
rw, ww = always_intersect(R, W), always_intersect(W, W)
assert rw == (R + W > N) and ww == (2 * W > N) # brute force agrees with the formulas
if rw:
print(f"{R:2} {W:2} {str(rw):18} {ww}")
# Why R + W = N is not enough: a write quorum and a read quorum that miss each other.
write_quorum, read_quorum = {0, 1, 2}, {3, 4} # W = 3, R = 2, R + W = 5 = N
print("R=2, W=3 overlap:", write_quorum & read_quorum or "none")
Output:
R W read-write overlap write-write overlap
1 5 True True
2 4 True True
2 5 True True
3 3 True True
3 4 True True
3 5 True True
4 2 True False
4 3 True True
4 4 True True
4 5 True True
5 1 True False
5 2 True False
5 3 True True
5 4 True True
5 5 True True
R=2, W=3 overlap: none
Three pairs worth discussing, all satisfying both overlap conditions:
| (R, W) | Read cost | Write cost | Replicas that can be down |
|---|---|---|---|
| (3, 3) | 3 replies | 3 acks | 2 for reads, 2 for writes. The balanced default. |
| (2, 4) | 2 replies | 4 acks | 3 for reads, only 1 for writes. Read-heavy workloads. |
| (1, 5) | 1 reply | all 5 acks | Reads survive 4 failures, but one dead replica blocks every write. |
Walk-through for (3, 3), using the same five replicas {0, 1, 2, 3, 4} as the code above: a write to key balance with version 7 is acknowledged by replicas {0, 1, 2}. A later read contacts {2, 3, 4}. Replicas 3 and 4 still hold version 6, replica 2 returns version 7. The reader picks 7 because it is highest. Any three-replica read must include at least one of {0, 1, 2}, because only two replicas (3 and 4) are outside the write set.
The pairs (4, 2), (5, 1) and (5, 2) give read/write overlap but not write/write overlap, so they only suit a design with a single writer per key (for example, a leader that serializes writes). The last line of the output shows the failure case: with R = 2 and W = 3, R + W equals N, and the read {3, 4} can miss the write {0, 1, 2} entirely.
From "sees the latest write" to linearizable
Overlap alone gives "a read after a completed write sees it". Linearizability is stricter, and two gaps remain:
- Writes in progress. Suppose a write of version 8 has reached only replica 1 when reader X queries {1, 4, 5} and returns 8, and then reader Y queries {3, 4, 5} and returns 7. Y started after X finished yet saw an older value, which violates linearizability. The fix (from the ABD algorithm, named after Attiya, Bar-Noy and Dolev, a classic quorum-based register protocol) is a read write-back: before returning version 8, X writes it to a quorum, so any later read is guaranteed to see it.
- Ordering concurrent writes. Each writer first reads the current highest version from a quorum and writes version + 1 tagged with its ID for tie-breaking. This is safe because of the write-write overlap condition above (
2W > N): any two write quorums share at least one replica, so a writer computing "current highest version" always reads a quorum that includes someone who saw every prior write, and two writers racing to use the same version number cannot both complete without a shared replica noticing the collision.
Pitfalls
- Sloppy quorums break the guarantee. Some Dynamo-style stores (leaderless, quorum-based key-value stores modeled on Amazon's Dynamo design, for example Cassandra or Riak) use a sloppy quorum: when a home replica is down, the coordinator accepts the write on a stand-in node outside the key's proper N replicas instead of blocking, and later moves that data to its rightful home once it recovers, a process called hinted handoff. The arithmetic R + W > N then no longer implies overlap, because the write quorum was not drawn from the same N.
- Counting acks from replicas that did not persist. An ack must mean the write is durable on that replica, otherwise a crash-restart silently removes it from the quorum.
- Choosing (1, N) for "fast reads" looks attractive until one replica is down and every write stalls.
- Last-writer-wins by wall-clock timestamp as the version can silently drop a write when clocks disagree; prefer logical versions (a counter or vector clock that orders writes by cause and effect, rather than relying on wall-clock time that can disagree across machines).
Unlock Full Question Bank
Get access to all 16 Consensus and Coordination Algorithms interview questions and detailed answers.
Sign in to ContinueJoin thousands of developers preparing for their dream job.