Chapter 4: Distributed Systems Foundations

"A distributed system is one in which the failure of a computer you didn't even know existed can render your own computer unusable." — Leslie Lamport, 1987

Overview

This is the theory chapter, and it is deliberately short, because most of distributed systems theory is not useful to a data engineer and the part that is fits in one chapter.

Here is the case for reading it anyway. Almost every genuinely hard bug you will meet in this field is a distributed systems bug wearing a costume. The duplicate rows are a retry that landed twice. The missing rows are a watermark that raced a commit. The consumer that keeps falling behind is a hot partition. The nightly job that produces different results when you re-run it is reading a replica with lag. The event that arrives on Thursday with a Tuesday timestamp is event time diverging from processing time.

None of those present as distributed systems problems. They present as data problems, and engineers who do not have the vocabulary spend days on them. Engineers who do recognize the shape in twenty minutes.

So this chapter teaches five things and refuses to teach several more. It teaches partial failure, which is the property that makes distributed systems categorically different rather than just bigger; partitioning, because every scaled data system splits data and the split is a decision with consequences; replication, because copies create lag and lag creates bugs; consistency, because "the latest value" stops being well-defined; and delivery semantics, because at-least-once plus idempotency is the pattern that everything in Part III depends on.

It does not teach consensus algorithms, Byzantine fault tolerance, CRDTs, or the formal treatment of any of the above. Those are genuinely interesting, they are covered superbly in Kleppmann's Designing Data-Intensive Applications, and you will not need them to debug a pipeline.

One warning about a word. "Exactly-once" is the most oversold term in this field. It is achievable in specific, narrow circumstances, it is claimed far more broadly than it is achieved, and §4.5 will make you appropriately suspicious of the claim — including when a vendor makes it about a product you are evaluating.

In this chapter, you will learn to:

• Explain partial failure and why "it might have worked, and I cannot tell" is the defining condition of distributed computing.
• Choose a partition key and predict the skew it will produce, including the specific skew Kestrel's obvious key choices would create.
• Distinguish leader-follower from leaderless replication, and diagnose the pipeline bugs that replication lag causes.
• State what CAP actually says and what it does not, and use PACELC instead, because the trade-off you make every day is latency versus consistency.
• Distinguish the three delivery semantics and explain why exactly-once is a property of one hop rather than of a system.
• Recognize failures caused by clock skew, event time versus processing time, and consumer rebalances.

Who needs this chapter: everyone eventually; streaming engineers immediately. On the Streaming path it is not optional — Chapters 14, 15, and 29 assume it. On the Quick Start path you can defer it, but read §4.5 before Chapter 13.

4.1 Partial Failure, and Why It Changes Everything

On a single machine, an operation succeeds or it fails, and you find out which.

In a distributed system there is a third outcome, and it is the common one: you sent a request, you did not get a response, and you cannot tell whether it was executed.

   your code                                    remote system
       │                                              │
       │──────── write order 88214 ──────────────────▶│
       │                                              │  ✓ written
       │◀ ─ ─ ─ ─ ─ ─ ─ ─ ─ ✗ (network drops) ─ ─ ─ ─ │
       │                                              │
   timeout.
   Was it written?  You cannot tell from here.

Your options are to retry, which may write it twice, or not to retry, which may lose it. There is no third option, and no amount of engineering makes the uncertainty go away — it is a property of the universe, not of your code.

Everything else in this chapter follows from that one fact:

• Idempotency exists so that retrying is safe, which converts an unanswerable question into an answerable one.
• At-least-once delivery is what you get when you choose retry; at-most-once is what you get when you choose not to.
• Consistency models exist because a write that may or may not have happened, replicated to machines that may or may not have received it, makes "the current value" ambiguous.
• Timeouts are the only way to detect failure, and a timeout cannot distinguish a dead machine from a slow one.

🏭 From the Pipeline — The retry that ran the job three times

An orchestrator submits a Spark job to a cluster over HTTP and waits for a submission ID. The cluster was under load; the response took longer than the 30-second client timeout. The orchestrator's retry policy did what it was configured to do and resubmitted. Twice.

Three identical jobs ran concurrently, all writing the same output partition. Two of them completed. The output had roughly double the rows it should have.

Every component behaved correctly. The cluster accepted the submissions it was sent. The orchestrator retried a request that timed out, which is the right default. The job wrote its output, which is its purpose.

