# DDIA Ch. 12: a straggler silently undercounts a fired window; a watermark recovers it ## Concept Group a stream of events into 1-minute **event-time** tumbling windows and count per window. Window `[37, 38)` accumulates 5 events; then time moves on — events start arriving with minute-38 and minute-39 timestamps — so the aggregator declares window 37 "complete" and **fires** it, emitting **count = 5**. Then a **straggler** arrives: an event whose event_time is minute 37, delayed by a network hiccup so it lands *after* the minute-39 events. The window it belongs to has already closed. Under the default **drop-late** policy the straggler is silently **dropped** — no error — so window 37's published count stays **5** while the true count is **6**: an undercount of one that nothing reports. Turn on a **watermark with allowed lateness** and the same window fires at 5 but is kept *open* for corrections; the straggler is accepted and the window emits a **correction: 5 → 6**. Same stream, same straggler; the only thing that changed is how long the window waits before it stops accepting late events. ## Provenance *Designing Data-Intensive Applications*, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 12 — *Stream Processing*, §"Handling straggler events" (and §"Knowing when you're ready to trigger a window"). The book poses the completeness problem directly: > You can never be sure when you have received all of the events for a particular > window, or whether there are some more events still to come. Its two responses are exactly the two policies here: ignore the stragglers (dropping them, and optionally tracking a dropped-events metric), or publish a correction once the late data arrives. The exercise makes a window do both. ## Prerequisites The surprise is *when a window is allowed to decide it is done*. These help; each line notes where it shows up. 1. **Tumbling windows** — the stream is chopped into fixed, non-overlapping 1-minute buckets by event_time; every event falls in exactly one. It shows up as `window start == the minute` and the `[37, 38)` bucket we watch. 2. **Event-time completeness — when to fire a window** — a windowed aggregate must eventually *emit*, but in an unbounded, out-of-order stream you can never be certain the window is complete. It shows up as the `FIRE` line and is the question that carries the whole result. 3. **Watermarks** — a heuristic assertion "no more events with event_time < *t* will arrive," letting the system decide a window is complete enough to fire. It shows up as the `watermark advances to …` lines and the `>= end` fire test. 4. **Allowed lateness / drop-vs-correct policies** — after firing, a window can be closed immediately (late events dropped) or kept open for a grace period (late events accepted, a correction re-emitted). It shows up as `CLOSE`/`DROP` under Policy A versus `LATE … accepted` / `CORRECT` under Policy B. ### Where to learn the prerequisites - **Windows, watermarks, and triggers (#1–#4):** DDIA §"Reasoning About Time" and §"Knowing when you're ready to trigger a window"; and Akidau et al., *The Dataflow Model* (VLDB 2015) — free PDF — whose *what / where / when / how* breakdown (windowing, watermarks, triggers, accumulation) is the canonical treatment. - **Watermarks specifically (#3):** the "Streaming 101 / 102" articles (oreilly.com, free) walk a watermark past a window boundary with the same drop-vs-correct fork. If only one is new, make it **#2** — "when is a window complete?" is the entire exercise; a watermark is just one heuristic answer, and the straggler is what that heuristic gets wrong. ## Environment Deterministic, so the counts reproduce exactly: - macOS 26.5.2 (Darwin 25.5.0), arm64 - Python 3.15.0a8, standard library only — no Docker, no deps - The stream and its arrival order are modeled explicitly, so the transcript is fully deterministic — the straggler always arrives at the same point - Fixed workload: 6 events truly belong to window `[37, 38)` (e1–e5 on time, s1 a straggler); the watermark is the max event_time seen so far ## Mental model A window must decide when it is **DONE** — but in an unbounded, out-of-order stream, a late event can always arrive. There is no moment you can *prove* completeness, so you pick a policy: - **Watermark** — a heuristic "no more events with event_time before *t*." Here it is a perfect arrival high-watermark: `watermark = max(event_time seen)`. When it reaches a window's end, the window fires. An event is **late** if its window's end is already ≤ the watermark — the watermark had already called that window complete. - **Drop late (allowed lateness = 0)** — the window closes the instant it fires. Any later event for it is dropped. Lowest latency, least complete. - **Watermark + allowed lateness (grace > 0)** — the window fires on time but is kept *open* for `grace` more minutes of watermark. A late event that arrives while open is accepted, and the window re-emits a **correction**. Higher completeness, at the cost of holding state longer and emitting updates. The trade is **latency vs completeness**: wait longer (or keep windows open longer) and you drop fewer stragglers but publish later. Both policies run the identical stream through one aggregator; the only difference is the grace period. The whole thing is one script, `code/straggler_watermark.py`. ## Setup ``` cd study/ddia/ch12/code python3 straggler_watermark.py ``` The stream is fixed. Nine events arrive in this processing order — note s1, a minute-37 event that shows up out of order, after the minute-39 event: ``` arrival order (processing time): e1@min37, e2@min37, e3@min37, e4@min37, e5@min37, e6@min38, e7@min39, s1@min37, e8@min40 events that TRULY belong to window [37, 38): 6 (e1 e2 e3 e4 e5 s1) ``` Do **not** read the output yet — make each prediction first. ## Steps ### Step 1 — the window fires **Predict.** Window `[37, 38)` collects e1–e5, all with event_time minute 37. Then e6 arrives with event_time minute 38, pushing the watermark to 38. The aggregator now declares window 37 complete and fires it. Six events *truly* belong to this window — so at what count does it fire? **Observe.** ``` arrive: event e1 event_time=min 37 (watermark is -) watermark advances to 37 arrive: event e2 event_time=min 37 (watermark is 37) arrive: event e3 event_time=min 37 (watermark is 37) arrive: event e4 event_time=min 37 (watermark is 37) arrive: event e5 event_time=min 37 (watermark is 37) arrive: event e6 event_time=min 38 (watermark is 37) watermark advances to 38 FIRE window [37,38) -> count 5 (watermark 38 >= 38) ``` **Fires at 5, not 6.** Only 5 of the window's 6 events had arrived when the watermark crossed the window's end. The watermark said "no more minute-37 events are coming" — and it was *wrong*, but the window had no way to know that. It fires optimistically at 5. ### Step 2 — drop-late: the straggler arrives **Predict.** Now s1 arrives with event_time minute 37 — but the watermark is already at 39, so its window is in the past. Under the default **drop-late** policy (the window closed the moment it fired), what happens to s1, and what is window 37's final published count? **Observe.** ``` CLOSE window [37,38) (watermark 38 >= 38+0) -- stops accepting late events arrive: event e7 event_time=min 39 (watermark is 38) watermark advances to 39 FIRE window [38,39) -> count 1 (watermark 39 >= 39) CLOSE window [38,39) (watermark 39 >= 39+0) -- stops accepting late events arrive: STRAGGLER s1 event_time=min 37 (watermark is 39) DROP s1: window [37,38) is CLOSED (watermark 39 >= 38+0) -- late event ignored, published count stays 5 arrive: event e8 event_time=min 40 (watermark is 39) watermark advances to 40 FIRE window [39,40) -> count 1 (watermark 40 >= 40) CLOSE window [39,40) (watermark 40 >= 40+0) -- stops accepting late events window [37, 38) FINAL published count = 5 events that truly belong there = 6 >> UNDERCOUNT: published 5, actual 6 -- the straggler was dropped, silently, with no error. ``` **Published 5, actual 6 — dropped with no error.** The window closed the instant it fired (`watermark 38 >= 38+0`), so when s1 shows up it hits a closed window and is discarded. No exception, no warning; window 37 is permanently one short. ### Step 3 — watermark + allowed lateness: the straggler is recovered **Predict.** Same stream, same straggler — but now the window carries a grace of 2 minutes: after firing it stays *open* until the watermark passes its end plus the grace. When s1 arrives (watermark 39), window 37's end + grace is 38 + 2 = 40, still ahead of the watermark. What does the window do with s1? **Observe.** ``` FIRE window [37,38) -> count 5 (watermark 38 >= 38) arrive: event e7 event_time=min 39 (watermark is 38) watermark advances to 39 FIRE window [38,39) -> count 1 (watermark 39 >= 39) arrive: STRAGGLER s1 event_time=min 37 (watermark is 39) LATE s1 accepted into still-open window [37,38) (watermark 39 < 38+2) CORRECT window [37,38): published 5 -> 6 arrive: event e8 event_time=min 40 (watermark is 39) watermark advances to 40 FIRE window [39,40) -> count 1 (watermark 40 >= 40) CLOSE window [37,38) (watermark 40 >= 38+2) -- stops accepting late events ``` **Corrected to 6.** The window fired on time at 5, exactly as before — but this time it did not close. Because `watermark 39 < 38+2`, window 37 was still open when s1 arrived, so s1 was accepted and the window re-emitted a correction, `5 -> 6`. Only when e8 pushes the watermark to 40 (`>= 38+2`) does window 37 finally close. ### Step 4 — side by side **Predict.** Same nine events, same straggler, two policies. How many distinct final published counts for window 37 will the two policies produce? **Observe.** ``` Same stream, same straggler: drop-late published 5 (undercount of 1) watermark published 6 (correct) ``` **Two different answers — 5 and 6 — from the identical stream.** The straggler never changed; what changed is whether the window was still willing to hear it. Drop-late traded completeness for latency and lost the event; the watermark with allowed lateness held the window open one more beat and recovered it. ## What you should see - Under **both** policies, window 37 **fires at 5** — the 5 on-time events, the moment the watermark crosses minute 38. - Under **drop-late**, the minute-37 straggler hits a closed window and is **dropped**: final published **5**, actual **6**, undercount of **1**, no error. - Under **watermark + allowed lateness**, the straggler lands in a still-open window and the count is **corrected to 6**. - Same stream yields **two** distinct published counts: **5** (drop-late) and **6** (watermark). ## Why In an unbounded, out-of-order stream you can never be *certain* a window is complete: at any moment a delayed event with an old timestamp might still be in flight. So a windowed aggregator cannot wait for proof — it must pick a policy for deciding "complete enough," and every policy is a bet. The watermark is that bet made explicit: "no more events before *t*." When it crosses a window's end the window fires, and it fires with whatever has arrived — here, 5. The straggler is dropped *precisely because* the window already closed on an optimistic watermark: the assertion "no more minute-37 events" turned out false, but the window had already acted on it. Nothing errored because dropping late data is not a failure in this model — it is the policy, working as designed. Allowed lateness changes the bet without changing the watermark. The window still fires on time (same latency for the common case), but instead of closing it keeps its state and stays open for a grace period, so a late arrival is folded in and the result re-emitted as a correction. You have moved along the **latency vs completeness** curve: same fast first answer, plus a slower, more-complete follow-up, paid for by holding window state longer and by making every downstream consumer tolerate updates. The correction only helps if the straggler arrives *within* the grace — a later straggler still hits `CLOSE` and is dropped, which is why the grace is a tuning knob, not a guarantee. **The boundary — no out-of-order arrival, or an infinite wait, and there is no straggler.** If events arrived in perfect event-time order, the watermark would never overtake an unarrived event, so nothing would ever be late and both policies would agree. And if you waited *forever* before firing (an infinitely late watermark) you would drop nothing — but never emit, which for an unbounded stream means never producing a result. The undercount bites only in the realistic middle: out-of-order arrival plus a *finite* wait. That is the regime every real streaming system lives in, which is why watermarks, allowed lateness, and corrections exist at all. ### Go deeper 1. **The trigger/accumulation matrix.** Map this exercise onto the Dataflow model: the watermark is the *trigger*, drop-vs-correct is the *accumulation mode* (discarding vs accumulating), and the 1-minute bucket is the *window*. Add an early trigger (fire before the watermark, on every event) and watch window 37 emit 1, 2, 3, 4, 5 partial counts, then a correction to 6. 2. **The processing-time sibling.** This is the event-time half of the event-vs-processing-time gap; build (or revisit) a ch12 exercise that windows the *same* stream by processing time and watch the straggler land in the wrong window entirely — a different failure from the same root cause. 3. **Shrink the grace to zero, widen it to infinity.** Set allowed lateness to 0 and confirm Policy B degenerates to drop-late; set it larger than the straggler's delay and confirm the correction always lands — then reason about what an unbounded grace costs in retained state.