Flink state backends, checkpointing and savepoints

What it is

A stateful stream operator holds data between records: a running count, a window's accumulated events, the buffered left side of a join, the last-seen value per key. That data is state, and because a stream is unbounded, the state has to survive the job outliving any individual machine.

Flink separates three concerns that are routinely collapsed into one:

Concept	What it is	Where it lives
State backend	How state is stored and accessed while the job runs	Task manager memory or local disk
Checkpoint	An automatic, periodic, consistent snapshot for failure recovery	Durable storage (S3, HDFS)
Savepoint	A manual, self-contained snapshot for operational change	Durable storage, owned by you

The distinction that matters most in an interview and in production: checkpoints are owned by Flink and savepoints are owned by you. A checkpoint is an internal artifact optimised for fast, frequent writes and fast recovery; Flink deletes old ones automatically and their format is not guaranteed to be portable across versions. A savepoint is a deliberate, aligned, self-contained snapshot that you trigger, that Flink never deletes, and that is designed to be restored into a modified job: new parallelism, changed topology, upgraded Flink version.

What this is confused with: checkpointing is not the same as exactly-once delivery. Checkpointing gives you exactly-once state semantics, meaning each record affects the state exactly once. Getting exactly-once end to end additionally requires the sink to participate in a two-phase commit, which is a separate mechanism described below and is where most "we have exactly-once" claims fall apart.

The problem it solves

A streaming job runs for months. Machines fail, get preempted, get rescheduled. A job with a one-hour session window holds an hour of accumulated state per key, and losing it means every session in flight is wrong. You need three things that pull against each other:

Recovery without recomputation. Rebuilding state by replaying the whole topic from the beginning is theoretically possible and practically absurd: a job with 30 days of retention behind it would take hours to recover, during which lag grows. Snapshots make recovery proportional to state size instead of to history.

A consistent snapshot across a distributed dataflow. This is the hard part. A job has dozens of operators across dozens of machines, each at a different point in the stream. A naive "pause everything and snapshot" costs throughput and does not scale. A naive "snapshot each operator independently" gives you an inconsistent cut: operator A has processed record 500 and operator B has processed record 400, and restoring that state double-counts or drops records depending on which way the inconsistency runs.

Operational change without data loss. You need to deploy a new version of the job, rescale from 12 to 40 parallel instances, or upgrade Flink. All of these mean stopping the job, and a streaming job's state is the accumulated value of everything it has consumed. Without a mechanism to carry that state across a redeploy, every deploy is a cold start.

Mechanics

The Chandy-Lamport insight, adapted

Flink's checkpointing is an asynchronous barrier snapshot, derived from the Chandy-Lamport distributed snapshot algorithm. The trick is a barrier: a special marker injected into the stream by the source operators.

     source            map              window            sink
       │                │                 │                 │
   ... r7 r6 [B] r5 ... │                 │                 │
                        │                 │                 │
   Barrier B flows WITH the records, in order, through every operator.
   When an operator has received B on ALL its input channels, it snapshots
   its own state and forwards B downstream.

The barrier separates records into "before this checkpoint" and "after." An operator that has seen the barrier on every input knows it has processed exactly the pre-barrier records, so its state at that instant is a consistent slice. Because barriers flow with the data rather than requiring a global pause, the job keeps processing while the snapshot is being written, which is what makes checkpointing every few seconds viable.

Barrier alignment is the complication. An operator with two inputs may receive the barrier on input 1 while input 2 is still delivering pre-barrier records. It must block input 1 (buffering its post-barrier records) until the barrier arrives on input 2. If one input is much slower, the operator stalls, and under backpressure that alignment time becomes the dominant cost. This is why checkpoint duration spikes exactly when the job is already struggling, which is the worst possible time.

Unaligned checkpoints (Flink 1.11+) fix this: instead of waiting, the operator immediately forwards the barrier and includes the in-flight buffered records in the checkpoint itself. Checkpoint duration becomes nearly independent of backpressure, at the cost of larger checkpoints. The rule I use: enable unaligned checkpoints for any job that experiences backpressure, which in practice means enable them and set execution.checkpointing.aligned-checkpoint-timeout so alignment is tried first and abandoned if slow.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
CheckpointConfig cfg = env.getCheckpointConfig();

