Distributed Systems Fundamentals Questions

Direct answer

These three terms describe how many times a record's effect can show up downstream, not how many times it crosses the wire. At-least-once guarantees nothing is silently dropped but tolerates re-delivery, so duplicates are possible. At-most-once guarantees no duplicates but tolerates silent loss. Exactly-once means the record's effect appears exactly once, even though delivery itself is usually still at-least-once under the hood; the "exactly" part is enforced by deduplication or a transactional write, not by literally never redelivering anything.

At-least-once

Mechanism: the consumer commits its read offset only after it has finished processing a record.

Scenario producing a duplicate: a consumer reads offset 100 from a partition and processes it, say incrementing an inventory counter, but crashes before committing that offset. On restart it resumes from the last committed offset, 99, and re-reads and reprocesses offset 100, incrementing the counter a second time. The counter's effect happened twice for one logical record.

At-most-once

Mechanism: the consumer commits its read offset before processing the record.

Scenario producing a loss: a consumer commits offset 100 immediately on receipt, then crashes while processing that record. On restart it resumes from offset 100 onward, since that offset is already committed, so the record at offset 100 is never processed. This is the mirror image of the at-least-once bug: it's the same commit-versus-process ordering, flipped.

Exactly-once: what it actually takes at the consumer

• Idempotent processing: design the effect so applying it twice produces the same result as applying it once, for example replacing "increment counter by 1" with "set counter to max(current, computed value)", or recording each processed key in a table with a uniqueness constraint so a repeat attempt is rejected rather than reapplied.
• Dedup window: a bounded structure, such as a set of recently-processed record ids, that the consumer checks before applying an effect. It must be sized to cover the maximum plausible redelivery delay; if consumer restarts are the main cause of redelivery and offsets commit every 30 seconds, a window covering the last few minutes of ids is enough, but an outage longer than the window falls back on whatever uniqueness constraint the underlying store provides.
• Transactional writes: commit the output write and the offset advance as a single atomic operation, so a crash between "wrote output" and "committed offset" cannot happen. Kafka's transactional producer does this across topic-partitions; a database sink can do the same by writing the output row and a processed-offsets row inside one database transaction.

Worked example: exactly-once via a transactional write

Consumer is at offset 100. It opens a transaction, upserts the inventory row (an idempotent write), writes offset=101 to an offsets table, and commits the transaction atomically.

• If the consumer crashes before the commit: nothing happened; the output was never applied and the offset was never advanced. Re-reading offset 100 next time reproduces the exact same atomic attempt, with no partial effect ever visible in between.
• If it crashes after the commit: offset 101 is already recorded, so the consumer will not re-read offset 100 on restart.

There is no window in which the output exists but the offset doesn't, or vice versa, because both changes are part of one transaction.

Practical guidance

Trade-offs & pitfalls

Exactly-once adds real coordination cost (transactions, dedup lookups), so its throughput and latency are worse than plain at-least-once, which is why it's reserved for cases where correctness genuinely requires it rather than applied everywhere by default. A dedup window sized too small silently degrades to at-least-once during a long outage, without any error being raised. A common interview trap is conflating a broker-level exactly-once guarantee (Kafka's own transactions between its topics) with true end-to-end exactly-once: Kafka's guarantee stops at the Kafka cluster boundary, so if the final effect lands somewhere outside it, an external database or a third-party API, that external system still has to be transactional or idempotent for the guarantee to actually hold all the way through.

Linearizability and serializability sound similar but answer different questions. Linearizability is about a single object: every operation on it must appear to happen instantaneously at some point between when it was invoked and when it returned, and that ordering must match real time. Serializability is about multiple objects touched by a transaction: the outcome of running several transactions concurrently must be equivalent to running them in some serial order, but that order does not have to match real time or even the order the transactions actually started in. A system can have one property without the other, and assuming the wrong one silently breaks a different class of guarantee.

Two mechanisms that actually enforce serializability

