Checkpoints

The most important hour

Everything else in Flink follows from one mechanism. A job is a set of parallel operators, each with local state; every few seconds Flink draws a consistent cut through all of them at once — every source's position, every operator's state — and writes it somewhere durable. When a TaskManager dies, Flink throws away everything that happened after the last cut, restores the cut, and re-reads the input from where the cut says it stopped. If this hour clicks, the rest of the course stops being surprising: why a sink is what decides whether a bill is right (module 8), why backpressure shows up first as slow checkpoints (module 9), why a savepoint can rescale a job (module 10). Keep your module 1 answer to "what survives" beside you — this is where it gets its second half.

Why an operator cannot snapshot whenever it likes

Start with the problem, not the mechanism. Module 6's tenant-cost subtask 2 holds an accumulator for tenant acme: $3.84 over 17 requests so far this minute. It receives RequestRecords from four assemble subtasks, because a keyBy is a shuffle and every upstream subtask may hold a request for acme. Suppose subtask 2 decided, on its own, to write its state to disk at 12:00:41.

That snapshot is useless unless everyone agrees where it is in the input. If the source subtask reading partition 0 records "I was at offset 100,240" and the one reading partition 2 records "I was at offset 101,319", but subtask 2's $3.84 already includes a record from offset 100,241 and does not yet include one from offset 101,300, then after recovery the first record is counted twice and the second is counted once — or not at all, depending on which source rewinds where. A snapshot that mixes "after the cut" records from one input with "before the cut" records from another is not a state the job was ever in. That is the distributed snapshot problem: the cut has to be the same cut on every input of every operator, without stopping the world to take it.

Flink's answer: barriers that flow with the records

Upstream's words, kept exact: Flink implements fault tolerance using a combination of stream replay and checkpointing. A checkpoint marks a specific point in each of the input streams along with the corresponding state for each of the operators. A streaming dataflow can be resumed from a checkpoint while maintaining consistency (exactly-once processing semantics) by restoring the state of the operators and replaying the records from the point of the checkpoint.

The cut is carried by stream barriers. Again upstream, exact: These barriers are injected into the data stream and flow with the records as part of the data stream. Barriers never overtake records, they flow strictly in line. A barrier separates the records in the data stream into the set of records that goes into the current snapshot, and the records that go into the next snapshot. Each barrier carries the ID of the snapshot whose records it pushed in front of it.

