Unit 5.6 · about 5 h
271 / 324Consumer lag, watermarks and state rebuild
Assume Programming's backpressure-graceful-shutdown unit. This unit moves beyond queue admission to persisted backlog and event-time state, where replay must recover the right data rather than merely restart a worker.
- explain
- implement
You can now
- Diagnose offset lag, retention gaps and poison-record stalls without confusing them with event-time lateness.
- Implement a window policy with explicit watermark and late-data rules.
- Rebuild state from a matching snapshot and log suffix, and explain when compaction preserves state but loses history.
Covers
- Consumer lag, backlogs, dead letters and poison records
- Event time, processing time, watermarks and late data
- Snapshots, compaction and state rebuild from a log
Study, in this order
2 steps
- DDIA, 2nd edition: backlog, snapshots and event timeDesigning Data-Intensive Applications, 2nd edition - distributed data path
Chapter 12, dead letter queues on p. 495; Consumer offsets through When consumers cannot keep up with producers, pp. 498 to 499; Initial snapshot and Log compaction, pp. 504 to 506; CDC versus event sourcing, pp. 507 to 508; Reasoning About Time, pp. 518 to 522; Microbatching and checkpointing, p. 527
- Kafka 4.3 design: what compaction retainsApache Kafka documentation - design
Log Compaction Basics and What guarantees does log compaction provide?
Depth · Read offset stability, holes and tombstone retention. Skip cleaner tuning and the diagrams' throughput claims; model compaction in Python instead of deploying Kafka.
Focus on
- For a dense toy log, offset lag is end-next-offset minus the consumer's next offset. Compaction holes mean that distance is not generally a count of remaining records; neither metric is event-time lateness.
- A poison record can block a partition indefinitely. A durable dead-letter handoff must preserve identity, reason and replay policy; advancing progress changes the main stream's completeness contract.
- A watermark is a progress claim or estimate with assumptions, not proof that no delayed event exists. State the input scope, idle-input policy and handling of events below it.
- A snapshot needs the exact log position that its state represents. Latest-value compaction works for replacement updates, not arbitrary increment events; tombstone expiry can invalidate rebuilding an old nonempty snapshot.
Reproduce closed-book
In Python 3.14, feed event timestamps
[2,12,4] in that arrival order, using width-10 half-open tumbling windows, initial watermark \(-\infty\), and \(\text{watermark}=\max(\text{timestamp seen})-3\) after each arrival. Before updating the watermark, send any timestamp strictly below the previous watermark to a late side output. Get one event in \([0,10)\), one in \([10,20)\), late output [4], and final watermark 9. This is a declared toy lateness policy, not a measured clock bound. Separately replay replacement records \((\text{offset}, \text{key}, \text{value})\): \((0, a, 1), (1, b, 2), (2, a, 3), (3, b, \text{null}), (4, c, 4)\), where null deletes. Snapshot after offset 2 is \(\{a:3, b:2\}\) with next offset 3; snapshot plus suffix and full replay both yield \(\{a:3, c:4\}\). Compact by retaining the highest-offset record per key, including \(b\)'s tombstone and original offsets, and rebuild the same final state from empty. Use exact dictionary assertions. Explain why deleting \(b\)'s tombstone before replaying into the old snapshot leaves \(b\) incorrectly present, and why latest-only compaction of two increment(+1) events loses one increment.Traps
- Publishing a watermark from the fastest input while ignoring a slower active input. A multi-input policy must account for each input's progress and explicitly handle idleness.
- Skipping a poison record without durably recording the dead-letter decision, or claiming the main output is still complete after the skip.
- Renumbering compacted offsets, replaying from the snapshot's last applied offset instead of its next offset, or rebuilding old state after its required log suffix was deleted.
Practice
The section mini-project covers this: offset and event-time progress, bounded retry and durable dead-letter handling, late-data tests, consistent snapshot positions, retained tombstones and rebuild equivalence after node loss.
Self-test
-
In a dense log with end-next-offset 100 and committed next offset 70, what is offset lag?
30 offset positions. It is not 30 seconds, and in a compacted log it need not mean 30 retained records. -
Does watermark 9 mean an event timestamped 4 cannot arrive?
No. The toy policy predicts progress from observed timestamps; a delayed event may still arrive and must follow the declared late-data rule. -
When should replay start for the snapshot after offset 2?
At next offset 3. Reapplying offset 2 could duplicate a non-idempotent effect even if this replacement-only example masks the mistake. -
Can latest-value compaction preserve both a balance and its transaction audit history?
It can preserve the current balance if records replace the full value. Discarded earlier records cannot supply the full audit history; increment-only events also cannot be compacted that way.
Completion criterion
Stop when
Your window and rebuild assertions pass, and you can give separate counterexamples for an unsafe watermark, an expired tombstone and compaction of increment events.
Mark passed, then open 5.7 Only mark this when you can do it without notes.Leads to
- 5.7 Distributed histories and deterministic failure simulation Inject gaps, duplicates and node loss into your state rebuild contract.
Comes after
- 4.2 Aggregations, windows, nulls and point-in-time snapshotsCarry availability and late-arrival rules into streaming state.
- 5.5 Durable logs, delivery guarantees and idempotent consumersAdd retention, poison records and state reconstruction to replay.
Resources for this unit
- Designing Data-Intensive Applications, 2nd edition - distributed data path Read the selected chapters on replication, partitioning, distributed-system trouble, consistency and consensus, batch processing and stream processing. For each guarantee, write its assumptions, a confirming history and one counterexample outside the boundary.
- Apache Kafka documentation - design Read Persistence, The Consumer (Consumer Position), Message Delivery Semantics, Using Transactions, Replication and Log Compaction for a concrete log: offsets, consumer positions, the at-most-once, at-least-once and exactly-once boundaries, in-sync replicas and compaction guarantees. Map each guarantee onto your own log before trusting it.
Also in this section
- Gossip Glomers - Echo, Unique ID and Broadcast Complete Challenge 1 Echo, Challenge 2 Unique ID Generation and Challenge 3a Single-Node Broadcast. Add 3b Multi-Node Broadcast only after writing the delivery, duplicate and partition assumptions. Preserve protocol histories rather than relying on a success banner.
- MIT 6.5840 - MapReduce, RPC and consistency bridge Use the introductory MapReduce, RPC, GFS and consistency lectures after the local RPC and store gates. The MapReduce lab is optional. Do not begin Raft or sharded KV as foundation work.
- Building Distributed Systems from Scratch - one challenge path Choose one messaging, durable-log or partitioning challenge that matches a failed checkpoint. State the model, failures and evidence independently because interactive grading behavior has not been verified.
- AlgoMaster System Design - distributed fundamentals slice Use Database Replication, Consistency Models, Logical Clocks, Vector Clocks, Network Partitions, Split Brain Problem, Leader Election, Database Partitioning, Consistent Hashing, Delivery Semantics, Dead Letter Queues, Idempotency, Two-Phase Commit (2PC), Outbox Pattern and Streaming Engines as bounded review after DDIA. Attempt only the exercise that matches your implemented history, then challenge its assumptions with a node-loss, delay or duplicate case.