• Two-phase locking (2PL): a transaction acquires every lock it needs before releasing any of them, and once it starts releasing locks it may acquire no more. This physically prevents conflicting concurrent access, at the cost of blocking and potential deadlock.
• Optimistic concurrency control (OCC): transactions proceed without locking, then get validated at commit time; if another transaction's concurrent writes conflict with what this one read, it aborts and retries. This avoids blocking under low contention but wastes work under high contention.

When you need one but not the other

Consider a key-value store advertising single-copy semantics: every replica must behave as if there is exactly one physical copy of the data, so any client reading a key right after a write, from any client, on any replica, sees that write or a later one, never a stale value. The same requirement shows up as a highly available configuration service needing linearizable reads: if a client reads a feature flag or a routing rule right after it changed, it must get the new value, since acting on a stale one applies the wrong policy. Neither of these needs serializability: there is no multi-key transaction to isolate, just one key's recency.

The mirror case: a reporting system running multi-row aggregate queries across many tables needs those queries to see an internally consistent snapshot (serializability, or at least snapshot isolation), but does not need that snapshot to be the absolute latest possible instant in real time. A report built from data a few hundred milliseconds behind the live system is fine, as long as every row it reads is mutually consistent with every other row it reads.

Worked example: what breaks if you assume the wrong one

Linearizable but not serializable, no cross-key transaction: a key-value store gives linearizable single-key reads and writes but has no multi-key transactions. A funds transfer moves 30 units from account A (currently 100) to account B (currently 50) as two separate linearizable writes: write A=70, then write B=80. A concurrent reader can land exactly between the two writes and read A=70 and B=50. Both individual reads are linearizable, each reflects the latest write to that specific key at the moment it was read, but the reader just observed a total of 70+50=120, when the true, fully-settled total is 70+80=150: 30 units appear to have vanished mid-transfer. That is the anomaly linearizability alone does not prevent, because it says nothing about atomicity across two different keys.

Serializable but write-skew possible, snapshot isolation only: a hospital scheduling system enforces one invariant, that at least one doctor remains on call.

$\text{doctors on call} \ge 1$

Two doctors, Alice and Bob, are both currently on call, so the on-call count is 2. Both, concurrently, read a snapshot showing 2 doctors on call and each independently decide it is safe to go off-call, and both commit that decision under snapshot isolation, since neither transaction's write conflicts with what the other actually wrote (each only writes their own on-call flag). The result: 0 doctors on call, violating the invariant, even though each transaction, viewed alone against its own snapshot, looks perfectly valid. Full serializability, not just snapshot isolation, would detect that these two transactions' reads and writes interfere and force one to abort; snapshot isolation's weaker check does not.

Trade-offs & pitfalls

• Common wrong turn: treating serializable as automatically meaning fresh or linearizable. It is not: transactions can be serialized in an order that does not match when they actually ran.
• Common wrong turn: treating a single-key linearizable store as if it gives transactional safety across several keys. It does not, by itself, unless the store also offers multi-key transactions on top.
• Snapshot isolation is cheaper than full serializability, since it does not need to detect every possible interleaving, only genuine write-write conflicts, and is what most production databases default to, which is exactly why the write-skew anomaly above shows up in practice more often than people expect.

Direct Answer

In a system with N replicas, a write is only considered committed once W of those replicas have acknowledged it, and a read is only considered complete once R replicas have been queried and the freshest value among their answers is returned. If you pick W and R so that

then every possible set of W replicas and every possible set of R replicas are guaranteed to overlap in at least one replica, which means any read is guaranteed to touch at least one replica that has the most recent write.

Why the Overlap Guarantee Holds

This falls out of a simple counting fact: if you pick two subsets of a set of N items, and the sizes of those two subsets add up to more than N, they cannot be disjoint. If a write-set of size W and a read-set of size R were completely disjoint, sharing no replica at all, together they would use W + R distinct replicas out of only N available, which is impossible once W + R > N. So the two sets must share at least one replica, and since the write-set includes every replica the write reached, the shared replica is guaranteed to have seen the latest write.