The bug was that "submit a job" was treated as an idempotent operation when it was not. The fix was a client-generated idempotency key — a deterministic job name derived from (dag_id, task_id, logical_date) — so a resubmission of the same logical work is recognized and rejected rather than accepted as new work.

This is the single most transferable pattern in the chapter. Any operation you might retry needs a client-generated key that identifies the logical work, not the attempt.

The eight fallacies, compressed

Peter Deutsch and colleagues at Sun catalogued the assumptions that distributed systems newcomers make. All eight are false; four bite data engineers regularly.

4.2 Partitioning

Partitioning — also called sharding — splits a dataset across machines or files so that work can proceed in parallel. It is the single most consequential layout decision in a data platform, and Chapter 1's $3,840 Spark job was a partitioning decision meeting a query that ignored it.

The three schemes

Range partitioning. Split by ordered ranges of a key: January in one partition, February in the next. Range queries are efficient because they touch few partitions. The risk is skew, and time-based ranges have a specific pathology — the current partition is always the hot one, because that is where all the writes go.

Hash partitioning. Apply a hash to the key and take it modulo the partition count. Distribution is even, provided the key has enough distinct values. The cost is that range queries touch every partition, because adjacent keys land in unrelated partitions.

Consistent hashing. A refinement that minimizes redistribution when the partition count changes. Under plain hash-modulo, going from 12 partitions to 16 remaps roughly 75% of keys; under consistent hashing, roughly $1/n$ of them move. Relevant to Chapter 12's key-value stores, and mostly invisible in the tools this book uses.

Skew is the failure mode

A partitioning scheme is only as good as its worst partition, because a job finishes when its slowest task finishes. This is worth stating as a rule: you are not paying for average partition size, you are paying for maximum partition size, times the number of nodes waiting.

Kestrel's clickstream is keyed by session_id in Kafka. Consider the alternatives that seem reasonable and what each would produce:

The pattern across the bad choices: low cardinality and natural popularity are both skew generators, and business dimensions tend to have both.

⚠️ Failure Mode — The hot partition nobody predicted

A pipeline partitioned order events by customer_id, which is high-cardinality and looked safe. It ran fine for a year.

Then a business-to-business customer opened an account. One customer_id, placing roughly 8% of all orders — a wholesale buyer submitting large batch orders through the same API.

One partition now had 8% of the data. Every job's runtime became that partition's runtime. The nightly aggregation went from 22 minutes to 71 minutes and the 6am SLA started failing about once a week, apparently at random — actually on days that customer placed a large batch.

High cardinality is not the same as even distribution. A key can have millions of distinct values and still be dominated by a few. Three practical defenses:

1. Measure the distribution before choosing, and again quarterly. One query: SELECT key, COUNT(*) FROM t GROUP BY key ORDER BY 2 DESC LIMIT 20.
2. Salt the hot keys: partition on customer_id || '-' || (row_number % 16) for known-hot keys, so one logical key spreads across sixteen partitions. Chapter 21 §21.7 covers this properly.
3. Alert on skew ratio — max partition size over median partition size. Above about 4×, something has changed.

Partition size

Two failure directions, and the guidance is different for each system, but the shape is universal.

Too large and you lose parallelism, individual tasks run long, a failure loses a lot of work, and memory pressure rises. Too small and per-partition overhead dominates: file listing, task scheduling, metadata, and — as computed in Chapter 2 §2.4 — request charges.

Working targets:

💸 Cost Check — What partition sizing cost Kestrel, restated as a rule

Chapter 1's job scanned 4.2 TB instead of 34 GB because a CAST defeated partition pruning: $3,840.00 a night against $74.88.

The general rule behind that number: the cost of a scan is not the size of the table, it is the size of the partitions your predicate fails to eliminate. Three things break pruning, and all three are written by careful engineers producing correct results:

1. A function on the partition column. WHERE CAST(event_ts AS DATE) = '...' cannot be matched to event_date. Filter the partition column directly.
2. A type mismatch. WHERE event_date = '2025-06-11' against a DATE column may force a cast on every row. Use DATE '2025-06-11'.
3. A join predicate the planner cannot push down. WHERE event_date IN (SELECT d FROM dates) frequently defeats pruning. Materialize the list or use a broadcast.

Check PartitionFilters in the plan before you check anything else. An empty bracket is the most expensive punctuation in data engineering.

4.3 Replication

Replication keeps copies of data on multiple machines, for three reasons: durability (a machine dies and the data survives), read scaling (reads spread across replicas), and locality (a replica near the reader).

