Distributed Systems and Consensus Algorithms Questions
Covers the fundamentals and practical application of distributed systems and the algorithms that produce agreement across nodes. Topics include consistency models such as strong consistency, eventual consistency, and causal consistency; replication strategies including master slave and peer to peer; and distributed transactions. Explain consensus algorithms and their trade offs, for example Raft, Paxos, and Practical Byzantine Fault Tolerance, including safety and liveness properties, quorum based decision making, leader election patterns, leader based versus leaderless designs, log replication, heartbeats, and lease based leadership. Understand the theorem that trades off consistency, availability, and partition tolerance and its implications for system design. Be able to reason about failure modes such as network partitions, split brain, and Byzantine faults and how algorithms and system choices mitigate them. Include familiarity with real world systems that implement these concepts such as Etcd, Consul, ZooKeeper, DynamoDB, and HBase. At senior and staff levels, demonstrate when consensus is required versus when it can be avoided, the operational complexity and performance costs of consensus, practical engineering decisions for scaling and fault tolerance, debugging and testing strategies, and how these choices affect application behavior.
MediumTechnical
37 practiced
For classic Paxos, describe the message complexity (number of messages and rounds) required for a single simulated consensus decision (single value) in the typical leader-based optimization (multi-Paxos) and compare it to Raft's AppendEntries for a batched set of commands. Discuss the effect of batching on throughput and latency.
Sample Answer
Short answer: With a stable leader multi-Paxos reduces to one replication round per proposal (like Raft). Message counts and rounds are similar; batching amortizes per-command overhead and boosts throughput at the cost of added queuing latency.Details and concrete counts (N = number of nodes, majority M = floor(N/2)+1):Multi-Paxos (leader established; skip Phase 1):- Rounds: client→leader (optional) then one leader→replicas→leader round-trip to get majority acks — effectively 1 replication round-trip.- Messages (simple accounting for one command): - client→leader: 1 - leader→N-1 followers (Phase 2a / Accept or Append): N-1 - followers→leader acks: at least M-1 replies needed (but followers usually reply): ~N-1 - leader→client confirmation: 1 => total ≈ 2(N-1)+2 ≈ 2N messages (in practice many systems count only leader↔followers: ~2(N-1)).Raft AppendEntries (single command, no batching):- Rounds: same pattern — client→leader then leader sends AppendEntries; followers reply — 1 replication round-trip.- Messages: - client→leader: 1 - leader→N-1 followers: N-1 AppendEntries - followers→leader: N-1 replies (majority needed) - leader→client: 1 => total ≈ 2(N-1)+2, effectively same as multi-Paxos under a stable leader.Effect of batching:- Throughput: If the leader batches B commands into one AppendEntries / Accept, the per-command network messages drop roughly by factor B (messages per command ≈ (2(N-1)+2)/B). So throughput improves nearly linearly with batch size until other bottlenecks (disk, CPU, network MTU) dominate.- Latency: End-to-end latency per command becomes: - Service latency = queuing time to form the batch + replication RTT + commit processing. - Larger B increases queuing delay (increases average and tail latency) because commands wait to fill batch. If leader flushes periodically (time-based batching), you trade latency for throughput.- Practical optimizations: pipelining (send more AppendEntries without waiting for replies) and batching together can give high throughput with moderate latency. Also batching interacts with durable persistence (fsync): batching reduces expensive fsyncs per command.Takeaway: Under a stable leader, multi-Paxos and Raft have essentially the same message complexity and rounds per consensus decision. Batching greatly increases throughput by amortizing the replication cost across many commands, but it increases per-command latency due to batching delay; use time-or-size-based flush policies and pipelining to balance the trade-off.
MediumSystem Design
37 practiced
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
Requirements & constraints:- Allow low-latency reads from followers with bounded staleness.- Provide an API that lets clients request fresher data when needed.- Preserve monotonic reads and session guarantees for clients.High-level design:- Every write is assigned a monotonically increasing commit timestamp (use hybrid logical clocks or Lamport + physical time).- Followers asynchronously replicate committed writes and expose read replicas.- Clients maintain a session state containing last_seen_ts (highest commit_ts observed).API semantics:- GET(key, options) where options include: - max_staleness_ms (optional): acceptable age of data - min_commit_ts (optional): require data at least as fresh as this timestamp - consistency = {follower_read | leader_read} (default follower_read if staleness bound satisfied)- Server response includes value, commit_ts, and source (follower/leader). If follower cannot satisfy min_commit_ts or max_staleness_ms, it returns TRY_LEADER or automatically forwards to leader.Staleness bounds:- Followers track last_applied_commit_ts per shard. For time-based bound, followers convert their last_applied_commit_ts to age using local clock vs commit_ts epoch and can accept reads where (now - commit_ts) <= max_staleness_ms.- For version-based bound, followers ensure last_applied_commit_ts >= min_commit_ts.Monotonic reads & session guarantees:- Clients attach their session.last_seen_ts on each read (or server stores sessions). On a read, follower will only serve values with commit_ts >= session.last_seen_ts unless client explicitly requests weaker guarantee.- After receiving a response with commit_ts R, client updates session.last_seen_ts = max(session.last_seen_ts, R).- For writes, leader assigns commit_ts > session.last_seen_ts; clients can include session.last_seen_ts with write to ensure serialization after prior reads.Ensuring correctness & freshness:- Followers periodically apply replication and expose last_applied_commit_ts. To avoid serving too-stale data due to clock skew, use HLC to compare timestamps.- If follower can't meet min_commit_ts, it rejects (TRY_LEADER) or proxies to leader which has the true latest commit_ts.- To avoid stale follower serving after failover, use leader leases: leader increments epoch on election; followers refuse to serve follower_reads beyond lease window without confirming replication progress.Operational notes and trade-offs:- Pros: low-latency reads, explicit client control over freshness.- Cons: complexity (HLC, session tracking), slightly higher tail latency if client asks for leader reads.- Alternatives: synchronous cross-region replication for stronger freshness (higher write latency) or bounded staleness by fixed replica lag monitoring.Example client flow:1. Client reads with max_staleness_ms=50ms; follower checks last_applied_commit_ts and local time; if satisfied returns value+commit_ts; client updates last_seen_ts.2. Later client issues a read requiring min_commit_ts = session.last_seen_ts+1 to ensure monotonicity; follower either serves (if applied) or returns TRY_LEADER.This design gives clients explicit control of freshness, bounded staleness guarantees, and enforces monotonic reads via session timestamps while keeping read latency low when possible.
HardTechnical
34 practiced
Design a chaos engineering test suite for a consensus cluster (e.g., a 5-node Raft cluster). Include fault injection scenarios (network partitions, message loss, delayed messages, disk stalls, process pauses), metrics to capture, automated checks to detect correctness regressions, and acceptance criteria for cluster resilience.
Sample Answer
Requirements & goals:- Validate safety (no split-brain / diverging logs), liveness (progress under faults), and durability (committed entries persist).- Target: 5-node Raft cluster, reproducible, automatable, CI-friendly, parameterizable fault intensity.Test-suite structure:- Orchestrator: test runner (Python/Go) that controls cluster nodes (containers/k8s), fault injector (tc/netem, iptables, cgroups, SIGSTOP), and telemetry collector (Prometheus + logs).Fault-injection scenarios (each run parameterized by duration, target nodes, severity, seed):1. Leader network partition: isolate leader from 1–3 followers; expect election or graceful takeover.2. Minority partition: isolate 2 nodes so majority still reachable; ensure progress continues.3. Split-brain attempt: partition into 2 vs 3 and 2 vs 2 transient; verify a single committed leader and no dual commit.4. Packet loss & corruption: apply 10–50% UDP/TCP loss or reorder to randomized pairs.5. Message delay/jitter: add 100–2000ms delay to simulate WAN.6. Disk stalls: pause /dev/sdX I/O for 500ms–5s on follower/leader.7. Process pause/resume: SIGSTOP for 1–30s to leader or followers.8. Crash and reboot: kill and restart nodes with state intact and with wiped WAL for recovery test.9. Minority slow join: bring up a node with lagging logs; test snapshot/install behavior.Metrics to capture:- Raft state (leader ID, term, commit index, lastApplied) per node.- RPC success/failure counts and latencies.- Election frequency and durations.- Log divergence: compare lastLogIndex/lastLogTerm across nodes.- Client-side observed throughput and latency.- Error rates, retransmissions.- System metrics: CPU, IO wait, network stats.Automated checks (post-scenario):- Safety assertion: at any time, there is ≤1 leader for a given term (verify leader ID per term).- Log convergence: after recovery window, all non-faulty nodes have identical committed log prefix up to commitIndex.- Progress: client writes achieve N successful commits within T after transient faults when majority available.- No lost committed entries: data written and acknowledged as committed must appear on all nodes after stabilization.- Latency/throughput regression: compare to baseline (±X% threshold).Acceptance criteria:- Under any single-fault scenario (network partition isolating ≤2 nodes, single process pause, transient disk stall ≤2s): system maintains safety; within S1 (e.g., 30s) after fault removal, cluster re-elects leader (if needed) and achieves ≥90% of baseline throughput.- Under combined or prolonged faults: no safety violations; eventual recovery within S2 (e.g., 2 minutes) and no more than 1% of committed entries lost or divergent.- CI threshold: deterministic fuzzing seed failures flagged; flaky runs above 5% rate require investigation.Example test flow (Leader network partition):- Start cluster; write baseline 1000 entries.- Identify leader L.- Inject iptables DROP between L and followers F1,F2 for 15s.- Continue client writes.- Collect metrics; after 15s, remove partition.- Assertions: no term with >1 leader; all committed entries visible on all nodes after 60s; commitIndex progressed during majority availability.Best practices:- Make tests idempotent and seedable; log full event timeline; record network captures for failures.- Run in emulated and real cloud environments.- Gradually increase fault intensity and add throttling to detect brittle components.- Fail-fast on safety assertions; use blameless postmortem artifacts for reproducibility.
EasyTechnical
33 practiced
When would you choose optimistic concurrency control (e.g., CAS / version compare) versus pessimistic locking in a distributed service? Give examples of workloads and trade-offs including throughput, latency, user experience, and complexity.
Sample Answer
Use optimistic concurrency (CAS/version compare) when conflicts are rare and you want higher throughput and lower latency; use pessimistic locking when conflicts are frequent or correctness requires serial access.Optimistic (CAS/version):- When: read-heavy workloads with occasional writes (e.g., user profile updates, product view counters, shopping cart merges).- Benefits: low latency for reads/writes (no blocking), simple horizontal scale, higher throughput under low contention.- Drawbacks: failed updates require retry logic, potential wasted work, more complex client-side retry/backoff and idempotency handling.- UX: fast for most users; occasional retries may cause slight delay or require user-visible conflict resolution.Pessimistic locking:- When: write-heavy or high-contention operations needing strong serializability (e.g., bank account transfers, inventory reservation during checkout, workflow state transitions).- Benefits: predictable correctness and linearizability, simpler server-side reasoning about invariants.- Drawbacks: higher latency (waits/queues), lower throughput under contention, deadlock risk, need for lock lease/renewal in distributed systems.- UX: consistent behavior but slower responses; timeouts must be handled gracefully.Trade-offs summary:- Throughput: optimistic wins under low contention; pessimistic wins when avoiding repeated retries under extreme contention.- Latency: optimistic typically lower; pessimistic can add blocking delays.- Complexity: optimistic pushes complexity to retry/idempotency; pessimistic requires lock management, deadlock handling, lease mechanics.Choose based on conflict frequency, tolerance for retries, need for strict serializability, and user experience expectations.
MediumTechnical
26 practiced
You see an etcd cluster with repeated leader changes every few seconds, causing client errors. Given access to metrics (election timeout, RPC latencies, disk stalls, GC pauses) and logs, outline a debugging plan to identify root cause(s) and propose immediate mitigations to stabilize the cluster.
Sample Answer
Situation: An etcd cluster is experiencing leader churn every few seconds, causing client errors. Goal: find root cause quickly using metrics/logs and stabilize the cluster with minimal disruption.Debugging plan (steps, tools, rationale)1. Confirm the symptom & scope- Run etcdctl endpoint status --write-out=table and etcdctl endpoint health to see which members report leader, lag, and health.- Check clients’ error patterns and timestamps to correlate with elections.2. Collect and correlate metrics (timeseries + logs)- Pull /metrics from each member or view Prometheus graphs for: - election_timeout_seconds and raft leader changes (counter) - raft_rpc_durations & raft_proposals_latency - disk IO metrics (disk stall time), fsync latencies - GC pause durations (Go runtime: go_memstats_last_gc_pause_ns, gc_pause_seconds) - CPU, memory, network retransmits- Grep logs (journalctl /var/log/etcd) for "election timeout", "lost leader", "failed to send", "disk stuck", or "heartbeat timeout".Rationale: Elections are triggered by missed heartbeats — caused by RPC latency, disk stalls, GC pauses, network packet loss, or overloaded CPU.3. Triangulate root cause patterns- If RPC latencies spike across nodes at same times → network (MTU, partition, firewall) or load spike.- If leader-only logs show long disk/fsync latency or kernel OOM/IO errors → disk stalls.- If GC pauses > election timeout (default ~1s) on any node → GC-induced pauses causing missed heartbeats.- If only one node shows missing heartbeats and rest healthy → that node flaps; consider hardware or cloud host issues.4. Quick checks on impacted hosts- top/htop, iostat -x 1, sar -q, vmstat 1, dmesg for I/O or NIC errors, tc qdisc, ifconfig for drops.- Check container limits (cgroups): cpu throttling, memory swap, disk quotas.- Verify clock skew (chrony/ntp) — large skew can cause weird behavior.Immediate mitigations to stabilize cluster (short-term, low-risk)1. Increase election timeout temporarily- Edit etcd static unit or start option --election-timeout to 5000ms (or 5x current) and rollout one by one. This reduces sensitivity to transient pauses.Rationale: Prevents elections from firing for short GC/disk spikes.2. Address GC/disk transient issues- If GC is culprit: set GOGC or upgrade Go/etcd; reduce heap growth (temporary: increase election timeout).- If disk stalls: move etcd to faster disks, disable noisy background jobs, reduce I/O contention, or move to separate volume.3. Remove offending node from quorum (if single bad node)- Use etcdctl member remove <id> to isolate a node causing churn, then investigate offline.Rationale: Restores cluster stability quickly if a single node is flapping.4. Throttle client load- If bursts trigger churn, throttle or batch writes, enable client retries/backoff.5. Network fixes- If network packet loss, adjust MTU, fix routing, or place leaders on nodes with stable NICs.Follow-up actions (medium-term)- Tune election timeout based on observed GC/disk latencies and SLA.- Improve observability: alert on leader changes, raft_rpc latencies, GC pause > election_timeout/2.- Capacity plan: ensure CPU, memory, and IOPS headroom; move etcd off noisy neighbors.- Test restart/upgrade strategy and run chaos tests to validate settings.Example quick commands- etcdctl endpoint status --write-out=table- curl http://<node>:2379/metrics | grep raft- journalctl -u etcd -o cat --since "1 hour ago" | grep -E "election|heartbeat|stalled|fsync|panic"- iostat -x 1 5; dmesg | tail -n 200This approach narrows cause by correlating election events with RPC/disk/GC/network signals, applies low-risk mitigations (increase election timeout, isolate bad node), then implements root-cause fixes and observability to prevent recurrence.
Unlock Full Question Bank
Get access to hundreds of Distributed Systems and Consensus Algorithms interview questions and detailed answers.