Worked Example, N = 5

Take five replicas, labeled 1 through 5. Choose W = 3 and R = 3 (3 + 3 = 6 > 5, so the guarantee holds).

A write commits to replicas {1, 2, 3}, the write quorum. A later read queries replicas {3, 4, 5}, the read quorum. The overlap between {1, 2, 3} and {3, 4, 5} is {3}, so replica 3 is guaranteed to be in both sets, and since replica 3 has the latest write, the read correctly returns the fresh value even though replicas 4 and 5 are still stale.

Now see what happens if you drop below the threshold: keep W = 3 but use R = 2 (3 + 2 = 5, not greater than N = 5, so the guarantee no longer holds). A read that happens to query {4, 5} shares no replica at all with the write quorum {1, 2, 3} and would return the stale value those two replicas still hold, with no way for the client to know it missed the latest write.

Trading Off Latency, Availability, and Durability

• Lowering W speeds up writes, since fewer replicas have to acknowledge, and lets writes succeed even if more replicas are down, but it weakens durability (fewer copies exist right after the write) and forces R to be larger to keep W + R > N, which slows reads down instead.
• Lowering R speeds up reads the same way, at the cost of needing a larger W.
• To keep serving at a chosen W or R while tolerating f replica failures, you need enough surviving replicas to still form that quorum, so majority quorums, such as W = R = 3 for N = 5 (the smallest quorum size bigger than half of 5), are a common default: they satisfy W + R > N for any N, and they keep working as long as a majority of replicas are reachable.

Leaderless Quorums vs. a Leader-Based Design

Quorum systems like this are naturally leaderless: any client can attempt a write or a read against any W or R replicas without funneling through one elected coordinator, unlike a Raft-based design where every write has to go through the single current leader. That gives quorum systems more availability during a partition, since any reachable set of W or R replicas can keep working, at the cost of needing real conflict handling: two writes that each reach a different, overlapping-but-not-identical set of replicas can produce concurrent versions that a read has to reconcile, by comparing versions and taking the latest or surfacing both to the application, which a single-leader system avoids by construction since all writes are already serialized through the leader.

Choosing Sane Defaults in a Client Library

A client library that exposes N, R, and W as tunable knobs should default to a majority quorum on both sides, W = R = the smallest integer greater than N / 2, rather than exposing the raw numbers with no guidance, because majority-on-both-sides is the smallest configuration that always satisfies W + R > N regardless of N, and it gives a reasonable latency-versus-safety balance without requiring the caller to re-derive the inequality themselves. The library should still let advanced callers override it, such as R = 1 for the fastest possible read when the caller is prepared to handle occasional staleness itself, or W = N for maximum durability when the caller can tolerate slower writes, with the safety trade-off documented at each override.

Trade-offs and Pitfalls

• Quorum overlap guarantees that a read touches at least one replica with the latest write; it does not by itself guarantee the read correctly identifies which of the R responses is the latest one. Without comparing versions or timestamps correctly across the R responses, you can still return a stale value even though the fresh one was right there in the response set.
• Concurrent writes are a real gap: if two writes race and land on different, only-partially-overlapping write quorums, you can end up with genuinely concurrent versions that need reconciliation, not just staleness that time will fix.
• Picking W = 1 to maximize write availability forces R = N to keep the safety guarantee, which makes every read fragile to a single unavailable replica; it's rarely a good default outside very read-light, write-heavy workloads that can tolerate that risk.

Direct answer

A checkpoint barrier is a special marker the coordinator injects into every source stream at a chosen moment. As it flows downstream mixed in with real data, each operator uses its arrival to mark a cut: everything on that input channel before the barrier belongs to checkpoint N, everything after belongs to checkpoint N+1. Once an operator has seen the barrier on all of its input channels, it takes a local snapshot of its own state and forwards the barrier onward. Because every operator's snapshot is cut at the same logical point in the data rather than the same wall-clock instant, the union of all local snapshots plus the recorded source read-positions forms one consistent global snapshot the whole job can be rewound to after a failure.