env.enableCheckpointing(30_000);                       // every 30s
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
cfg.setMinPauseBetweenCheckpoints(10_000);             // >= 10s of real work between
cfg.setCheckpointTimeout(600_000);                     // 10 min before declaring failure
cfg.setTolerableCheckpointFailureNumber(3);            // do not fail the job on one miss
cfg.enableUnalignedCheckpoints(true);
cfg.setAlignedCheckpointTimeout(Duration.ofSeconds(5)); // try aligned, fall back
cfg.setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);  // keep on cancel

setMinPauseBetweenCheckpoints is the setting people miss. Without it, a job whose checkpoints take 28 seconds at a 30-second interval spends nearly all its time checkpointing, and each checkpoint's cost makes the next one slower. The minimum pause guarantees real processing time between snapshots and is the difference between a job that degrades gracefully and one that spirals.

State backends: the actual decision

Flink 1.13 renamed these and separated where working state lives from where checkpoints go, which was a genuine improvement in clarity.

HashMapStateBackend: state is Java objects on the JVM heap.

Fastest possible access: a hash map lookup, no serialisation.
State must fit in memory, and it is Java objects, so a 4 GB working set can easily be 12 GB of heap with object overhead.
Full GC pauses scale with state size, and a multi-second pause on a task manager looks like a failure to the job manager.

EmbeddedRocksDBStateBackend: state lives in a RocksDB instance on local disk, one per operator instance.

State can far exceed memory: terabytes per task manager are normal.
Every access serialises and deserialises, so it is roughly an order of magnitude slower per operation than heap.
Supports incremental checkpoints, which is the decisive advantage and is covered below.
State is off-heap, so GC pressure is largely unrelated to state size.

The choice is not subtle: use RocksDB unless your state is small and latency is your binding constraint. The threshold is roughly "does the working set fit comfortably in a few GB of heap per task manager," and the reason to lean RocksDB even below that threshold is incremental checkpointing.

// RocksDB working state, checkpoints to S3.
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));   // true = incremental
env.getCheckpointConfig().setCheckpointStorage("s3://flink-state/checkpoints");

Incremental checkpoints: the reason RocksDB wins

RocksDB is an LSM tree, so its on-disk state is a set of immutable SST files. A checkpoint can therefore upload only the SST files that are new since the last checkpoint and reference the rest.

The arithmetic is what makes this matter. A job with 400 GB of state, checkpointing every 30 seconds:

Full checkpoint:         400 GB uploaded every 30s = 13.3 GB/s sustained.
                         Not possible. The job cannot checkpoint at all.

Incremental checkpoint:  ~2 GB of new SST files per interval = 67 MB/s.
                         Entirely routine.

Full checkpointing puts a hard ceiling on state size that has nothing to do with whether the job can process the data. Incremental checkpointing removes it, and it is available only with RocksDB.

The catch, and it is worth stating because it surprises people: recovery is not incremental. Restoring reads the full state, so a 400 GB job's recovery time is bounded by download bandwidth. Plan for it: with 1 Gbps per task manager and 20 task managers, 400 GB is roughly three minutes of download before processing resumes, before any catch-up on the lag accumulated meanwhile. Local recovery (state.backend.local-recovery: true) keeps a copy on the task manager's local disk, so a job manager failover or a task restart on the same machine skips the download entirely. It does not help when the machine itself is gone.

The second catch: incremental checkpoints reference older files, so checkpoint N depends on files uploaded for checkpoints N-1, N-5, N-40. This makes "how much storage do my checkpoints use" much harder to answer and means you cannot delete an old checkpoint's directory by hand. Flink manages the reference counting; manual cleanup corrupts the chain.

Savepoints: the operational tool

# Trigger a savepoint and keep the job running.
flink savepoint <jobId> s3://flink-state/savepoints

# Stop the job with a final savepoint, draining in-flight data cleanly.
flink stop --savepointPath s3://flink-state/savepoints <jobId>

# Restart the new version from it, allowing operators that no longer exist.
flink run -s s3://flink-state/savepoints/savepoint-abc123 \
          --allowNonRestoredState my-job-v2.jar

The property that makes savepoints work across job changes is that state is keyed by operator UID, not by position in the graph. If you do not set UIDs explicitly, Flink generates them from the topology's structure, so adding a single operator changes the generated UIDs of others and your savepoint no longer matches.

stream
  .keyBy(Order::customerId)
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .aggregate(new OrderAggregator())
  .uid("order-5min-aggregate")          // NOT optional. Set it on every stateful operator.
  .name("5-minute order aggregate");    // display name only, not identity

Setting explicit UIDs on every stateful operator costs nothing and is the single highest-value habit in Flink development. Skipping it means the first time you need to modify the job under pressure, the savepoint will not restore.