Where they start: Stream barriers are injected into the parallel data flow at the stream sources. The point where the barriers for snapshot n are injected (let's call it S_n) is the position in the source stream up to which the snapshot covers the data. For example, in Apache Kafka, this position would be the last record's offset in the partition. This position S_n is reported to the checkpoint coordinator (Flink's JobManager).

So for GatewayMeter a checkpoint begins when the coordinator, in the JobManager, tells the eight source subtasks (one per partition since stage 4) to start checkpoint n. Each records the offset of the partition it owns, reports it, and pushes barrier n into its output right behind the last record it read. From there the barrier is just another item in the stream: it queues behind records in the same network buffers, it crosses the keyBy shuffle to every downstream subtask, and no record that was ahead of it ever falls behind it.

Alignment at an operator with several inputs

Barrier n reaches tenant-cost subtask 2 on four channels, one per assemble subtask, and not at the same moment. Here are two of the four, mid-alignment:

The rule for the operator, in upstream's four bullets, exact:

• As soon as the operator receives snapshot barrier n from an incoming stream, it cannot process any further records from that stream until it has received the barrier n from the other inputs as well. Otherwise, it would mix records that belong to snapshot n and with records that belong to snapshot n+1.
• Once the last stream has received barrier n, the operator emits all pending outgoing records, and then emits snapshot n barriers itself.
• It snapshots the state and resumes processing records from all input streams, processing records from the input buffers before processing the records from the streams.
• Finally, the operator writes the state asynchronously to the state backend.

That is the whole answer to the problem above. The blocked channel is exactly the "after the cut" input; holding it until the other channels catch up is what makes the cut the same cut everywhere. Upstream adds the scope you need for GatewayMeter: the alignment is needed for all operators with multiple inputs and for operators after a shuffle when they consume output streams of multiple upstream subtasks. Every operator downstream of a keyBy — assemble, tenant-cost, model-latency, alerts — aligns. parse, chained to the source, does not.

The time between the first and last barrier arriving is the alignment duration, and the Web UI reports it per subtask. Under no backpressure it is milliseconds; module 9 is about what happens to it when a downstream operator is slow.

What gets written, and when the checkpoint is complete

Timing, exact: Operators snapshot their state at the point in time when they have received all snapshot barriers from their input streams, and before emitting the barriers to their output streams. At that point, all updates to the state from records before the barriers have been made, and no updates that depend on records from after the barriers have been applied. The write itself is asynchronous: the operator keeps processing while the state backend copies the snapshot to storage. Upstream's learn-flink page notes the state backends use a copy-on-write mechanism to allow stream processing to continue unimpeded while older versions of the state are being asynchronously snapshotted.

Completion is decided at the sinks: Once a sink operator (the end of a streaming DAG) has received the barrier n from all of its input streams, it acknowledges that snapshot n to the checkpoint coordinator. After all sinks have acknowledged a snapshot, it is considered completed. Only then does the coordinator treat checkpoint n as something it can recover from — and only then does the Kafka source commit offsets to the broker, for monitoring, because Kafka source commits the current consuming offset when checkpoints are completed and does NOT rely on committed offsets for fault tolerance (connector docs).

What the snapshot holds, exact: For each parallel stream data source, the offset/position in the stream when the snapshot was started and for each operator, a pointer to the state that was stored as part of the snapshot. Positions plus pointers. Not records, not a program counter, not the contents of the network buffers (unless unaligned — below). On disk, with filesystem storage, that is a directory per job under execution.checkpointing.dir holding chk-41/, chk-42/, plus shared/ for state that several checkpoints reference and taskowned/ for state the JobManager must never drop. Upstream warns the layout is not a public API.

Recovery

Exact: Upon a failure, Flink selects the latest completed checkpoint k. The system then re-deploys the entire distributed dataflow, and gives each operator the state that was snapshotted as part of checkpoint k. The sources are set to start reading the stream from position S_k. For example in Apache Kafka, that means telling the consumer to start fetching from offset S_k. And the guarantee that makes it correct: Any records that are processed as part of the restarted parallel dataflow are guaranteed to not have affected the previously checkpointed state.

Now the module's failure, with numbers. Checkpoint interval 10 s, parallelism 4, eight partitions of gateway-events at roughly 24 events per second each. Checkpoint 41 completed at t = 0. TaskManager 3 hosts assemble subtask 3 and tenant-cost subtask 2.

The offsets that rewind, per partition (source subtask i owns partition i):

About 2,688 records are processed twice inside Flink. Not one of them lands in state twice. Read the learn-flink page's definition, exact: when the ideal is described as exactly once this does not mean that every event will be processed exactly once. Instead, it means that every event will affect the state being managed by Flink exactly once. The second processing is against state that has never seen those records, because the state that did see them was on the discarded side of the cut. That is exactly-once state semantics, and it is the only exactly-once this module claims.

Two things did not rewind. The line printed at t = 6 s is on your terminal. The TenantCostMinute(acme, 12:00:00, $4.91, 21) written at t = 6 s is in the tenant-cost topic, and after recovery the 12:00 window fires again and the at-least-once KafkaSink writes it again — the connector docs say so plainly: messages may be duplicated when Flink restarts because Flink reprocesses old input records. A billing consumer reading that topic double-bills the minute: a chargeback. Everything the job did to the outside world between checkpoint 41 and the crash happened and will happen again. The state boundary is Flink's; the sink boundary is module 8.

Unaligned checkpoints

Alignment blocks channels, and a blocked channel under backpressure can hold a barrier for a long time. Exact: The basic idea is that checkpoints can overtake all in-flight data as long as the in-flight data becomes part of the operator state. The operator reacts to the first barrier it sees in its input buffers, forwards it immediately to the end of its output buffers, marks every overtaken record for asynchronous storage, and snapshots its own state. The buffered records become channel state in the checkpoint; on recovery operators first recover the in-flight data before starting processing any data from upstream operators.

When it helps and when it does not, exact: It's especially suited for applications with at least one slow moving data path, where alignment times can reach hours. However, since it's adding additional I/O pressure, it doesn't help when the I/O to the state backends is the bottleneck. Enable it with env.get_checkpoint_config().enable_unaligned_checkpoints() or execution.checkpointing.unaligned: true; the checkpointing page says this only works for exactly-once checkpoints and with one concurrent checkpoint. With execution.checkpointing.aligned-checkpoint-timeout set, each checkpoint starts aligned and switches to unaligned only if alignment has not finished by the timeout (Java setAlignedCheckpointTimeout; not shown for Python upstream).

The limitations the backpressure page lists, as of 2026-09: no concurrent unaligned checkpoints, and savepoints cannot run concurrently with them; watermarks are regenerated after in-flight data is restored, so an operator that applies the latest watermark per record can produce different results than under aligned checkpoints; a barrier still cannot interrupt the record currently being processed (a large record, a flatMap that emits many outputs, a burst of timers); and pointwise and broadcast exchanges are excluded from unaligned handling. Checkpoints in AT_LEAST_ONCE mode skip alignment differently — the operator keeps consuming every input, so on a restore, these records will occur as duplicates, because they are both included in the state snapshot of checkpoint n, and will be replayed as part of the data after checkpoint n. GatewayMeter feeds a bill, so it stays in EXACTLY_ONCE. And, exact: savepoints will always be aligned.

Restart strategies and failover strategies

Two decisions, exact: Restart strategies decide whether and when the failed/affected tasks can be restarted. Failover strategies decide which tasks should be restarted to recover the job.

The default restart strategy depends on checkpointing: If checkpointing is not enabled, the no restart strategy is used. If checkpointing is activated and the restart strategy has not been configured, the exponential-delay restart strategy and the default values of exponential-delay related config options will be used. That is the missing half of module 1's answer: your stage-1 job, with no checkpoint, failed on docker kill because none was in force. The key is restart-strategy.type, values fixed-delay, exponential-delay, failure-rate, none. Fixed delay takes restart-strategy.fixed-delay.attempts and restart-strategy.fixed-delay.delay; upstream's example is 3 attempts, 10 s apart. Upstream strongly recommends exponential-delay in production (fast retry for a one-off, backoff plus jitter so a Kafka outage does not have every job hammering the broker in lockstep); its programmatic form is marked not supported in Python API as of 2026-09, and the config-file keys (restart-strategy.exponential-delay.initial-backoff, max-backoff, backoff-multiplier, reset-backoff-threshold, jitter-factor, attempts-before-reset-backoff) are on the upstream page. Stage 7 uses fixed-delay because its arithmetic is legible in a lab.

Failover strategy is jobmanager.execution.failover-strategy: full restarts every task; region restarts the smallest set of pipelined regions containing the failed task, plus producers of any lost result partitions, plus all their consumer regions. For the default, see the upstream configuration reference: https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/deployment/config/. Read the definition of a region — a set of tasks that communicate via pipelined data exchanges — against GatewayMeter: every keyBy is a pipelined all-to-all exchange, so the whole streaming job is one region and region restarts the whole job, exactly as full would. The distinction earns its keep in batch mode, where blocking exchanges bound regions.

Configuration: GatewayMeter stage 7

Checkpointing is off by default. Stage 7 turns it on and changes nothing else — build_job(env) is module 6's, operator for operator, uids included.

# gatewaymeter/job.py, stage 7
from pyflink.common import Configuration
from pyflink.datastream import (StreamExecutionEnvironment, CheckpointingMode,
                                ExternalizedCheckpointRetention)


def build_env() -> StreamExecutionEnvironment:
    config = Configuration()
    # state backend: RocksDB, incremental snapshots
    config.set_string('state.backend.type', 'rocksdb')
    config.set_string('execution.checkpointing.incremental', 'true')
    # checkpoint storage: a volume compose mounts at /checkpoints on the
    # JobManager and every TaskManager (all of them must reach the files)
    config.set_string('execution.checkpointing.storage', 'filesystem')
    config.set_string('execution.checkpointing.dir', 'file:///checkpoints/gatewaymeter')
    # restart strategy: 3 attempts, 10 s apart (exercise 7.4 removes this)
    config.set_string('restart-strategy.type', 'fixed-delay')
    config.set_string('restart-strategy.fixed-delay.attempts', '3')
    config.set_string('restart-strategy.fixed-delay.delay', '10000 ms')

    env = StreamExecutionEnvironment.get_execution_environment(config)

    env.enable_checkpointing(10_000)                       # interval, ms
    cc = env.get_checkpoint_config()
    cc.set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)   # the default; say it anyway
    cc.set_min_pause_between_checkpoints(2_000)            # progress between checkpoints
    cc.set_checkpoint_timeout(60_000)                      # abort an in-progress checkpoint after 1 min
    cc.set_max_concurrent_checkpoints(1)
    cc.set_tolerable_checkpoint_failure_number(0)          # the default: first failure fails over
    cc.set_externalized_checkpoint_retention(
        ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION)
    return env

What each line buys, from the checkpointing page. The interval trades checkpoint overhead against how much is replayed on recovery — 10 s means at most about 10 s of input is reprocessed. Min pause guarantees progress: the next checkpoint starts no sooner than 2 s after the previous one completed, so a slow storage system cannot make the job checkpoint back-to-back (setting it implies one concurrent checkpoint). Timeout aborts a checkpoint that has not completed — the default is on the upstream configuration page. Tolerable failures defaults to 0: the first checkpoint that times out or fails in its async phase fails the job over. Retention: checkpoints are normally deleted when you cancel; RETAIN_ON_CANCELLATION keeps the last one so you can resume with bin/flink run -s <checkpoint metadata path> — and you clean it up yourself. Storage: with no directory configured, checkpoints live in the JobManager's heap (each state capped at 5 MB by default, fine for a laptop, useless in production); configuring execution.checkpointing.dir selects FileSystemCheckpointStorage, and on a real cluster the URL is s3:// or hdfs://, not a compose volume.

State backends

The state backend is where the working state lives and how it is snapshotted; checkpoint storage is where the snapshot goes. Two are bundled, plus one experimental, as of 2026-09:

Incremental checkpoints only record the changes that happened since the latest completed checkpoint — enable with execution.checkpointing.incremental: true. The upstream tuning page calls this one of the first things to try when checkpoints are slow, and the state-backends page warns that with it on, the UI's Checkpointed Data Size is the delta, not the full state; Full Checkpoint Data Size is the other column. Stage 7 uses RocksDB not because GatewayMeter's state is large yet — it is not — but so that the checkpoint you read in exercise 7.1 has a size, a sync duration and an async duration that look like production's. Module 11 comes back to ForSt.

Temporal, one paragraph

Temporal's history replays decisions: the same code re-run against the recorded log makes the same Commands, and Activity results come from the log. Flink replays records: the same operators are restored to a snapshot and re-fed the input from S_k. Neither persists a program counter. Temporal saves what the program did and reconstructs where it was; Flink saves what the state was and re-derives what the program would have done. The costs land in different places: Temporal pays per event in history size (module 6 of that course); Flink pays per checkpoint in alignment and I/O, and in the replay window on recovery.

If a TaskManager dies right now

Survives: everything in the latest completed checkpoint k — every operator's state as of the cut (open requests, per-tenant accumulators, the sorted-sample latency sketches, registered timers) and every source's position S_k. Held in execution.checkpointing.dir, owned by no TaskManager.

Recomputed: every record after S_k, on every partition, re-read from Kafka and re-applied to the restored state. Also everything from checkpoint k+1 if it was in progress — it is abandoned, not finished. Inside Flink the result is the same numbers as a failure-free run, with a delay of the restart strategy's wait plus the catch-up.

Emitted twice: every external effect between the cut and the crash — printed lines, TenantCostMinutes the at-least-once sink already wrote, alerts already published. Flink does not know they happened. Module 8 makes the sink know.

Try it

Go to labs/07-checkpoints/. Stage 7 of GatewayMeter on the compose cluster: one JobManager, four TaskManagers with two slots each, /checkpoints mounted everywhere.

• 7.1 Submit stage 7 with traffic flowing and record a completed checkpoint's ID, trigger time, duration, full size and alignment data in the Web UI.
• 7.2 Find a TaskManager that owns a subtask, kill it between checkpoints and restore cluster capacity. Record the checkpoint ID and Kafka source positions used for recovery.
• 7.3 Retain a finite seeded trace and diff materialized (tenant_id, window_end) totals against the lab's no-failure oracle. A return to RUNNING alone is not enough evidence.
• 7.4 Draw the cut at checkpoint k: two input channels and barriers, operator state, and source positions S_k. Explain what alignment waits for and which records replay.

Unaligned-checkpoint and flink run -s recovery experiments are not in the current lab; the sections above explain those mechanisms without asking you to run them.

Lab page: lab-07-checkpoints.html — the exercises, every source file, and the run-it-locally steps.