Barrier injection and alignment

The checkpoint coordinator periodically assigns an increasing checkpoint id and injects a barrier carrying that id into every source partition. Barriers travel with the data on each channel, in order, never overtaking a record. When an operator with multiple input channels receives the barrier on one channel before the others, it stops consuming further records on that channel and buffers them, while continuing to process the channels where the barrier hasn't arrived yet. This buffering-until-all-channels-caught-up step is called alignment; it guarantees the operator's eventual snapshot reflects exactly the same cut point on every input.

Local snapshot and coordinator commit

Once aligned, the operator snapshots its local state (for a keyed aggregation, the current value per key) to durable storage and forwards the barrier to its downstream operators. The coordinator marks a checkpoint complete only once every operator, all the way to the sinks, has acknowledged it, and only then persists the checkpoint's metadata as the new restore point.

Unaligned checkpoints

Alignment can add latency under backpressure or skew, since a fast channel has to wait on a slow one before the operator can snapshot. Unaligned checkpoints avoid this by not waiting at all: the framework snapshots the buffered in-flight records themselves as part of the checkpoint, alongside the operator's own state, trading more storage and I/O for lower checkpoint latency.

Worked example: barrier flow through a keyed aggregation

Job topology: Source -> Map -> KeyedSum -> Sink, with KeyedSum running as two parallel instances, A and B.

sequenceDiagram
    participant Coord as Coordinator
    participant A as KeyedSum A
    participant B as KeyedSum B
    participant Sink
    Coord->>A: barrier(42) on P0
    Coord->>A: barrier(42) on P1
    A->>A: snapshot state
    A->>Sink: forward barrier(42)
    Coord->>B: barrier(42) on P0
    Coord->>B: barrier(42) on P1
    B->>B: snapshot state
    B->>Sink: forward barrier(42)
    Sink->>Coord: ack checkpoint 42