Rescaling is the other savepoint use. Keyed state is partitioned into key groups (default maxParallelism of 128, settable up to 32768), and rescaling redistributes key groups across the new parallelism. The constraint that catches teams: maxParallelism is fixed at the moment state is first created and cannot be changed by restoring from a savepoint. If you set parallelism 4 and let maxParallelism default to 128, you can never scale past 128. Set it deliberately (1024 or 4096) at the start, accepting slightly more metadata overhead, because the alternative is a full state rebuild.

Exactly-once end to end

Checkpointing gives exactly-once state. For end-to-end you need a sink implementing TwoPhaseCommitSinkFunction:

On each checkpoint, the sink pre-commits (Kafka: writes in a transaction; filesystem: writes to a temp file).
When the job manager confirms the checkpoint is complete, the sink commits (Kafka: commits the transaction; filesystem: renames the file).
On recovery, uncommitted transactions are aborted or committed based on what the restored checkpoint knows.

This means downstream visibility is tied to your checkpoint interval. A 30-second checkpoint interval with a Kafka exactly-once sink means downstream consumers using read_committed see data in 30-second batches. That is a latency floor imposed by the delivery guarantee, and if the product needs sub-second visibility you must either shorten the interval (which raises checkpoint overhead) or accept at-least-once and make the downstream idempotent. It is the same trade discussed on the Kafka exactly-once page, viewed from the Flink side.

A worked example: a job that could not checkpoint

A clickstream sessionisation job: 90 million active sessions, 30-minute session windows, running at 240,000 events per second across 24 task managers.

Initial configuration and symptoms:

state backend:        HashMapStateBackend (heap)
checkpoint interval:  60s
checkpoint storage:   s3://.../checkpoints
task manager heap:    16 GB

Symptoms:
  - checkpoint duration: 3m 40s (interval is 60s, so always checkpointing)
  - full GC pauses:      8-14s, several per hour
  - job restarts:        ~4 per day, each losing 6-10 minutes to recovery
  - state size:          ~310 GB total

Three separate problems, and it is worth separating them because the team had been treating it as one.

Problem 1: state does not fit. 310 GB of state across 24 task managers is 13 GB per manager on a 16 GB heap of Java objects. The job was surviving only because sessions expired. Full GC pauses of 8 to 14 seconds exceeded the task manager heartbeat timeout, so the job manager declared task managers dead and restarted the job, which is where the four restarts a day came from. The GC pauses were not a symptom of a memory leak; they were the direct consequence of holding hundreds of gigabytes as live Java objects.

Problem 2: checkpoints are full. Heap state backend has no incremental option, so each checkpoint uploaded all 310 GB. At an aggregate upload bandwidth of about 1.4 GB/s that is 3m 40s, which matched the observation exactly.

Problem 3: no minimum pause. With a 60-second interval and a 220-second duration, a new checkpoint was triggered the instant the previous finished. The job never had a window of uncontended CPU and network.

The changes:

env.setStateBackend(new EmbeddedRocksDBStateBackend(true));   // incremental
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.enableCheckpointing(60_000);
cfg.setMinPauseBetweenCheckpoints(30_000);
cfg.enableUnalignedCheckpoints(true);
cfg.setAlignedCheckpointTimeout(Duration.ofSeconds(5));

plus state.backend.local-recovery: true, maxParallelism set explicitly to 1024 (it had defaulted to 128, capping future scale-out at 128), and RocksDB tuned with the SPINNING_DISK_OPTIMIZED_HIGH_MEM predefined options since the task managers had NVMe and spare memory for block cache.

Measured after:

                              before        after
checkpoint duration           3m 40s        14s
checkpoint size (per cycle)   310 GB        ~2.1 GB
full GC pause (p99)           14s           340ms
job restarts per day          4             0.1
recovery time                 6-10 min      2m 10s   (55s with local recovery)
end-to-end p99 latency        varies wildly 1.9s
throughput                    240k/s        238k/s

Throughput was flat, within noise. That is the finding worth carrying: RocksDB's per-access serialisation cost, which is the standard argument against it, was invisible here, because the job was never CPU-bound on state access. It was bound by GC and by checkpoint upload, and both of those are what RocksDB fixes. The theoretical slowdown is real for a job doing many state accesses per record on small state; it was irrelevant for a job whose problem was that the state did not fit.

The maxParallelism change deserves a separate note because it was the one that required a full state rebuild. maxParallelism cannot be changed via savepoint restore, so raising it from 128 to 1024 meant replaying from Kafka: 11 hours of catch-up. Had it been set correctly at the start, the whole migration would have been a savepoint, a config change and a restart. That is the cost of a default nobody examined at job creation.

