Event Time, Watermarks & Checkpoints

Part 0. Start here

A replay that counts one click twice, although no write adds anything

It is Super Bowl night, and an ad network is counting 100,000 clicks a second. One worker of the stream job that counts them dies. The job restores its last checkpoint (a saved copy of the job's state and of its read position in every input) and replays the records that came after it. Every write to the sink (the database the job writes its results to) is an absolute SET, never an ADD, and the job's state is exactly-once (after a restore, every input record has affected the state exactly one time).

Yet one ad's hour row now counts a click twice: once in fast_valid (the minute it belongs to) and once in late_valid (the late total for its hour). Nothing was added twice. The replay simply decided differently about that click.

The numbers

The replay is not small, and it is not rare: every crash, rescale and upgrade causes one.

• How much is replayed. The job checkpoints every 30 s and takes about 50 s to restart (the ad-click loop's assumption). If the last checkpoint completed on time, a crash replays up to 30 + 50 = 80 s of clicks; at the Super Bowl rate that is 80 × 100,000 = 8,000,000 clicks. On an average day the loop's trace replays 70 s × 11,574 clicks/s ≈ 810,000. When checkpoints stall, the replay is longer. The small example on this page replays 40 s of input after a 50 s restart: 40 + 50 = 90 s.
• A second problem occurs in the first 200 ms after a restore. After a restore, the job's watermark (its estimate of how far event time has progressed, Part 2) has no value, and it is recomputed only every 200 ms. The restored job reads its backlog at its catch-up rate, about 64,000 clicks/s (the loop's 32 KPUs × ~2,000/s; a KPU, or Kinesis Processing Unit, is the managed service's unit of 1 vCPU and 4 GB of memory, Part 12), so 64,000 × 0.2 = 12,800 clicks are judged with no watermark at all. If 2% of them belong to minutes that closed before the checkpoint, that is 12,800 × 0.02 = up to 256 clicks that each re-open a finished minute (several can share a row), and each such minute's row is overwritten with a count of about 1 (say 812 → 1). If one quiet shard holds the watermark down for the whole 30 s idle timeout, it is 64,000 × 30 × 0.02 = 38,400, up to about 38,000 rows. (An estimate at the loop's normal-day sizing; on a Super Bowl night at 64 KPUs the rate doubles, and so do these numbers. Not a measurement.)
• What the cleanup costs. The loop's cleanup after a restore rewrites about 300,000 keys at 2,800 write units a second: 300,000 ÷ 2,800 ≈ 107 s, about 2 minutes (Part 8 shows how to make it smaller).

The big picture

Synthesizing vector architecture diagram...

What you'll be able to do after this page

1. Tell event time from processing time, and name the three questions every streaming design answers (Part 1).
2. Compute a job's watermark across shards, and say what idleness buys and costs (Part 2).
3. Choose tumbling, sliding or session windows, and build hour rows from closed minutes without double counting (Part 3).
4. Choose between allowed lateness and a side output, and make early results safe for their consumers (Part 4).
5. Say what lives in keyed state, why event-time timers (not TTL) clear dedup state, and what a stuck watermark does (Part 5).
6. Walk a checkpoint barrier through the job, aligned and unaligned, and say what backpressure does to it (Part 6).
7. List what a restore brings back and what it doesn't, and explain why a replay can decide differently (Part 7).
8. Make a sink safe after a restore: per-writer tags, the flush, the closed-row condition, or a transactional sink with timeouts that fit (Part 8).
9. Rescale and upgrade a job without losing or duplicating output (Part 9).
10. Compare continuous streams, micro-batches and a batch recount on equal terms (Part 10).
11. Trace one click through every layer, and name the four numbers to watch (Part 11).
12. Map all of it to AWS (Part 12).

You may have arrived from step 2.2 ("When Is a Minute Complete?") or 2.4 ("A Worker Crash Replayed a Minute and Double-Billed") of the ad-click aggregation loop, step 2.5 of the metrics and alerting loop ("Late and Out-of-Order Samples Are Rejected"), or step 2.5 of the YouTube loop ("One Counter Item per Video Is Throttled on Viral Videos"). This page is the "why" behind all of them.

Part 1. The core mental model: two clocks

A click happens at one moment and reaches the job at another. Everything on this page follows from keeping those two moments apart.

Two clocks

Take one click from our example, c15. A phone clicked an ad at 10:13:40, lost signal, and uploaded the click at 10:15:04. The job reads it at 10:15:04.

Why processing time moves clicks

Suppose the job falls 2 minutes behind (a slow sink, a restart). With processing-time minutes, every click read during the catch-up lands in the minute the job happened to read it, not the minute it happened. The same input gives different counts depending on the job's health. In Part 3 we count the same clicks both ways: processing-time minutes give one answer before the crash and another after it, and neither is the event-time answer.

Three questions every streaming design answers

Until the third question's answer has run out, every result is provisional.

Other meanings of the word "watermark"

The word "watermark" means other things in the loops. They share the word, not the mechanism:

• Kafka's high watermark: the offset up to which a partition's messages are replicated (Message Queues vs Event Streams). Nothing about event time.
• Read-receipt watermarks in chat: a per-conversation "read up to here" cursor.
• A scheduler's watermark: the last fully processed time bucket, advanced by a job.
• A watermark on a video: a mark on the picture.

On this page a watermark is always an event-time progress marker.

The running example: one minute of clicks from three shards

Real jobs read billions of clicks, too many to watch. So one small story runs through the whole page: one minute of clicks from three shards. Clicks on two ads, ad_7 and ad_9, of one campaign, for the event-time minute 10:14, read by one job from a Kinesis stream with three shards, S1, S2 and S3 (a shard is one ordered sequence of records inside the stream; the job reads each shard separately). The job runs as operators (processing steps, such as the pre-aggregation or the window), and each operator runs as several subtasks (parallel copies of the operator, each handling part of the data). Every trace on this page was produced by running a private reference implementation of this job, not worked out by hand.

Before the story starts, each shard has already read some clicks; among them are six clicks on ad_7 in minute 10:13 (so minute 10:13 will close with 6), and S3's last one has event time 10:13:52. The clicks:

(Click numbers follow a first draft of the story, so c6 isn't used; c14 and c16 are live traffic after the restart.)

The story in seven beats: the clicks arrive out of order (Parts 1 and 2), a shard goes quiet (Part 2), the minute closes (Part 3), two clicks come too late (Part 4), the checkpoints are taken (Parts 5 and 6), a worker crashes and the replay decides differently (Parts 7 and 8), and the job is rescaled and upgraded (Part 9). Each Part shows only its slice; the full table is in Part 13.

Part 2. Watermarks

A window must close sometime, so its row can be written and its memory freed. Closing by the wall clock fails after any backlog; waiting forever never writes anything. We need a way for the data itself to say "event time has got this far".

What a watermark promises

A watermark is a timestamp that flows through the job with the records and says: "event time has reached X; I don't expect anything older." It is a promise the job makes to itself, based on what it has read. When the watermark passes the end of a window, the window closes.

One shard

Each shard's reader tracks the highest event time it has read and, every 200 ms (Flink's default pipeline.auto-watermark-interval), emits a watermark of highest event time − allowance. Records in a shard are a little out of order (phones, retries, batching), and the allowance (5 s here) is how much disorder we wait for. This strategy is called bounded out-of-orderness.

Flink subtracts one more millisecond: its generator emits highest − allowance − 1 ms. So after c1 (10:14:05) on S1, S1's watermark is 10:13:59.999. The millisecond makes the arithmetic come out cleanly: a window covers [start, end), its last millisecond is end − 1 ms, and it fires when the watermark reaches that last millisecond. So minute 10:14 closes exactly when every active shard has read an event time of 10:15:05 or later.

Three shards and the minimum

A job reading three shards can't take any single shard's watermark: S1 may be at 10:14:53 while S3 still has clicks from 10:14:15 in flight. So the job's watermark is the minimum over the shards, and it only moves forward:

From the reference implementation (watermarks shown with Flink's − 1 ms; "tick" is the 200 ms emission that makes a change visible):

Two things to see. At event 2, minute 10:13 was held open by S3 alone (its last record was 10:13:52, so its watermark was 10:13:46.999); c3 lifted S3, the minimum became S1's 10:13:59.999, which is minute 10:13's last millisecond, and the minute fired at once. And at event 11, minute 10:14 closed at 10:15:07.2, the first 200 ms tick after c11 was read at 10:15:07.1: min(10:15:00.999, 10:15:01.999) = 10:15:00.999, past 10:14:59.999.

Synthesizing vector architecture diagram...

Snapshot W2, 10:15:07.2: S3 is idle, so the minimum covers S1 and S2 only; it passes minute 10:14's last millisecond and the minute closes with 5 and 2. (W1, the checkpoint about 6 seconds earlier, is drawn in Part 6.) The steps, in time order: in the Shards panel, S1 has read up to event time 10:15:06, so its watermark is 10:15:00.999, and S2 has read up to 10:15:07, so its watermark is 10:15:01.999. S3 has had no records since it was marked idle at 10:14:50.8, so its dashed arrow means it is left out. The Job watermark panel takes the minimum of S1 and S2, 10:15:00.999. That value is past 10:14:59.999, the last millisecond of minute 10:14, so the Keyed state panel's two window counts, ad_7 = 5 and ad_9 = 2, are emitted, and the Sink panel receives SET writes of those counts, tagged fast_ckpt 41. The keyed state also holds the late total for ad_7, hour 10, which c15 created at 10:15:04.0. The Checkpoint storage panel holds c41, the last completed checkpoint, with its floor of 10:14:34.999.

Out of order is not late

c8 was clicked at 10:14:55 and read at 10:15:05.0, 3 s behind c9 (10:15:02), which S2 had already read. It is out of order. But it is not late: a record is late only when the watermark has already passed the end of its window. At 10:15:05.0 the job watermark is 10:14:52.999, and c8's minute ends at 10:15:00, so c8 joins minute 10:14: ad_7 goes from 4 to 5, and Valkey shows 5 at the next once-a-second firing (10:15:06).

A quiet shard

S3 carries phone uploads, and after c3 it is silent. Without help, its watermark would sit at 10:14:14.999 and hold the whole job there.

Flink's withIdleness works this way, and its Kafka connector generates watermarks per partition and combines them the same way. (When every input is idle at once, recent Flink versions emit the highest of their watermarks; our story never lets that happen, and your design shouldn't rely on it.)

Clamping the device's time

Event time comes from the device. A phone whose clock says 2031 would lift its shard's watermark by years. The ad-click loop clamps at ingest: event_time = clamp(device_time, impression_time, arrival_time) (impression ≤ event time ≤ arrival), and keeps the raw value for fraud analysis.

From the reference implementation, c7's phone reporting a time in 2031, no crash:

A watermark is an estimate, and it can be wrong

A watermark is a heuristic (a rule that is usually right, not a guarantee). The Dataflow paper (Akidau et al., 2015) names both ways it fails: too fast (late data exists after it passed) and too slow (one slow source holds everyone back). A bigger allowance trades delay for completeness, but the loop's clicks that are really late are minutes late, so past a few seconds a bigger allowance buys almost nothing. Flink can also align watermarks: pause reading sources that run too far ahead of the slowest one, so a fast source doesn't fill memory with windows that can't close yet.

Watching the watermark

Watermark lag = wall clock − job watermark. At 10:15:07.2 it is 10:15:07.2 − 10:15:00.999 ≈ 6.2 s: the 5 s allowance plus about a second of disorder and emission timing. If it keeps increasing, the watermark has stopped advancing, for one of these reasons: a shard with no data and no idleness timeout, a reader that has stopped reading, or an operator that has stopped processing. The ad-click loop publishes it as data_complete_through and alarms when the lag exceeds 60 s for 5 minutes. It is a different number from consumer lag (how far behind the stream's tip the reader is): a job can be fully caught up with a stuck watermark, or far behind with a healthy one.

Part 3. Windows

Records now carry event times and the job knows how far event time has got. Next: what to count over. Reports, rolling fraud rules and user sessions need three different shapes, and each has a trap.

Tumbling: the minute closes

A tumbling window is a fixed, non-overlapping bucket: [10:14:00, 10:15:00), then [10:15:00, 10:16:00). Each record belongs to exactly one, found by rounding its event time down to the minute.

pseudocodeCopy
TUMBLING WINDOW (per ad, size 1 minute, allowed lateness 0)
    1. window = floor(event time / 1 minute)
    2. add the record to that window's running count        -- one number per ad and window
    3. register an event-time timer at the window's last millisecond
    4. when the watermark reaches the timer:
        emit the count (SET fast_valid)
        add it to the hour roll-up
        delete the window's state

This procedure counts clicks per ad in fixed one-minute windows and writes each minute's count once the watermark shows that the minute is complete. event time is the time the click happened, carried in the click itself, and window is that time rounded down to the minute (for c12, at 10:14:57, the window is 10:14). The window's running count is one number per ad and window. The procedure registers an event-time timer at the window's last millisecond (10:14:59.999 for the 10:14 window). When the watermark (the job's estimate that no earlier events are still coming) reaches the timer, the procedure writes the count to the sink with SET fast_valid, adds it to the hour roll-up and deletes the window's state. Allowed lateness is 0, so a click whose window has already fired goes to the late path instead. The result is one row per ad and minute (ad_7 10:14 = 5, written at 10:15:07.2) and freed state.

Event 11 from the reference implementation:

An aggregate function (a count, a sum) keeps one number per window; a function that needs every record of the window at the end (Flink's ProcessWindowFunction alone) keeps them all. At 100,000 clicks a second, that difference is the state size.

The hour from closed minutes

Hour rows must equal the sum of their minutes. The obvious design is a second, 1-hour window.

Sliding windows

A sliding window has a size and a slide: 60 s, every 10 s. A fraud rule like "more than 5 clicks from one device in a minute" needs it, because tumbling minutes miss a burst that straddles a boundary.

From the reference implementation, device d_5 clicks ad_7 at 10:14:40, :48, :55, 10:15:02, :09 and :15 (six clicks in 35 s):

The cost: each click belongs to 60 ÷ 10 = 6 windows, so state and output are 6 times larger (or the job keeps a count per 10 s slice and adds the last 6 slices).

Session windows

A session window groups a key's activity separated by gaps. Each record opens [t, t + gap); overlapping windows of one key merge; the session fires when the watermark passes its end. Device d_8, gap 30 s, from the reference implementation:

A late or out-of-order element can join two sessions into one. Sessions can stay open as long as activity continues.

Synthesizing vector architecture diagram...

Processing-time windows

A processing-time window buckets by the job's wall clock. The same clicks, from the reference implementation:

The event-time answer for minute 10:14 is ad_7 5 and ad_9 3 (Part 8), with c15 in 10:13. Processing-time counts depend on lag and on crashes, so they are fine only for metrics about the pipeline itself (records read per minute), never for billing.

Pre-aggregation without losing clicks

The ad-click loop's hot ad (its most-clicked ad) gets 40,000 clicks a second, and the shuffle by ad (the step that sends each click to the subtask that owns its ad) would send them all to one subtask. So each upstream subtask pre-aggregates: it adds up its own clicks per (ad, minute) for 500 ms and forwards one partial sum. The window adds partials just as it would add clicks. Two rules keep this exact (a third, at the checkpoint barrier, is in Part 6):

pseudocodeCopy
PRE-AGGREGATE (per upstream subtask, before the shuffle by ad)
    on each click:
        1. if its window ended at or before the current watermark:   -- decided per click
            send the click, alone, to the late path
            never fold it into a partial
        2. else:
            add +1 to the buffered partial for (ad, window)
    every 500 ms:
        forward the buffered partials
    on a new watermark:
        3. forward every buffered partial FIRST
           then forward the watermark

This procedure lets each upstream subtask add up its own clicks before the shuffle by ad, so the hot ad's 40,000 clicks a second reach its window subtask as a few partial sums, without losing a click or counting a late one as on time. A partial is the buffered count of clicks for one ad and one window, where the window is the click's one-minute window. For each click, the procedure first compares the end of its window with the current watermark: if the window ended at or before the watermark, the click is late and is sent alone to the late path. Otherwise it adds 1 to the partial for that ad and window. Every 500 ms the buffered partials are forwarded to the window. When a new watermark arrives, every buffered partial is forwarded first and the watermark after it. The result is partial sums that the window adds just as it adds single clicks, and no late click hidden inside a sum.

Rule 1: a partial is a sum, so once a late click is folded in, nobody downstream can separate it. Rule 3: without it, a partial can reach the window after the watermark that closed the window.

Branch P1 (from the reference implementation): c8 is read at 10:15:06.9 instead of 10:15:05.0. It is on time (its minute ends after the watermark 10:14:56.999), so its +1 sits in the 500 ms buffer when c11 lifts the watermark at 10:15:07.2.

Part 4. Late data and early answers

Minute 10:14 is closed and written. Two clicks came too late for their minutes, and dashboards wanted numbers long before the minute closed. Both are about the same thing: results that change after we first showed them.

Two late clicks

Both late totals are new attributes written after checkpoint c41. Remember that; Part 8 is about exactly these values. This is snapshot W3, 10:15:20.0:

Allowed lateness or a side output

A late record must go somewhere. Flink's default allowed lateness is 0: a late record is dropped, unless you route it. Two ways to keep it:

• Allowed lateness: keep each window's state for a while after it closes; a late record inside that time updates the window and fires it again. Flink's documentation says late firings "should be treated as updated results of a previous computation".
• A side output: a second output of the operator for records late for their window (Flink's sideOutputLateData). A separate operator keeps late totals; the minute row is written once.

Branch A (from the reference implementation), allowed lateness 60 s, no crash:

The ad-click loop picks the side output because its late clicks come minutes late: 45 minutes of allowed lateness would keep 45 times the window state.

The acceptance horizon and the recount

Late can't mean "forever". The acceptance horizon (2 hours after the impression in the ad-click loop) is the point after which a record is rejected as stale; the section Streams: exactly-once state plus idempotent sinks of Idempotency & Effectively-Once Processing states the rule and ties dedup expiry to it. Inside the horizon, the stream's numbers are provisional, and an exact batch recount from the lake writes final_* attributes once the hour can no longer change. The S3-like object storage loop's usage metering does the same with a day: a late batch "marks that hour dirty", and "hours are final after a day".

Synthesizing vector architecture diagram...

Early firings

The ad-click dashboards need numbers within 3 seconds, not when the minute closes. A trigger decides when a window emits; here it fires every second of processing time with the running total (an early firing), again at the watermark, and the side output handles anything later. From the reference implementation, the running totals Valkey received for minute 10:14 up to 10:14:41:

Part 5. Keyed state and timers

Windows, dedup, roll-ups and late totals all need memory that belongs to a key, survives restarts, and is cleared at the right moment. How it is cleared turns out to matter as much as how it is kept.

State that belongs to a key

Keyed state is memory scoped to the current key: when the job processes a click for ad_7, it sees only ad_7's entries. It is partitioned by key (Part 9 shows how), so each entry lives on exactly one subtask. Operator state belongs to a subtask instead, whatever keys pass through it.

Event 3: at 10:14:45.0, c2′ arrives on S1 with c2's click_id. Dedup state already holds c2, so c2′ is dropped as a duplicate and written to the exceptions log. Minute 10:14 for ad_9 stays at 1 until c7.

Timers on two clocks

A timer asks the job to call back at a time. Event-time timers fire when the watermark passes them: a window closes, a dedup id expires. Processing-time timers fire on the wall clock: the once-a-second early firing, the 500 ms buffer.

Dedup ids are cleared by an event-time timer at impression + 2 h, the acceptance horizon. For c1: impression 10:13:55 + 2 h = 12:13:55, in event time. However far behind the job falls, the timer can't fire before the watermark reaches 12:13:55, so every copy the horizon still accepts finds its original in state.

Why TTL can't be the dedup window

Flink also has state TTL: an entry expires a set time after it was last written. But Flink's TTL counts processing time only, and expired entries are removed lazily on read and in the background (for RocksDB, the on-disk state store described under Where state lives below, during compaction). So a 2-hour TTL removes a click id 2 hours after the job saw it, however far behind the job is; after a long outage, copies that the horizon still accepts find nothing and are counted twice. The section Streams: exactly-once state plus idempotent sinks of Idempotency & Effectively-Once Processing traces exactly that. Keep a much longer TTL only as a guard (the loop uses 26 h) in case the watermark gets stuck.

Where state lives

Keyed state bigger than memory lives in RocksDB (an embedded key-value store that keeps its data in files on the worker's local disk): reads go through its caches and Bloom filters, and snapshots copy its immutable files (the LSM-Trees & Compaction loop primitive explains how RocksDB stores and compacts them). The ad-click loop's numbers:

That local disk is shared by the state itself, RocksDB compaction, and the files each checkpoint stages for upload.

Part 6. Checkpoints

The job's state is spread over many subtasks, all changing thousands of times a second, and its input is three shards moving independently. To restart after a crash we need a snapshot of all of it that matches one moment of the input, taken without stopping the job.

A consistent cut

A checkpoint saves every operator's state and every source's read position, such that the state reflects exactly the records before those positions. After a crash, the job loads both and replays from the positions; every record then affects the state exactly once. This is the same rule as a write-ahead log's "log first, then act" (Write-Ahead Log, fsync & Group Commit): the state and the position that produced it are saved together, and the metrics loop's ingesters follow it by hand (upload the block, then checkpoint the stream position). Flink's method comes from the Chandy-Lamport snapshot algorithm.

Barriers, step by step

pseudocodeCopy
CHECKPOINT n (aligned)
    1. the coordinator tells every source: inject barrier n
    2. each source:
        save its read position
        send barrier n downstream, in line with its records
    3. each operator with several inputs:
        wait until barrier n has arrived on ALL of its inputs
        meanwhile, hold back records from the inputs that already delivered barrier n
    4. then the operator:
        snapshot its state
        forward barrier n
    5. snapshots are uploaded to checkpoint storage in the background
    6. when every task has acknowledged:
        checkpoint n is complete
        operators are notified

This procedure takes a consistent snapshot of a running job: every operator's state together with every source's read position, so that the state reflects exactly the records before those positions. n is the checkpoint id (41 for c41). The coordinator (the part of Flink's JobManager, the job's controlling process, that starts and tracks checkpoints) asks every source to inject barrier n, a marker that travels in the stream and separates the records before checkpoint n from the records after it; each source saves its read position and sends the barrier downstream in line with its records (in c41, after c7 on S1, c5 on S2 and c3 on S3). An operator with several inputs waits until barrier n has arrived on all of them, and holds back records from inputs that already delivered it; this wait is the alignment time. Then it snapshots its state and forwards the barrier. Snapshots are uploaded to checkpoint storage in the background, and when every task has acknowledged, checkpoint n is complete (10:15:01.5 for c41) and the operators are notified. The result is a stored checkpoint from which the job can restore and replay.

Synthesizing vector architecture diagram...

Event 6, checkpoint c41, from the reference implementation:

Synthesizing vector architecture diagram...

Snapshot W1, 10:15:01.5: everything the restore in Part 7 will come back to. Note what is not in the Checkpoint storage panel: the watermarks themselves, the idle flag on S3, and anything in the sink. The panels, in the order the values flow: the Shards panel shows S1's watermark 10:14:52.999 after c7, S2's 10:14:34.999 after c5, and S3 idle. The Job watermark panel therefore shows the minimum of S1 and S2, 10:14:34.999. The Keyed state panel holds the window counts at the barrier, ad_7 = 4 and ad_9 = 2 for minute 10:14, and the dedup ids c1 to c7. The dashed arrow is the snapshot: the Checkpoint storage panel receives the read positions, that state and the floor 10:14:34.999, and c41 is complete at 10:15:01.5. The Sink panel still holds only minute 10:13 of ad_7 = 6, tagged fast_ckpt 39, because minute 10:14 has not closed yet.

Buffers outside state: flush at the barrier

The pre-aggregator's 500 ms buffer is memory, not state. If a barrier passes while a partial sits in it, the snapshot has neither the partial nor the click (the source position is already past the click, and dedup already remembers its id).

Branch P2 (from the reference implementation): c7 is read at 10:14:59.9, so its +1 for ad_9 is still buffered when barrier 41 arrives at 10:15:00.0.

So the pre-aggregation rule's third line: on a barrier, forward every buffered partial before snapshotting (or snapshot the buffer itself). Any buffer outside checkpointed state needs a rule like this.

When the sink is slow

In the toy the queue holds only other ads' output; in the real job it holds thousands of records, and an aligned barrier must wait for all of them.

Unaligned barriers

An unaligned checkpoint lets the barrier overtake the queued records: the operator forwards the barrier at once and stores the records it overtook (the in-flight data) as part of its snapshot; on restore they are replayed from the checkpoint. Flink requires exactly-once checkpointing for it and doesn't take unaligned checkpoints concurrently. The setting is execution.checkpointing.unaligned.enabled (the older name, execution.checkpointing.unaligned, still appears in the managed service's list of settings).

Branch U (from the reference implementation): same run, but c42 is unaligned.

Incremental uploads, interval, pause and timeout

• Incremental checkpoints upload only the RocksDB files created since the last checkpoint, not the whole state. At the loop's scale, about 11,574 clicks/s × 30 s × 60 B ≈ 20.8 MB of new dedup state per checkpoint, plus whatever compaction rewrote, instead of 6 to 43 GB.
• Interval (how often a checkpoint starts), minimum pause (the gap after one completes before the next may start) and timeout (when an unfinished checkpoint is abandoned; 10 minutes by default in Flink). Managed Service for Apache Flink's defaults are a 60,000 ms interval and a 5,000 ms minimum pause; the ad-click job sets 30 s.
• What a failure costs: everything since the last completed checkpoint, plus the restart: 30 + 50 = 80 s when checkpoints complete on time.
• Checkpoints complete even if the sink is "ahead" of them: output is not part of a checkpoint.

Savepoints

Part 7. Crash, restore and replay

The worker died at 10:15:40.0, and the last completed checkpoint is c41. A restore must bring the job back to a state that matches its input, and then catch up. What it brings back, and what it doesn't, decides whether the output stays right.

What comes back, and what doesn't

The replay, from the reference implementation:

The replay makes a different lateness decision

Synthesizing vector architecture diagram...

Lateness is not a function of the input alone. It depends on:

1. How records from different shards interleave when they reach the operator. In run 1, S1 and S2 had moved on by the time c12 arrived; in the replay, every shard's backlog arrives together.
2. Idleness, which is processing time: S3 was idle in run 1 and active in the replay.
3. When watermarks are emitted: every 200 ms of wall clock, which a replay doesn't reproduce.
4. The watermark itself, which restarts from nothing.

None of these is in the checkpoint. So each run makes a valid, but possibly different, split into on time and late. Two consequences: the first run's decisions left values in the sink that the replay won't rewrite (Part 8), and a window that closed before the checkpoint can re-open (next).

A closed minute re-opens

Right after the restore, the pre-aggregator has no watermark until every shard reports one. Branch W (from the reference implementation): the same restore without a floor.

Had hour 10 already closed before the checkpoint, the re-created minute would also have re-created the hour's roll-up, and its row would have been written with fast_valid = 1 over the whole hour.

How long is the exposure? Until every shard has reported a watermark. In the main line all three shards had backlog, so the first emission at 10:16:30.2 covered them: 200 ms. In branch U (restore from c42), S3 had nothing to replay, and the job's watermark stayed at the minimum value from 10:16:30.0 until S3 was marked idle at 10:17:00.4: 30.4 s, the whole idle timeout (the timer starts at the first check that sees no records, 10:16:30.2). That is the case in Part 0 that re-opens up to about 38,000 rows.

The watermark floor

pseudocodeCopy
WATERMARK FLOOR (in the operator that decides lateness, before the window)
    on a new watermark w:
        floor_candidate = w
    on a barrier:
        save floor_candidate in operator state   -- union list state
    on restore:
        floor = the MINIMUM of the saved values   -- safe after rescaling
    on each record:
        judge = max(floor, current watermark)
        if the record's window ended at or before judge:
            late path (and the horizon check uses judge too)
        else:
            on to the window

This procedure keeps a restored job from judging records against a lower watermark than the one the job had reached at the checkpoint. w is each new watermark the operator receives, and floor_candidate holds the latest one. At each checkpoint barrier the operator saves floor_candidate in its state as union list state (Flink keeps every subtask's value and gives the whole list to every subtask on restore). On restore, floor is the minimum of the saved values, which is safe even when the number of subtasks changed (10:14:34.999 after restoring c41). For each record, judge is the larger of floor and the current watermark; a record whose window ended at or before judge goes to the late path, and the acceptance-horizon check uses judge as well. Every other record goes on to the window. The result is a lateness decision per record that never uses a watermark below the checkpoint's.

It lives in the operator that decides lateness before the window: in the ad-click job, the pre-aggregation operator (the loop's step 2.4, item 3a); in a job without one, a small filter right before the window. Flink's built-in window operator can't apply it, because it judges lateness against its own input watermark; a custom keyed process function that does the windowing itself can. Timers need no floor: after a restore they can only fire late, never early. The acceptance-horizon check (the section Streams: exactly-once state plus idempotent sinks of Idempotency & Effectively-Once Processing) uses the same max(floor, watermark). The floor is a design choice, not a Flink feature.

External state isn't rolled back

The restore rewinds Flink's state, not the world's. If dedup ids lived in an external Redis instead of keyed state, Redis would still hold c9 … c13 from run 1; the replay would find them there and drop them as duplicates, and the restore would under-count (the ad-click loop's step 2.6).

Part 8. Sinks: making repeated and changed output harmless

Everything the job wrote between barrier 41 and the crash is written again by the replay, and some of it differently. The sink must end up as if only the replay had run. This Part is the mechanism behind the rule that Idempotency & Effectively-Once Processing states: effectively-once = exactly-once state + a sink that makes repeats harmless.

What a replay does to output written after the checkpoint

What the replay can do to each kind of value, if nothing else is done:

Absolute values per key

Every write sets an absolute value per key: SET fast_valid = 5 on ad_7 / M10:14, never ADD. That handles the first row. For the others, the sink needs to know which values were written after the checkpoint.

Each writer tags its own attributes

Several writers share one item. The hour row carries the stream's fast_* (the roll-up), the late path's late_* and the recount's final_*. So a tag per item isn't enough: after a restore we must remove the late path's values without touching the recount's. Each writer tags only its own attributes, with the same names as the ad-click loop:

pseudocodeCopy
TAGGED WRITE (late path)
    UpdateItem ad_metrics key (entity, bucket)
        SET late_valid = :n, late_spend_micros = :s, late_ckpt = :epoch, job_gen = :g,
            late_ix = ":epoch#:shard"                       -- :shard = hash(entity, bucket) mod 8
        CONDITION attribute_not_exists(job_gen) OR job_gen <= :g

FULL FLUSH (after restoring checkpoint c, before any new output)
    for each epoch e >= c:
        for each shard n in 0..7:
            query each writer's tag index for "e#n"
                minute item, fast_ckpt >= c:
                    delete the item   -- its window is open or absent in restored state
                hour item, late_ckpt >= c:
                    if restored state holds that late total:
                        rewrite it with tag c
                    else:
                        REMOVE late_valid, late_spend_micros, late_ckpt, late_ix
                hour item, fast_ckpt >= c:
                    REMOVE the fast_* attributes, their tag and fast_ix
    never delete an item that carries another writer's attributes
    rewrite Valkey's running totals from restored state
    delete totals written since c that state lacks
    then replay

These two procedures let the sink find and undo every value written after the checkpoint that a job restores from. In the tagged write, the late path writes one hour row of the ad_metrics table, keyed by entity (the ad) and bucket (the hour, such as H10). :n and :s are the late click count and the late spend in micros; :epoch is the id of the last checkpoint barrier the sink subtask passed, stored in late_ckpt; :g is the job generation, stored in job_gen; late_ix is the index key ":epoch#:shard", where :shard is hash(entity, bucket) mod 8, a number from 0 to 7. The condition refuses the write if the row already carries a higher job_gen. In the full flush, c is the restored checkpoint id (41). For every epoch at or above c and every shard 0 to 7, the flush queries each writer's tag index: it deletes minute items tagged at or above c, rewrites or removes the late totals on hour items depending on whether restored state holds them, and removes tagged fast_* attributes from hour items, and it never deletes an item that carries another writer's attributes. It then rewrites the Valkey running totals from restored state and deletes the ones that state lacks. The result is a sink that holds only values written from state at or before the checkpoint; the replay then writes the rest.

The full flush, and the index that finds it

Event 18, from the reference implementation, at 10:16:30.0:

Synthesizing vector architecture diagram...

Snapshot W4, 10:16:30.0: the sink now holds nothing the first run decided after c41. The minute row finished before the checkpoint (M10:13) is untouched. The arrows start at the Checkpoint storage panel, because at this moment everything comes from the restored c41. It sets the Shards panel back to the saved positions: S1 after c7, S2 after c5, S3 after c3. It gives the Keyed state panel the counts at the barrier, ad_7 = 4 and ad_9 = 2 for minute 10:14, and no late totals. The Job watermark panel has no watermark yet, only the restored floor 10:14:34.999. From that restored state the flush makes the Sink after the flush panel agree with it: the two M10:14 items are deleted, the late_* attributes are removed from the two H10 items while the items stay, and Valkey's total for ad_7 drops from 5 to 4, which is the dip.

Why the index is sharded. An index keyed only by the checkpoint id puts every write of one checkpoint interval on one index partition. In the loop, that is up to 2,800 write units a second, and one DynamoDB partition takes at most 1,000; a throttled index throttles writes to its table. With a suffix 0 to 7, each index partition takes about 2,800 ÷ 8 = 350. The flush queries all 8 shards for each epoch ≥ c. (Every sink-written item carries a tag, so the index isn't sparse in any useful sense; querying by tag value is what keeps the flush's read small.)

Why exact tags shrink the flush. With an exact tag, only values tagged ≥ c need attention: values written earlier were written from state at or before the checkpoint, so they can never be ahead of restored state. The ad-click loop also rewrites every key restored state holds (about 300,000 late totals, about 107 s): safe, but more than necessary. The cost that remains: an index entry with every sink write, the flush's own writes competing with live output (rate-limit them), and the dip.

The tag must be set at the barrier

Which id goes into the tag? The obvious answer, "the last completed checkpoint", has a gap: a checkpoint completes some time after the sink passed its barrier, and a write in between carries the older id.

Branch G (from the reference implementation): c41's upload is slow and completes at 10:15:20.5 instead of 10:15:01.5. c12's late total is written at 10:15:20.0.

So the tag is the id of the last barrier this sink subtask has snapshotted. Flink hands it to the sink when it snapshots: Sink V2's StatefulSinkWriter.snapshotState(checkpointId), or FunctionSnapshotContext.getCheckpointId() for older sink functions. On start after a restore, initialise it from the restored checkpoint's id (getRestoredCheckpointId(), an optional value). Never use notifyCheckpointComplete: it arrives later, and Flink calls these notifications "best effort", so they can be skipped. Checkpoint ids never fall below the restored id. A failover continues the counter, and restoring a savepoint sets it to the savepoint's id + 1, so "≥ c" finds everything written since. (After a savepoint restore, two job versions can use overlapping ids: branch B.)

Closed rows are final

The flush handles values written after the checkpoint. Branch W showed the other danger: a replay rewriting a row finished before it. So closed-window attributes are written with a condition:

pseudocodeCopy
CLOSED-WINDOW WRITE
    UpdateItem ad_metrics key (ad, "M10:14")
        SET fast_valid = :n, fast_ckpt = :epoch, job_gen = :g, fast_ix = ":epoch#:shard"
        CONDITION (attribute_not_exists(fast_ckpt) OR fast_ckpt >= :c)      -- :c = the restored checkpoint (before any restore, the id the job started from)
              AND (attribute_not_exists(job_gen) OR job_gen <= :g)
    a refused write is counted, not retried
    the recount repairs what it leaves out

This write stores a closed minute's count and refuses to overwrite a row that was finished before the checkpoint the job restored from. The row is in ad_metrics, keyed by the ad and the minute (such as M10:14). :n is the minute's valid click count, stored in fast_valid; :epoch is the id of the last barrier the sink subtask passed, stored in fast_ckpt; :g is the job generation, stored in job_gen; fast_ix is the index key ":epoch#:shard". :c is the restored checkpoint id or, before any restore, the id the job started from. The condition allows the write only if the row has no fast_ckpt or its fast_ckpt is at least :c, and only if the row has no job_gen or one no higher than :g. In branch G above, a minute row tagged 40 refuses the replay's value of 3 after the restore from c41. A refused write is counted and not retried, and the batch recount later repairs what it left out. The result is either the new value with its tags, or an unchanged row and one more refused write in the count.

Transactional sinks and their timeouts

The other way to make output safe is to not show it until the checkpoint that covers it completes. A two-phase-commit sink (Flink's Kafka sink in exactly-once mode, or its file sink) writes each checkpoint interval's output inside a transaction:

pseudocodeCopy
TWO-PHASE-COMMIT SINK
    1. after barrier n:
        open transaction T(n)
        write output into it
    2. at barrier n + 1:
        flush and pre-commit T(n)
        open T(n + 1)
    3. when checkpoint n + 1 completes:
        commit T(n)
    4. after a failure:
        commit or abort the open transactions according to the restored state
    readers use isolation.level = read_committed

This procedure keeps a sink's output invisible to readers until the checkpoint that covers it has completed, so a replay never shows a reader the same row twice. n is a checkpoint id, and T(n) is the sink's transaction that holds the output written after barrier n. At barrier n + 1 the sink flushes and pre-commits T(n), which makes it ready to commit, and opens T(n + 1). When checkpoint n + 1 completes, the sink commits T(n). After a failure, the restored job commits or aborts each open transaction according to its restored state. Readers set isolation.level to read_committed, so they see only committed transactions. In branch K below, T41 holds minute 10:14's rows, the crash comes before c42 completes, and T41 is aborted. The result is committed output that readers see only after the next checkpoint completes.

Branch K (from the reference implementation): the job also writes closed minutes to a Kafka topic in exactly-once mode.

Synthesizing vector architecture diagram...

A read_committed reader stops at the partition's last stable offset, the first message of any open transaction, so an open transaction also holds back everything written after it. And a transaction the broker times out before the sink commits it is lost ("data loss may happen when Kafka expires an uncommitted transaction", in Flink's words). Flink's documentation puts the requirement as "maximum checkpoint duration + maximum restart duration"; the checkpoint timeout bounds the duration:

With our numbers: we set the checkpoint timeout to 60 s, so 30 + 60 + 50 = 140 s; choose a transaction timeout of 600 s: 140 < 600 ≤ 900 s, the broker's 15-minute default maximum (MSK doesn't change it). With Flink's default 10-minute checkpoint timeout the left side would be 30 + 600 + 50 = 680 s, still under 900 but with little room. And one trap: Flink's KafkaSink sets the producer's transaction.timeout.ms to 1 hour by default, 3,600 s > 900 s, and the broker refuses a producer whose timeout is above its maximum when it initialises (InitProducerId). A KafkaSink left at its default fails on a default broker: lower it.

The replay then continues, from the reference implementation:

Synthesizing vector architecture diagram...

Snapshot W5, 10:17:00.6: minute 10:14 closed again, with c12 on time this time and no trace of run 1's late decision. Every click is counted exactly once: ad_7 5 in minute 10:14 plus c15 in hour 10's late total; ad_9 3. The steps, in time order: in the Shards panel, S1 has read the live click c14 and S2 the live click c16, so their watermarks are 10:16:39.999 and 10:16:44.999; S3 is marked idle at 10:17:00.6. The Job watermark panel therefore moves to the minimum of S1 and S2, 10:16:39.999, which is past the end of minutes 10:14 and 10:15. The Keyed state panel holds the replay's counts for minute 10:14, ad_7 = 5 and ad_9 = 3, and the late total for ad_7, hour 10, = 1. The Sink panel receives the two minute counts, tagged fast_ckpt 43, and still holds ad_7's late total tagged late_ckpt 41 from the replay, no late_* on ad_9's hour row, and minute 10:13 = 6. The Checkpoint storage panel holds c43, taken at 10:17:00.0 with the floor 10:14:51.999.

Part 9. Rescaling and upgrades

The Super Bowl traffic arrives, and the job must go from 2 subtasks to 4. Later, a new version must replace the old one. Both move keyed state, and both are restores with everything Parts 7 and 8 require.

Key groups

Keyed state can't be rehashed key by key on every rescale: that would mean reading every key. Instead, Flink splits the key space into a fixed number of key groups, the unit that moves:

• Key group = murmurHash(key's hash code) mod max parallelism (128 here).
• Owner = key group × parallelism ÷ max parallelism, rounded down. So each subtask owns one contiguous range of key groups.

Branch Q, computed with Java's string hash and Flink's murmur hash in the reference implementation:

Synthesizing vector architecture diagram...

Max parallelism is therefore the ceiling for rescaling and must be chosen before the first deployment: changing it later means the job can't restore from its old snapshots. The default differs: open-source Flink uses min(max(roundUpToPowerOfTwo(1.5 × parallelism), 128), 32,768), while Managed Service for Apache Flink sets 128 for every operator if the parallelism is at most 128 and nothing is set.

Rescaling is a restore

A rescale is a savepoint plus a restart with the new parallelism. Everything from Parts 7 and 8 applies: the replay from the savepoint's positions, the flush (every writer's values tagged ≥ the savepoint's id), and the floor: the floor is saved as union list state, so after a rescale every new subtask receives all the old subtasks' values and takes their minimum, which is safe whichever keys it now owns.

The managed service's automatic scaling does exactly this: when the maximum containerCPUUtilization stays at or above 75% for 15 minutes, it doubles the parallelism (and halves it after 6 hours below 10%), and the application has downtime while it restarts from a snapshot.

Blue/green and job_gen

An in-place update on the managed service uses stop-with-savepoint, so two versions never overlap. The risk appears when you run two applications side by side: v2 starts from a savepoint of v1 and both write to the same sink until v1 is stopped. Each version gets a higher generation number, job_gen, and every write carries attribute_not_exists(job_gen) OR job_gen <= :g.

Branch B (from the reference implementation): v1 is job_gen 7; v2 is job_gen 8, started from v1's savepoint c51 taken at 10:20:00; v1 keeps running until 10:23:00. v2 excludes one click v1 counts.

So the rule, the ad-click loop's too: the new version flushes at its restore as always, and once the old one has stopped writing, sweeps once more to remove only what the old one wrote since the savepoint. Until then v2 writes normally. A caveat: job_gen here is per item, so if v2 writes an item's fast_* after v1 wrote its late_*, the item's job_gen becomes 8 and v1's late_* can no longer be told apart; a fully general sweep needs a generation per writer (fast_gen and late_gen, or gen#epoch#shard in the index key).

Two different numbers, two jobs: job_gen says which version wrote a value (fencing between versions, like a fencing token in Leases, Fencing Tokens & Distributed Locks); fast_ckpt/late_ckpt say which checkpoint interval wrote it (for the flush). The two versions' checkpoint ids overlap (both continue from 51), and that is fine: the flush asks "≥ 51" and finds both versions' writes, and job_gen decides between them.

Part 10. Compared on equal terms: stream, micro-batch, batch recount

A continuous stream is one way to count. Micro-batch engines, Kafka Streams and a nightly batch are others, and each answers the three questions of Part 1 differently.

Four ways to count minute 10:14

From the reference implementation, the same clicks (run 1's read order, dedup applied to all), with comparable settings: a 5 s allowance, grace or threshold everywhere.

Two things to see. Only the batch recount and the Flink replay put c12 in minute 10:14, and Flink's run-1 answer plus its late total gives the same hour total. And Kafka Streams and Spark lose c15 and c12 from the stream result entirely unless something else counts them.

The table

Lambda and kappa architectures, and what each costs

Lambda is a stream for fast, provisional answers plus a batch recount for exact ones. Its cost: the rules (dedup, fraud, lateness) live in two code paths that must agree, and the ad-click loop reconciles them every hour with final = fast + late − A + B. Kappa is one stream job, and reprocessing means replaying the stream. Its cost: a replay needs the input kept long enough, and replaying raw input brings back records that were later rejected, so the source of truth must be stored with its verdicts. The ad-click loop keeps lambda for billing because a bug in the stream job would cost money, and the recount is the second opinion.

Reprocessing a day

A bug found after a day can't be fixed by replaying the stream: Kinesis keeps records 24 hours by default. So:

1. Read the day from the S3 lake as a bounded source, with the same event-time code.
2. Run it in batch execution mode: the input is complete, so there are no watermarks and no lateness to decide.
3. Write under a new job_gen, into final_*-style attributes or a separate table that is swapped in when checked; never over the live stream's attributes.
4. Never replay raw input without the exceptions log (duplicates, invalid clicks, late clicks, each with its verdict), or records that were rejected come back.

Part 11. End to end through the layers

The statement "Flink handles it" leaves out six places where a click is held, each with different rules about what survives a crash. Following c12 through all six turns the page into a checklist.

One click through six layers

Synthesizing vector architecture diagram...

The four numbers to watch

Part 12. On AWS

The mechanism is Flink's, whoever runs it. What AWS adds is where it runs, where the state and checkpoints live, and which sources and sinks it uses.

Managed services that use it

ElastiCache for Valkey appears on this page as the running-total store that the flush also rewrites; it is a plain store the job writes to, not a service that uses the mechanism.

Running it yourself

Shared limits

• The watermark is shared by every key: one slow or stuck shard holds back every window of every ad; idleness trades that for late data.
• Checkpoint uploads share the worker's network with source reads, sink writes and shuffles; RocksDB compaction and checkpoint staging share the local disk with processing.
• Backpressure delays aligned barriers: one slow sink lengthens checkpoints, which lengthens the replay, which lengthens the flush.
• The 50 GB per KPU holds state, compaction space and staging; a stuck watermark lets state grow until the guard TTL.
• The flush shares the sink's write capacity with live output (2,800 WCU in the loop); rate-limit it.
• The tag index takes every sink write: shard it (<ckpt>#<0-7>: about 350 WCU per index partition against 1,000).
• The replay shares processing capacity with live input: the backlog drains at capacity − arrivals (the loop: 64K − 11.6K ≈ 52K clicks/s on an average day; 28K/s at the Super Bowl).
• An open Kafka transaction holds back every read_committed reader of its partitions.

Look-alikes that are not this mechanism

Part 13. What you've learned

Back to the double count

A worker died, the job restored c41 and replayed, and a click could have been counted twice without anything being added twice. Here is how each piece did its job in our minute:

• Event time put every click in the minute it happened (Part 1), and the watermark, the minimum over active shards, decided when minute 10:14 was complete: 10:15:07.2, with S3 left out as idle (Part 2, snapshot W2).
• Hours are built from closed minutes, and the pre-aggregator decides lateness per click and flushes before watermarks, so nothing is counted in two places or lost between them (Part 3).
• c15 and c12 went to the side output, and the dashboards overwrite early results by key (Part 4, snapshot W3).
• Checkpoint c41 cut state and positions together, with the pre-aggregator's buffer flushed at the barrier (Parts 5 and 6, snapshot W1).
• The restore brought state back but not the watermark: c12 was on time in the replay, and the floor kept c15 (and minute 10:13) from re-opening a closed minute (Part 7).
• The flush removed every value tagged 41 or later, per writer, and the closed-row condition protected rows finished earlier; the replay rewrote minute 10:14 as 5 and 3, with no late total for ad_9 (Part 8, snapshots W4 and W5).
• Rescales and blue/green switches are restores with the same rules, plus job_gen (Part 9).

The whole story, event by event

The cheat card

Failure checklist

• Are windows assigned by event time, and is device time clamped before the watermark sees it?
• Is the job's watermark the minimum over sources, with idleness on, and is watermark lag alarmed?
• Do hour rows come from closed minutes, never a separate hour window?
• Does the pre-aggregator decide lateness per click, flush before forwarding a watermark, and flush at the barrier?
• Is dedup state cleared by event-time timers, with TTL only as a much longer guard?
• Does anyone believe "exactly-once" covers the output? Is every sink write absolute, never ADD, or fenced with last_window?
• After every restore and rescale, does the flush remove every writer's values tagged ≥ the restored checkpoint, and only that writer's, before new output?
• Is each tag the last barrier the sink passed, initialised from the restored id, never the last completed checkpoint?
• Is a watermark floor restored and enforced before the window, and are closed rows conditioned on their tag?
• For a 2PC sink: interval + checkpoint timeout + restart < transaction timeout ≤ the broker's maximum; KafkaSink's 1-hour default lowered; readers read_committed?
• For side-by-side versions: a generation on every write, and a second sweep (the old version's values only) after the old one stops?

Think-first drills

Interview questions

Where to go next

• Idempotency & Effectively-Once Processing: the rules this page implements (effectively-once, the flush, dedup expiry and the acceptance horizon), which this page applies in its Parts 4 and 8.
• Change Streams & the Transactional Outbox: how events get out of databases and into the streams this page counts.
• Leases, Fencing Tokens & Distributed Locks: fencing tokens in general; job_gen is one.
• LSM-Trees & Compaction: how RocksDB, the state backend, stores and compacts state.
• Replication, Quorums & Read-Your-Writes: replication of the stores the job reads and writes.
• Background primitives: Message Queues vs Event Streams, Two-Phase Commit & Saga Orchestration, Event Sourcing & CQRS.
• Loops: the ad-click aggregation loop (steps 2.1 to 2.6, R2.5 to R2.8, R3.9), the metrics and alerting loop (steps 2.2, 2.3, 2.5), YouTube (step 2.5), the URL shortener (step 2.2), the rate limiter (step 3.2), search autocomplete (step 2.1) and the Google Maps loop.