It also creates the single most under-diagnosed class of data pipeline bug, so this section spends most of its length there.

Leader-follower

One node accepts writes; others replicate from it and serve reads. This is PostgreSQL's model, MySQL's model, and the model of most transactional databases you will extract from.

Synchronous replication waits for a follower to acknowledge before confirming the write. No data loss on leader failure; higher write latency; and if a follower is slow, writes stall.

Asynchronous replication confirms immediately and replicates in the background. Fast writes, possible data loss on failure, and — the part that matters to you — followers are behind by an amount that varies.

That amount is replication lag, it is usually milliseconds, and it is occasionally minutes.

⚠️ Failure Mode — The extract that read a stale replica

Kestrel extracts orders from a PostgreSQL read replica, which is correct practice: you do not run analytical extracts against the primary that is serving checkout.

The extract's watermark logic reads the maximum updated_at it has seen and stores it. On a normal night, replica lag is under 200 ms and nothing goes wrong.

On a night when a large batch update ran on the primary — a bulk price change across 12,000 products — the replica fell four minutes behind. The extract ran at 02:00, read up to the replica's view of the world, and stored a watermark of 01:56. Rows committed on the primary between 01:56 and 02:00 had updated_at values below that watermark by the time they arrived on the replica, and the next night's extract, filtering on updated_at > 01:56, skipped them permanently.

Same shape as the watermark bug in Chapter 2 §2.3, arriving through a different door. The general form: any process that records "where I got to" using a value read from a lagging replica will skip whatever arrives in the lag window.

Three fixes:

1. Read the watermark from the primary even when reading data from the replica. One cheap query against the primary; solves it exactly.
2. Subtract a safety margin larger than your worst observed lag — watermark - 15 minutes — which converts loss into duplicates, and duplicates are handled by idempotent writes.
3. Monitor replication lag and refuse to extract when it exceeds a threshold. Failing loudly beats succeeding incorrectly (Chapter 2, Case Study 1).

Kestrel does all three. They are not redundant: the first is correct, the second is defense in depth, and the third tells you when something upstream is wrong.

Leaderless and quorums

Some systems (Cassandra, DynamoDB, and their descendants) let any replica accept a write. Consistency comes from quorums: write to $W$ replicas, read from $R$, out of $N$ total. If

$$W + R > N$$

then any read set overlaps any write set, so a read is guaranteed to see the most recent write.

With $N = 3$: $W = 2, R = 2$ satisfies it and is the common default. $W = 1, R = 1$ does not, and gives you fast operations and stale reads.

Where this matters to a data engineer: if you extract from such a store, you must know what consistency level the extract uses. An extract at $R = 1$ can read a stale value for a row that was updated seconds ago — and, worse, can read different values on two consecutive attempts. That looks exactly like a source data problem and is not one.

4.4 Consistency, CAP, and the Thing People Get Wrong

What CAP actually says

The CAP theorem is the most cited and least accurately quoted result in this field.

What it says: in the presence of a network partition — messages between nodes being dropped — a distributed system must choose between remaining available (every request gets a response) and remaining consistent (every read sees the most recent write). You cannot have both, during a partition.

What people think it says: "pick two of consistency, availability, and partition tolerance."

That framing is wrong in a way that matters, because partition tolerance is not optional. Networks partition. You do not get to choose otherwise. So the actual choice is binary and it only applies during a partition: when the network splits, do you refuse requests or serve possibly-stale data?

And here is the part that renders CAP nearly useless for daily work: partitions are rare. CAP tells you nothing about the other 99.9% of the time.

PACELC, which is the useful version

Daniel Abadi's extension states the whole trade-off:

If there is a Partition, choose between Availability and Consistency; Else — in normal operation — choose between Latency and Consistency.

The second clause is the one you actually make decisions about. Every day, in every system, there is a dial between "wait for the write to be confirmed everywhere" and "respond now and reconcile later." That is a latency-versus-consistency choice and it has nothing to do with partitions.

🧭 Version Note — S3's consistency changed, and old advice persists

Before December 2020, S3 was eventually consistent for overwrites and deletes: you could write an object, immediately read it, and get the previous version. An enormous amount of data engineering practice was built around this, including whole subsystems — Netflix's S3mper, the EMRFS consistent view, and the reason early Delta Lake and Iceberg designs are so careful about listing.

S3 is now strongly read-after-write consistent for all operations, including overwrites, list, and delete, at no extra cost and with no configuration.