Production evidence

Alibaba runs Flink at Singles' Day scale and their published figures describe jobs with multi-terabyte state and checkpoint intervals in the tens of seconds, which is only possible with incremental RocksDB checkpoints. Their contributions to Flink, including much of the RocksDB state backend work and later the changelog state backend, came from operating at that size. Alibaba's Blink fork was merged back into Flink and much of the state backend maturity dates from it.

Netflix uses Flink for its keystone pipeline and for real-time personalisation features, and their engineering posts describe savepoints as the deploy mechanism: stop with savepoint, deploy, restore. That workflow only works if operator UIDs are set, which is why it appears in every serious Flink style guide.

Uber's AthenaX and their Flink platform publish on running thousands of streaming jobs, and their operational writing highlights checkpoint duration and alignment time as the primary health metrics, ahead of throughput, for the reason given above: checkpoint duration degrades first and predicts a restart loop.

Flink's own release notes document the progression: FLIP-76 introduced unaligned checkpoints (1.11), FLIP-158 the generic log-based incremental checkpointing that decouples checkpoint frequency from the state backend's compaction schedule (1.15), and the 1.13 refactor separated state backend from checkpoint storage. That sequence is itself evidence that checkpointing under backpressure was the dominant pain point in production.

Ververica's (the company founded by Flink's creators) operational guidance is explicit that RocksDB should be the default and that heap state is for small state with strict latency needs, which matches the position taken here.

The debate

Heap versus RocksDB. The argument for heap is genuine: no serialisation, so state access is roughly 10x faster per operation, which matters for a job doing many state reads per record. The argument against is everything else: bounded by memory, GC pauses scale with state, and no incremental checkpointing.

My position: RocksDB by default, heap only when you have measured that state access is your bottleneck and your state is small. The reason to default to RocksDB even for small state is that state grows, and the migration from heap to RocksDB requires a savepoint and a restart at a moment you did not choose. Starting on RocksDB costs some throughput you probably cannot measure and removes an entire class of future incident.

Checkpoint interval: how often? The variables, since "it depends" is not an answer: how much reprocessing can you tolerate on recovery (the interval is your worst-case replay), what is your checkpoint duration (the interval must be several times longer), and does your sink use two-phase commit (in which case the interval is your downstream latency floor). A concrete default: 60 seconds, with a minimum pause of half that, then tune down if recovery time matters more than overhead or up if checkpoints are expensive. Below 10 seconds you are usually fighting the mechanism rather than using it.

Exactly-once or at-least-once? Exactly-once costs alignment (mitigated by unaligned checkpoints) and, with a transactional sink, adds checkpoint-interval latency to downstream visibility. At-least-once removes both and requires the downstream to be idempotent. My position: choose at-least-once plus an idempotent sink when you can, because idempotency is a property you want anyway for retries, replays and backfills, and it removes a coupling between your checkpoint interval and your product's latency. Choose exactly-once when the sink genuinely cannot be made idempotent, which most often means an append-only ledger or an external system you do not control.

Savepoints versus retained checkpoints for deploys. Retained checkpoints (RETAIN_ON_CANCELLATION) can be restored from, and they are cheaper because incremental. Savepoints are self-contained and portable across versions and state backends. Use retained checkpoints for a fast restart of the same job version; use savepoints for anything that changes the job, the parallelism, or the Flink version. The failure to avoid is discovering during an upgrade that you only have incremental checkpoints in a format the new version cannot read.

Follow-up Q&A

"Checkpoint duration is climbing. What do you look at?"

Four things, in order. Alignment time (checkpointAlignmentTime): if it dominates, the job is backpressured and the fix is unaligned checkpoints plus finding the backpressured operator. Sync versus async duration: the sync phase is the operator pausing to snapshot; if that is long, the state backend is the issue, usually a heap backend with large state. Upload size: if it is growing, either state is growing (check state size per key and whether TTL is set) or you are on full rather than incremental checkpoints. The slowest subtask: checkpoint duration is the maximum across subtasks, so one skewed key group can set the number for the whole job, and the fix is a partitioning change, not a checkpoint setting.

"Why does a savepoint fail to restore after a code change?"

