Checkpointing & Exactly-Once

## What a checkpoint is A Flink checkpoint is a globally consistent snapshot of a running streaming job. "Globally consistent" means every part of the snapshot corresponds to the same logical cut of the input: if source subtask 0 has consumed up to offset 1,000 and source subtask 1 up to offset 4,500, then the state stored for every downstream operator is exactly the state produced by processing everything up to those positions and nothing after them. The snapshot holds two kinds of thing: - **Source positions** — the offsets or file positions each source subtask had reached. These are what make replay possible. - **Operator state** — the keyed state a job holds per key (`ValueState`, `ListState`, `MapState`, and so on) and the non-keyed operator state a task holds, such as a source's list of assigned splits. What it does *not* contain, in the default aligned mode, is data sitting in the network buffers between tasks; that data is drained into the operators before the snapshot is taken. ## Turning it on Checkpointing is disabled by default — in Flink 2.3 the `execution.checkpointing.interval` key has no default value — which surprises people. In the DataStream API: ```java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30_000); // one checkpoint every 30 seconds ``` The interval is the main knob. Others live on `env.getCheckpointConfig()`: `setCheckpointTimeout` (abandon an attempt that runs too long), `setMinPauseBetweenCheckpoints` (guarantee useful work between attempts), `setMaxConcurrentCheckpoints`, `setCheckpointingConsistencyMode` (exactly-once, the default, or at-least-once; the older `setCheckpointingMode` is deprecated in 2.x), and `setTolerableCheckpointFailureNumber` (how many consecutive failed attempts to tolerate before Flink fails the job over; the default is zero). Each setter also has an `execution.checkpointing.*` configuration key. ## Who drives a checkpoint The **checkpoint coordinator** lives on the JobManager. On each interval it assigns a checkpoint ID and instructs the source tasks to inject a *barrier* — a special marker — into their output. The barrier flows with the records through the job graph. Each task, when the barrier has passed it, snapshots its state and sends an acknowledgement carrying a handle to that state back to the coordinator. Only when every task has acknowledged does the coordinator mark the checkpoint complete, write its metadata, and notify the tasks that the checkpoint is committed. A single unacknowledged task means no completed checkpoint. ## Where the state actually goes Two separate settings matter. The **state backend** decides where working state lives while the job runs — on the JVM heap (`HashMapStateBackend`, the default), in an embedded RocksDB instance on local disk (`EmbeddedRocksDBStateBackend`), or — with the experimental ForSt backend new in 2.x — in disaggregated remote storage. **Checkpoint storage** decides where a snapshot is persisted: normally a distributed filesystem or object store such as HDFS or S3, so that a snapshot taken by a TaskManager that later dies is still readable by whichever machine takes over. ## What happens on failure When a task throws or a TaskManager is lost, the restart strategy kicks in — `exponential-delay` by default once checkpointing is on. Flink restarts the affected tasks — in the simplest case the whole job, or with pipelined-region failover only the connected region — restores their state from the last completed checkpoint, and resets the sources to the positions in that snapshot. Records after that point are read again and reprocessed. Because the state was rolled back at the same time, reprocessing rebuilds the same state rather than adding to it: a counter that had reached 500 at the checkpoint does not jump to 1,000. If checkpointing was never enabled, the default restart strategy is `none`: the failure simply fails the job, and a restart strategy you configure yourself still starts every operator with empty state. ## What checkpointing does not give you Three honest limits worth knowing early: 1. It makes *Flink's own state* consistent. It does not undo side effects already sent to the outside world; a row already inserted into a database or a record already produced to Kafka is still there after a rollback. End-to-end correctness needs cooperation from the sink. 2. It is not the mechanism for planned changes. Upgrading a job's code, changing its parallelism, or migrating it to another cluster is what savepoints are for. 3. It is not free. Every checkpoint costs I/O to persist state and, in aligned mode, some processing stall while barriers line up. A too-short interval on a job with large state can leave the job doing more snapshotting than work. ## A useful mental picture Think of a game that autosaves. The autosave records the player's inventory *and* the point in the level they had reached, together. Reloading puts both back in step. Saving inventory without position, or position without inventory, would give you a corrupted game — and that consistency between state and read position is exactly what a checkpoint buys.