Two consequences. First, advice written before 2021 about working around S3 eventual consistency is obsolete and you will still find it near the top of search results. Second — the part that generalizes — a system property you designed around can change underneath you, and nothing will tell you. When you find advice that seems oddly baroque, check its date and check whether the premise still holds. Other object stores have their own consistency models; verify rather than assume.

Eventual consistency, stated precisely

"Eventually consistent" means: if writes stop, all replicas will converge to the same value, eventually. Two things it does not promise, and both catch people:

• When. "Eventually" is unbounded. Usually milliseconds. Occasionally not.
• What you see meanwhile. Reads may go backwards in time. A value can appear, disappear, and reappear.

For a data engineer this matters most when reading a source system directly. A pipeline that reads an eventually consistent store twice may legitimately get two different answers, and neither is wrong. If your reconciliation compares warehouse to source and the source is eventually consistent, your reconciliation needs a tolerance and a settling period, or it will page you for physics.

4.5 Delivery Semantics

Three guarantees, and the third is oversold.

At-most-once

Send it; do not retry. Messages may be lost; they are never duplicated. Appropriate for high-volume, low-value telemetry where loss is preferable to cost. Almost never appropriate for business data, and you should be suspicious of any pipeline that has it by accident — which usually means "we do not retry, because nobody thought about it."

At-least-once

Retry until acknowledged. Nothing is lost; things may be duplicated. This is the default in nearly every real system, and correctly so, because it is achievable with a timeout and a retry loop.

The duplicate is not a defect to be eliminated. It is a cost you pay to guarantee delivery, and you pay for it by making the write idempotent.

Exactly-once, and why to be suspicious

Every message is processed once, no loss, no duplication. This is what everyone wants and what many products advertise.

The honest position: exactly-once is achievable within a closed system that controls both the message log and the state store, and it is a claim about that hop rather than about your pipeline.

Kafka's transactional producer with read_process_write genuinely provides exactly-once semantics from a Kafka topic, through a Kafka Streams application, to another Kafka topic. That is real and it is well-engineered. What it does not cover is the moment your consumer writes to PostgreSQL, or calls a payment API, or writes a Parquet file to S3 — because Kafka cannot make an external system participate in its transaction.

   ┌────────────────────────────────────────┐
   │   Kafka transactional boundary         │        ┌───────────────┐
   │                                        │        │  PostgreSQL   │
   │  topic A ──▶ Streams app ──▶ topic B   │───────▶│  (outside the │
   │                                        │        │  transaction) │
   │  exactly-once genuinely holds HERE     │        └───────────────┘
   └────────────────────────────────────────┘              ▲
                                                           │
                                    at-least-once, and your problem

In words: Kafka's transaction covers the consume-process-produce loop inside Kafka. The write to an external system sits outside that boundary and is at-least-once, whatever the marketing says.

So the pattern that actually works, and that this entire book relies on:

At-least-once delivery + idempotent writes = effectively-once processing.

Duplicates arrive; the write makes them harmless. This works with any transport, any external system, and no distributed transaction coordinator. Four ways to make a write idempotent:

Chapter 20 §20.3 works through all four with code.

🎓 Interview Angle — "How do you achieve exactly-once processing?"

A trap question, and the trap is answering it as asked. Candidates who confidently describe a configuration flag are demonstrating that they have read a feature list.

"I'd push back gently on the framing. Exactly-once end to end across systems that don't share a transaction coordinator is generally not achievable — Kafka's transactional producer gives it genuinely, but only within Kafka, and the moment I write to Postgres or S3 I'm outside that boundary. What I'd build is at-least-once delivery with idempotent writes, which gives effectively-once results. Concretely that means a deterministic key for each unit of work and a write that's a merge or a partition replacement rather than an append. The advantage is it works with any transport and any sink."

Then name the case where you would use Kafka transactions: a Kafka-to-Kafka stream processor where the boundary genuinely holds. Knowing when the strong claim is true is what separates skepticism from cynicism.

4.6 Time, Clocks, and Ordering

Time in a distributed system is not what you think it is, and three specific confusions cause most data bugs in this area.

There is no global clock

Machine clocks drift. NTP corrects them, and correction means a clock can jump backwards. A timestamp taken on machine A and one taken on machine B are not reliably comparable, and the error is typically milliseconds and occasionally much worse.

Three practical consequences:

• Never order events from different machines by wall-clock timestamp and expect correctness.
• Never compute a duration by subtracting two wall-clock readings taken on different machines.
• Never assume created_at < updated_at. With clock skew or a backwards jump, it can be false, and a CHECK constraint asserting it will eventually fire in production.

For durations on one machine, use a monotonic clock (time.monotonic() in Python), which never goes backwards.

Event time versus processing time

The most important distinction in this section.

Event time is when the thing happened. Processing time is when your system saw it.

For Kestrel's clickstream they differ by milliseconds most of the time — and by hours for a mobile client that went into a tunnel, buffered events locally, and flushed them when it reconnected.

event time:      09:14:22   user taps "add to cart" on a train
                    │
                    │   ... 47 minutes offline ...
                    ▼
processing time: 10:01:05   the app reconnects and flushes its buffer

Now: which day does that event belong to? If it happened at 23:58 and arrived at 00:07, does it count toward Tuesday or Wednesday?

Event time is almost always the right answer for analytics, because the business question is "what did customers do on Tuesday," not "what did our servers see on Tuesday." But event time requires you to handle late data — you cannot close Tuesday's books at midnight if events from Tuesday will still arrive on Wednesday.

That is what a watermark is: a declaration that you no longer expect events older than a given event time, and that anything arriving later will be handled by a stated policy — dropped, sent to a side output, or triggering a recomputation. It is a business decision wearing technical clothes, and Chapter 29 §29.4 treats it properly.

Kestrel's policy, stated once and used throughout: bronze is partitioned by processing date (ingest_date), because that is what the writer knows and it makes the write path simple and idempotent. Silver and gold are partitioned by event date, and the transformation from bronze to silver re-buckets by event time, with a three-day late-arrival window and a daily reprocessing of the trailing three days. Events later than three days go to a side table and a weekly report.

📐 Design Decision — The three-day late window

Three days is a choice, and it costs something in both directions.

Longer (say seven days) catches more late events and makes numbers more accurate. It also means reprocessing seven days of data every night instead of three — more compute — and it means published numbers keep changing for a week, which analysts and finance find intolerable.

Shorter (say one day) settles numbers faster and costs less. It drops more genuinely late events. At Kestrel, measurement showed 99.2% of events arrive within one hour and 99.97% within three days; the remaining 0.03% is roughly 4,200 events a day, which matters for completeness claims and does not move any aggregate.

What you give up by choosing three days: finality. A number published Tuesday can change until Friday. The mitigation is to state this explicitly in the serving layer — every daily aggregate carries a is_final flag — so that a consumer knows whether they are looking at a settled number. Silently changing history is far worse than changing it visibly.

Ordering

Kafka guarantees ordering within a partition, not across partitions. This is why Kestrel keys the clickstream by session_id: all of one session's events land in one partition and are therefore ordered relative to each other, which is what sessionization requires. Global ordering across the whole topic is not guaranteed and would require a single partition, which would destroy throughput.

The general principle: order is guaranteed exactly as far as the partition boundary and no further. If you need two things ordered relative to each other, they must share a partition key.

4.7 The Failure Modes You Will Actually Meet

Five, in rough order of how often they appear in a data engineer's life.

1. The consumer rebalance

When a consumer joins or leaves a group, partitions are reassigned. During the rebalance, consumption stops. If your consumer takes too long between polls — because it is doing a slow write — the broker concludes it is dead and triggers a rebalance, which makes everything slower, which causes another rebalance.

The symptom: lag climbing and falling in a sawtooth, log lines about group coordination, and throughput far below what a single consumer achieves in isolation.

The fixes: poll frequently and process asynchronously; tune max.poll.interval.ms to exceed your worst-case processing time; and batch writes so that a single poll's work is bounded. Chapter 15 §15.5.

2. The straggler

In any parallel job, completion time is the maximum over tasks, not the average. One slow task — from skew, a slow node, or a large file — holds up everything.

The symptom: 199 of 200 tasks finish in two minutes and the job takes forty.

The fixes: fix the skew (§4.2); enable speculative execution so the slow task is duplicated on another node; and size partitions evenly. Chapter 21 §21.5.

3. Backpressure, or its absence

A producer faster than its consumer must be slowed down, or something must buffer, or something must drop. Systems with backpressure slow the producer. Systems without it accumulate an unbounded queue until memory runs out.

The symptom: memory growth ending in a crash, or a queue depth graph that only goes up.

The fix: bound every queue explicitly and decide what happens when it fills. "Unbounded" is a decision to fail later, in a worse way.

4. The thundering herd