Almost always operator UIDs. Without explicit .uid(), Flink derives UIDs from the topology's structure, so inserting or removing any operator changes them and the savepoint's state no longer maps. The fixes are: set explicit UIDs on every stateful operator from day one, and use --allowNonRestoredState when you have deliberately removed a stateful operator. The second most common cause is a state schema change, where the serialiser cannot read the old bytes; Flink supports schema evolution for Avro and POJOs with restrictions, and a change outside those restrictions requires a state migration or a rebuild.

"Does checkpointing give exactly-once?"

Exactly-once state: each record affects operator state once. End-to-end exactly-once additionally requires a transactional sink implementing two-phase commit, coordinated with checkpoints: pre-commit on checkpoint, commit on checkpoint completion. Without that, a recovery replays records from the last checkpoint and the sink writes them again. And note that even with it, downstream visibility is delayed by up to one checkpoint interval, which is a real product constraint people discover after shipping.

"How long does recovery take, and what drives it?"

Downloading the full state (recovery is not incremental even when checkpointing is), plus rebuilding RocksDB from the downloaded files, plus catching up on the lag accumulated during downtime. For 400 GB across 20 task managers at 1 Gbps each, downloading is roughly three minutes. Local recovery eliminates the download when the task restarts on the same machine, which covers task-level failures but not machine loss. The catch-up phase is often the longest part and is the one people forget to count: five minutes of downtime at 240,000 events per second is 72 million events of backlog.

"State keeps growing. What do you do?"

First determine whether it should. Unbounded state usually means keys that never expire: a keyed state on user ID where users stop appearing but their state stays forever. The fix is state TTL:

StateTtlConfig ttl = StateTtlConfig.newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .cleanupInRocksdbCompactFilter(1000)     // clean during RocksDB compaction
    .build();
descriptor.enableTimeToLive(ttl);

The cleanupInRocksdbCompactFilter part is what actually reclaims disk; without a cleanup strategy, TTL only hides expired values from reads and the bytes stay. The other common cause is a windowed operator without a proper trigger or with allowed lateness set very high, keeping windows alive far past their usefulness.

"maxParallelism: what is it and why does it bite?"

It is the number of key groups, which is the unit of state redistribution during rescaling. It is fixed when state is first created and cannot be changed by restoring a savepoint, so it is a permanent ceiling on parallelism. Default is derived from the initial parallelism (roughly 1.5 x parallelism rounded up, minimum 128), which for a job that starts small silently caps it low. Set it explicitly to something generous, 1024 or 4096, at job creation. Changing it later means rebuilding state from the source.

Common misconceptions

"Checkpoints and savepoints are the same thing with different names." Different owners, formats, lifecycles and purposes. Checkpoints are Flink's, automatic, incremental, deleted automatically, optimised for recovery. Savepoints are yours, manual, self-contained, retained forever, portable across versions and state backends. Using a checkpoint where you needed a savepoint is discovered during an upgrade, which is exactly when you cannot afford it.

"RocksDB is slower so use heap when you can." Slower per state access, and faster for everything that actually limits large jobs: it removes the memory ceiling, the GC pauses, and the full-checkpoint upload. In the worked example, switching to RocksDB changed throughput by less than 1 percent and eliminated four restarts a day.

"Incremental checkpointing makes recovery incremental." It does not. Recovery reads the full state. Incremental checkpointing shrinks the write side only, which is what lets you checkpoint frequently; recovery time still scales with total state size.

"Exactly-once checkpointing means my sink is exactly-once." Only if the sink implements two-phase commit. A plain JDBC or HTTP sink replays writes after recovery.

"Setting operator UIDs is a best practice." It is a requirement for any job you intend to modify. The failure is silent until the moment you need to restore, and by then the savepoint you have does not match the job you want to run.

Interview delivery note

Say this verbatim: "I default to RocksDB with incremental checkpoints, because full checkpointing puts a hard ceiling on state size that has nothing to do with whether the job can process the data. A 400 GB job checkpointing every 30 seconds would need 13 GB/s of upload; incrementally it needs about 70 MB/s." The arithmetic is what makes it a real argument rather than a preference.

The senior-versus-staff separator is recovery is not incremental. A senior engineer explains incremental checkpointing correctly. A staff engineer follows it with "but restore reads the full state, so plan recovery time from your download bandwidth, and enable local recovery so task-level restarts skip the download entirely." That asymmetry between write and read side is where recovery-time estimates go wrong.

The second signal is naming maxParallelism unprompted as a one-way door. It is a default nobody examines, it silently caps future scale-out, and changing it requires rebuilding state from the source rather than a savepoint restore. Knowing which settings are permanent at job creation is a different kind of knowledge from knowing what the settings do.

The Staff & Team Lead Interview Reference