1. Coordinator starts checkpoint 42 and injects barrier(42) into the source's two partitions, P0 at read-offset 1000 and P1 at read-offset 850.
2. Instance A receives input from both P0 and P1 after the shuffle (the keyed redistribution that routes every record for a given key consistently to the same parallel instance). barrier(42) arrives on A's P0 channel first: A stops consuming new P0 records and buffers them (alignment), while continuing to process P1 records normally, since P1's barrier hasn't arrived yet.
3. barrier(42) arrives on A's P1 channel. A has now seen the barrier on every input channel, so it snapshots its local running sums, say {key=X: 17, key=Y: 42}, to durable storage, forwards barrier(42) downstream to the Sink, and unblocks the buffered P0 records, which now belong to checkpoint 43.
4. Instance B does the same independently for its own keys.
5. The Sink receives barrier(42) from both A and B, snapshots (or, if it's a transactional sink, pre-commits) its own pending output, and acknowledges checkpoint 42 to the coordinator.
6. Once the coordinator has acknowledgements from the source (offsets P0=1000, P1=850), A, B, and the Sink, checkpoint 42 is marked complete and persisted.
7. If the job then crashes after checkpoint 42 completed but before checkpoint 43 finished, restart loads checkpoint 42: sources reset to offsets P0=1000/P1=850, A and B restore their snapshotted key-sums, and the Sink either commits its pre-committed checkpoint-42 output or relies on idempotent writes if it isn't transactional. Processing then resumes from exactly those offsets, so no record before the barrier is reprocessed and no record after it is lost.

End-to-end exactly-once needs the sink to participate

Internal exactly-once state is only as strong as what happens at the external sink. A transactional (two-phase commit, 2PC) sink pattern treats each checkpoint as a transaction boundary: during the checkpoint, the sink prepares and flushes its output but does not commit; once the coordinator marks the checkpoint complete, it tells the sink to commit. On failure, any uncommitted transaction is aborted, so restored state and committed output stay consistent. If a sink can't participate in a transaction this way (for example, one that only supports single-item idempotent writes rather than a cross-partition transaction), the fallback is idempotent writes keyed by (checkpoint_id, record_id), so the same output re-emitted after restoring from checkpoint 42 doesn't double-apply. The same barrier/snapshot mechanism underlies this regardless of what the job is computing: a change-data-capture (CDC) ingestion pipeline, a feature-computation job for machine learning, or an ordinary aggregation are all just other kinds of stateful stream jobs from the checkpoint coordinator's point of view.

Trade-offs & pitfalls

Rescaling (changing parallelism) needs a savepoint, a user-triggered durable checkpoint, plus a defined remapping of keyed state across the new number of instances. The most common pitfall is assuming "exactly-once" automatically covers the whole pipeline: if the sink isn't transactional or idempotent, restoring from a checkpoint after a crash can re-emit output that was already delivered before the crash, turning exactly-once internal state into at-least-once external effects.

A gossip protocol is a decentralized way for nodes to spread state (cluster membership, health, small pieces of shared metadata) by periodically picking one or a few random peers and exchanging what each side knows, the same way a rumor spreads through a population. There is no coordinator and no single point of failure: every node's job is identical, and information reaches the whole cluster in a small, predictable number of rounds even as the cluster grows large. Distributed systems reach for gossip for membership tracking and metadata propagation specifically because it scales without needing a central registry to stay in sync.

Basic mechanics

• Each node keeps a small local view of cluster state: who is alive, version numbers, small metadata.
• On a fixed interval, each node picks one or a handful of random peers and exchanges state with them, in one of three common shapes:
  • Push: a node sends its state to a random peer, unprompted.
  • Pull: a node asks a random peer for its state.
  • Push-pull: both directions in one round trip, which converges roughly twice as fast for the same message volume.
• On receipt, each side merges what it learned (for example, keeping whichever version of each entry has the higher counter) and continues gossiping on the next interval.
• This exchange is also called anti-entropy when it specifically reconciles divergent replicas rather than just spreading membership news. A well-known concrete implementation of gossip-based membership is SWIM (Scalable Weakly-consistent Infection-style process group Membership protocol), which layers a lightweight ping and acknowledgment failure-detection scheme on top of the same gossip fan-out.

Convergence: why it scales like an epidemic

Assume, as a simplifying model, that each round of push gossip roughly doubles the number of nodes that have heard a given piece of information, since every already-informed node infects one new random peer per round. Starting from one informed node, after r rounds roughly 2 to the power r nodes are informed. For the whole cluster of N nodes to be informed:

$2^r \ge N \implies r \ge \log_2 N$

For a 1,000-node cluster, log base 2 of 1000 is about 9.97, so full propagation takes on the order of 10 rounds. For a 10,000-node cluster, log base 2 of 10,000 is about 13.3, so about 14 rounds. Ten-fold-ing the cluster size only adds a handful of rounds, because the doubling model grows exponentially, not linearly, in the number of informed nodes; this is the scaling property that makes gossip viable at cluster sizes where a centralized broadcast would become a bottleneck.

Trade-offs & pitfalls

• The doubling assumption above is a simplified model (uniform random peer selection, no message loss, no adversarial behavior); real convergence is probabilistic, and pathological cases (a persistently unlucky peer-selection pattern, high churn, network partitions) can leave a minority of nodes lagging well past the expected round count.
• Load per node stays roughly constant regardless of cluster size, since each node only ever talks to a handful of peers per round, which is the actual scalability win over a centralized registry that every node would otherwise have to poll or push to directly.
• Common wrong turn: assuming gossip gives a hard, guaranteed-delivery bound. It gives a probabilistic, high-confidence bound. A system that needs a strict deadline for propagation, a security revocation for instance, usually pairs gossip with an explicit acknowledgment or a stronger consensus-backed registry for the small set of facts that truly cannot wait.