Everything retries at once after a failure, and the retries themselves keep the system down. Common after a brief outage: a hundred workers all wake, all retry at the same second, and overwhelm the recovering service.

The fix: exponential backoff with jitter. The jitter is not optional and is the part people omit — without randomization, backoff merely synchronizes the herd at a longer interval. Chapter 16 §16.5 has the code.

5. Cascading failure

One component slows, its callers block waiting, their callers block waiting, and a slowdown becomes an outage.

The fixes: timeouts on every remote call, without exception; circuit breakers that fail fast when a dependency is unhealthy; and bulkheads that isolate one failing dependency from unrelated work.

🧪 Try It — Find the unbounded queue

Open any pipeline you have access to — one of yours, or one of this book's code samples — and answer three questions in writing:

1. Where does data buffer? In memory, in a queue, in a table, in a topic. List every place.
2. What is the bound on each? A number, in rows or bytes or messages. If the answer is "none," you have found an outage waiting for a slow day.
3. What happens when each one fills? Block the producer, drop the data, spill to disk, or crash. If you do not know, the answer is crash.

Ten minutes, and it is the fastest structural review of a pipeline there is. Most engineers discover at least one unbounded queue on their first attempt.

Backpressure, and the queue you did not know you had

Every pipeline is a series of stages connected by buffers, and a buffer is a queue whether or not anybody called it one. A Kafka topic is a queue. A directory of unprocessed files is a queue. A warehouse's query concurrency limit turns a connection pool into a queue. So is a Python list that a producer thread appends to and a consumer thread pops from.

Little's Law from Chapter 3 §3.5 tells you what a queue does at steady state. It says nothing about what happens when the arrival rate exceeds the service rate, and that is the case worth understanding, because it is the case that produces incidents.

arrival rate < service rate     the queue stays short. Nothing to see.
arrival rate = service rate     the queue length is a random walk. It grows.
arrival rate > service rate     the queue grows WITHOUT BOUND until
                                something stops it

"Until something stops it" is the whole subject. Four things can:

The buffer fills and the producer blocks. This is backpressure working. The slowness propagates upstream, arrival slows to match service, and the system degrades gracefully. It is the outcome you want and it is not the default.

The buffer fills and the producer drops. Also a legitimate design — a metrics pipeline should shed load rather than block the application emitting the metrics — and it is only legitimate if the drops are counted. An uncounted drop is silent data loss with a configuration flag in front of it.

The buffer fills and the process runs out of memory. This is the default for an unbounded in-process queue, and the failure arrives as an OutOfMemoryError in a component that is not the problem.

Or the buffer is durable and unbounded, and fills a disk. Chapter 14's replication slot is exactly this: a queue with no consumer, retained on the source database's volume, taking down a system that has nothing to do with your pipeline.

The audit that finds them

List every place data waits, and for each one answer three questions:

1. What is the bound?              a size, a count, a duration -- or "none"
2. What happens when it fills?     block / drop / crash / fill a disk
3. Is the "what happened" COUNTED? and does anyone look at the counter?

Question 1 is where most of the findings are. A queue whose bound you cannot state is a queue that will be discovered at its limit, and "none" is a valid answer only if you have computed what "none" costs at the maximum plausible arrival rate.

Question 3 is where the rest are. A bounded queue that drops silently is indistinguishable from a queue that never filled, which is Chapter 25's absence problem in a new place.

At Kestrel the audit found seven buffers and two unbounded ones: the replication slot (bounded later by max_slot_wal_keep_size, which turns a disk-full into a slot invalidation you chose) and an in-process list in the API ingester that held a page of results while the writer was slow. The second had never filled and would have, on the day the warehouse was slow and the API was fast.

Why backpressure is not free

Blocking the producer is correct and it moves the problem. A consumer that cannot keep up now slows the producer, and if the producer is an application serving customers, you have connected your analytics pipeline's throughput to your checkout page's latency. That is precisely the coupling Chapter 7 §7.1 says to avoid, arriving from the other direction.

The resolution is the same one as everywhere else in this chapter: put a durable, bounded buffer at the trust boundary. The application writes to a log that can absorb a day of arrivals; the pipeline reads from it at whatever rate it manages. Backpressure then propagates within your system and stops at the boundary, which is what a log is for and is a better reason to adopt one than throughput.

Consensus, briefly, and where you meet it

You will not implement consensus. You will depend on it constantly, and knowing roughly what it guarantees explains several behaviours that otherwise look like bugs.

The problem: several machines must agree on a value — who is the leader, what is the next entry in the log, which of two writers holds the lease — in the presence of §4.1's third outcome. Any machine may be slow, may be unreachable, or may have been unreachable and come back believing it is still in charge.

The guarantee that Raft and Paxos provide is narrower than people assume:

SAFETY    no two machines ever decide DIFFERENT values.
          Holds always, including during a partition.
LIVENESS  a decision is eventually reached.
          Holds only while a MAJORITY can communicate.

Safety always; liveness sometimes. That is CAP (§4.4) restated at the mechanism level, and it is why a consensus-backed system becomes unavailable rather than wrong during a partition.

The three places a data engineer meets it

A leader election, and the fencing token that goes with it. Kafka's controller, a Postgres failover, a distributed lock in your own code. The important detail is the token: a machine that was the leader, was partitioned, and comes back still believes it is the leader for as long as its lease has not visibly expired to it. Consensus does not prevent that; it prevents its writes from being accepted. Every correct lease implementation therefore carries a monotonically increasing number, and the storage layer rejects a write bearing an old one (Exercise 3.15's protocol, and Chapter 10's commit).

A quorum, and why the numbers are odd. Three, five, seven — never four. A majority of four is three, the same as a majority of five, so the fourth node adds cost and no fault tolerance. f failures require 2f + 1 nodes, and that single line explains every cluster-sizing recommendation you will read.

And a commit that must be atomic across machines — a table format's transaction log (Chapter 10), a warehouse's distributed write. The compare-and-swap on a pointer is consensus with one participant doing the deciding, which is why it works and why it needs an external store that provides it.

What this buys you when reading an incident

Three behaviours that look like bugs and are the algorithm working:

A cluster refuses writes while a majority is unreachable. It is choosing safety. The alternative is two halves accepting conflicting writes, and you would rather have the outage.

A failover takes tens of seconds. The election cannot start until the old leader is believed dead, which requires a timeout, and the timeout must exceed the worst normal pause — including a garbage collection. Shortening it causes spurious failovers, which is worse than a slow one.

And a node that was partitioned rejoins with stale state and is corrected. It does not serve the stale state, because it cannot commit without the majority. The window in which it might have served a stale read is the one thing to check in your client's configuration, and it is usually a setting named something like "read from replica."

🧱 Kestrel Platform — where each of this chapter's failures actually lives

text failure mode where it is in Kestrel what defends it ───────────────────────────────────────────────────────────────────────── timeout, unknown outcome the carrier API (ch 16) idempotency keys every warehouse write a keyed merge replication lag the orders extract (ch 13) CDC, eventually partial failure the nightly Spark job partition-level idempotent writes duplicate delivery the clickstream (ch 15) dedup on event_id clock skew client-side event_ts the watermark and its lateness (ch 29) an unbounded buffer the replication slot (ch 14) max_slot_wal_keep_size the API ingester's page list a bounded queue

Every row is a chapter, and the chapter is longer than the row. That is the point of putting the table here: the failures are five or six in number, and the rest of this book is those five or six in different clothing.

The column worth staring at is the third. Four of the seven defences are the same mechanism — make the write idempotent and key it on something stable — and the other three are bound the buffer and count what it drops. There is no third category, which is a genuinely useful thing to know before you read Part III.

📏 Scale Note — which of these failures you meet at which size

Every failure in this chapter is real at every size. They are not equally likely, and knowing the order they arrive in tells you what to build first.

text one machine a few nodes a cluster many clusters ────────────────────────────────────────────────────────────────────────────── timeout / unknown rare COMMON constant constant partial failure n/a occasional COMMON constant replication lag n/a COMMON common common clock skew n/a rare occasional COMMON network partition n/a rare occasional common an unbounded buffer COMMON common common common

Two rows are worth reading against the others.

The unbounded buffer is common at every size, including one machine. It is the only failure in this chapter you can meet with no network at all — a Python list, a file being written faster than it is read — and it is the one people do not associate with distributed systems, which is why it is unguarded.

Clock skew is the reverse. It is negligible on one machine and unavoidable across clusters, and it is the failure whose symptom changes most with scale: at a few nodes it is a puzzling ordering anomaly; at many, it is a systematic bias that shows up as a business trend (Chapter 29's watermarks exist for it).

The practical ordering for a small team, which is the useful output of the table: bound your buffers and make your writes idempotent first, because those two cover the failures you will actually meet this year. Consensus, partition behaviour, and clock discipline are things you will depend on long before you have to implement anything about them.

🔐 Privacy & Governance — at-least-once means at-least-once copies

Every mechanism in this chapter creates copies, and a copy of personal data is an obligation whether or not anybody meant to create it.

text mechanism the copy it creates ──────────────────────────────────────────────────────────────────── a retry a duplicate row, until the merge collapses it a dead-letter queue a full payload, retained, outside the model a replication slot's backlog hours of WAL, on the SOURCE's disk an idempotency key store a record of every request, by definition a quarantine table rejected rows, retained, with the reason a buffer that spilled to disk a partial dataset, in a temp path

The bottom four are the ones outside every manifest, and they share a cause: they exist to make a failure recoverable, so they are designed to be durable and nobody assigns them a retention.

The specific hazard with a dead-letter queue is that it holds exactly the records that a deletion job would fail to process — malformed ones, ones with an unexpected identifier — so a deletion sweep over the model can complete successfully while the DLQ still holds the person.

Three lines to add to any of these, at the moment you build it:

text a retention 30 days is a decision; "forever" is not a classification tag so the generated manifest (ch 31) finds it a place in the register Chapter 4 section 4.21's table, with its key

And the register is the artifact that makes this tractable, because a write operation with a key is a write operation you can delete from. A durable copy with no key is a copy you can only delete in its entirety, which is why the register asks for the key rather than for the target.

4.8 Summary

Partial failure is the property that makes distributed systems categorically different. You sent a request, got no response, and cannot tell whether it executed. Retry and you may duplicate; do not and you may lose. Everything else in the chapter follows: idempotency exists so retrying is safe, delivery semantics name which side of the choice you took, consistency models exist because "the current value" becomes ambiguous, and a timeout cannot distinguish a dead machine from a slow one. Any operation you might retry needs a client-generated key identifying the logical work, not the attempt.

Partitioning is the most consequential layout decision you make, and skew is its failure mode. You pay for the maximum partition, not the average, because a job finishes when its slowest task does. Low cardinality and natural popularity both generate skew, and business dimensions tend to have both — event_type, country_code, and warehouse_id are all wrong for Kestrel's clickstream and session_id is right. High cardinality is not even distribution: one B2B customer at 8% of orders turned a 22-minute job into 71 minutes and broke the SLA weekly, apparently at random.

Replication creates lag, and lag creates a specific class of bug: any process that records "where I got to" from a lagging replica skips whatever arrives in the lag window. Read the watermark from the primary, subtract a safety margin larger than your worst observed lag, and refuse to extract when lag exceeds a threshold. All three, because they do different jobs.

CAP is nearly useless for daily work because partition tolerance is not optional and partitions are rare. PACELC is the useful form: if Partitioned, choose Availability or Consistency; Else, choose Latency or Consistency — and the second clause is the one you decide every day. "Eventually consistent" promises convergence if writes stop; it does not promise when, and it does not promise that reads move forward in time.

Exactly-once is a claim about one hop, not about your system. Kafka's transactional producer genuinely provides it from Kafka, through a stream processor, to Kafka. The moment you write to PostgreSQL or S3 you are outside that boundary, whatever a feature list says. At-least-once delivery plus idempotent writes gives effectively-once processing, works with any transport and any sink, and is what this book uses throughout. Four ways to get idempotency: delete-insert, merge on a natural key, partition replacement, deduplicate on read.

Time is three separate confusions. There is no global clock — never order cross-machine events by wall clock, never subtract cross-machine timestamps, and never assume created_at < updated_at. Event time is when it happened and processing time is when you saw it, and for a mobile client they differ by hours; event time is right for analytics and it obliges you to handle late data, which is what a watermark is for. And order is guaranteed exactly as far as the partition boundary and no further.

Five failure modes you will actually meet: consumer rebalance (sawtooth lag), stragglers (199 tasks in two minutes and one in forty), missing backpressure (an unbounded queue is a decision to fail later, worse), thundering herd (backoff without jitter just synchronizes the herd), and cascading failure (timeouts on every remote call, without exception).

What's next

Chapter 5 surveys the tool landscape — what each category of tool does, what it costs to operate, and what you can safely leave out. It is the most perishable chapter in the book and it says so. It is also where the Kestrel Platform gets its first docker-compose.yml, and the theory in this chapter stops being abstract: you will be able to see, in a running system, why the clickstream topic has twelve partitions and why the consumer commits when it does.