How Apache Flink checkpoints streaming state

The key insight for interviews

Use RocksDB with incremental checkpointing for any production job with state larger than 1 GB. The checkpoint I/O drops from O(total_state) to O(delta), which is typically 10-100x smaller. This is the single most impactful configuration choice for Flink checkpoint performance.

When a TaskManager crashes, Flink's recovery process is deterministic and relies entirely on the last successful checkpoint. Understanding this flow is essential for debugging production failures.

Recovery time depends on three factors: slot allocation time (seconds on Kubernetes), state download size (the dominant factor for large-state jobs), and the number of records to replay from source (proportional to checkpoint interval). For a job with 50GB of state on S3, expect 30-90 seconds of recovery time.

RocksDB Tuning for Checkpoint Performance

RocksDB's default configuration is tuned for general-purpose workloads, not for Flink's checkpoint pattern. A few settings make a major difference:

Block cache size: RocksDB caches SST file blocks in memory. The default is small (8 MB). For Flink jobs that do frequent key lookups (e.g., windowed aggregation, joins), increasing the block cache to 256 MB or more reduces disk reads and speeds up both processing and checkpoint snapshot creation.

Write buffer size and count: RocksDB accumulates writes in memory (memtables) before flushing to SST files. Larger write buffers (64 MB instead of the default 4 MB) reduce the number of SST files created between checkpoints, which directly reduces incremental checkpoint upload size. Using 2-3 write buffers allows Flink to keep writing while one buffer flushes.

Compaction style: The default level compaction is fine for most Flink jobs. For workloads with high write throughput and limited key range (e.g., session windows with TTL), FIFO compaction can reduce write amplification because expired entries are dropped by simply deleting old SST files.

// RocksDB tuning for Flink (in flink-conf.yaml)
state.backend.rocksdb.block.cache-size: 256mb
state.backend.rocksdb.writebuffer.size: 64mb
state.backend.rocksdb.writebuffer.count: 3
state.backend.rocksdb.compaction.level.max-size-level-base: 256mb

These settings are not theoretical. On a production job with 80GB of keyed state processing 500K events/second, tuning the block cache from 8 MB to 256 MB reduced checkpoint duration by 40% because RocksDB stopped doing disk reads during the state snapshot phase.

Operator UIDs: The Bridge Between Savepoints and New Job Graphs

When restoring from a savepoint, Flink maps saved state to operators using Operator UIDs. If you assign stable UIDs to your operators (via .uid("my-operator-id")), Flink can match state from the savepoint to operators in the new graph, even if the graph topology changed.

If you do not assign UIDs, Flink auto-generates them based on the operator's position in the DAG. Any change to the DAG (adding an operator, reordering, changing parallelism) shifts positions and breaks the mapping. The restore fails with "cannot find state for operator" errors.

// Always assign stable UIDs to stateful operators
env.addSource(kafkaSource)
   .uid("kafka-source")
   .keyBy(event -> event.getUserId())
   .window(TumblingEventTimeWindows.of(Time.minutes(5)))
   .aggregate(new UserActivityAggregator())
   .uid("user-activity-aggregator")
   .addSink(kafkaSink)
   .uid("kafka-sink");

This is one of those rules that costs nothing to follow but is catastrophic to ignore. I always assign UIDs to every stateful operator from day one. Retrofitting UIDs after the fact means you cannot restore from existing savepoints because the auto-generated UIDs will not match.

Savepoint compatibility is fragile

Savepoints only work if the state serializers are compatible between the old and new job versions. If you change the data type of a keyed state value (e.g., from Long to a custom POJO), the savepoint restore fails unless you implement state migration. Plan schema evolution before deploying stateful Flink jobs.

Retained Checkpoints on Cancellation

By default, Flink deletes checkpoints when a job is cancelled. For production jobs, I always set ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION. This keeps the last checkpoint in S3/HDFS even after cancellation, letting you restart from it manually. Without this setting, cancelling a job during a deployment means losing the last checkpoint and starting from scratch.

CheckpointConfig config = env.getCheckpointConfig();
config.setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

Restart Strategies

When a task fails, Flink's restart strategy determines how the job recovers. There are three built-in strategies:

Fixed-delay: Restarts the job up to N times, waiting a fixed delay between attempts. If the job fails more than N times, it transitions to the FAILED state.

// Fixed-delay restart: 5 attempts, 30 seconds between each
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 5
restart-strategy.fixed-delay.delay: 30s

Failure-rate: Allows a fixed number of failures within a time window. More forgiving than fixed-delay for jobs that occasionally fail due to transient issues (e.g., network blips, garbage collection pauses).

// Failure-rate restart: max 3 failures per 5 minutes, 30s delay
restart-strategy: failure-rate
restart-strategy.failure-rate.max-failures-per-interval: 3
restart-strategy.failure-rate.failure-rate-interval: 5min
restart-strategy.failure-rate.delay: 30s

No restart: The job transitions to FAILED immediately on the first failure. Only useful for batch jobs or testing.

For production streaming jobs, I always use failure-rate. A single transient failure should not count against a hard retry limit that was set hours ago. With failure-rate, the failure budget refreshes over time, so a job that fails once an hour can run indefinitely, but a rapid-fire failure loop (3 failures in 5 minutes) still triggers a hard stop.

Key Metrics to Monitor

For any production Flink job, I set up alerts on these metrics:

Metric	Alert Threshold	Why
`lastCheckpointDuration`	> 50% of checkpoint interval	Approaching timeout. Risk of checkpoint overlap.
`numberOfFailedCheckpoints`	> 0 in 10 min window	Indicates storage issues, timeout, or barrier problems.
`checkpointAlignmentDuration`	> 10 seconds	Partition skew causing backpressure. Consider unaligned checkpoints.
`lastCheckpointSize`	Unexpected growth trend	State growing unbounded. Check for missing TTL or window cleanup.
`numberOfCompletedCheckpoints`	Stopped incrementing	Job may be stuck or checkpoints are silently failing.
`rocksdb.compaction-pending`	> 5	RocksDB falling behind on compaction. Slows reads and checkpoints.

The Flink Web UI shows checkpoint history with duration, size, and per-operator breakdown. I use this daily when tuning production jobs. The per-operator view reveals which specific operator is the bottleneck (e.g., a single operator with 40GB of state taking 90% of the total checkpoint time).

How Barrier Alignment Works

When checkpoint N starts, Flink needs a consistent cut across every source and operator. It achieves that with checkpoint barriers that flow in-band with the data stream.

Here is the sequence step by step:

Trigger: The JobManager sends a "trigger checkpoint N" RPC to every source operator.
Source snapshot: Each source stores its current position (Kafka offsets, file positions) and injects a barrier with ID N into its output stream.
Barrier propagation: The barrier flows downstream mixed in with normal records. Everything before the barrier belongs to checkpoint N; everything after belongs to checkpoint N+1.
Barrier alignment: When a multi-input operator receives a barrier on one input channel, it blocks that channel and keeps draining the others until their barriers arrive too.
Operator snapshot: Once all inputs are aligned, the operator snapshots its state to checkpoint storage.
Forward barrier: The operator forwards the barrier to its downstream operators.
ACK: Each operator acknowledges checkpoint N back to the JobManager.
Completion: When the JobManager receives ACKs from every operator, the checkpoint is marked complete.

Why barrier alignment guarantees exactly-once

Barrier alignment ensures that every record is counted in exactly one checkpoint. Records before the barrier on all input channels are reflected in the operator's snapshot. Records after the barrier are not. This creates a consistent cut across the DAG, so there are no in-flight records counted by one operator but missing from another.

Async snapshots: why checkpoints don't block processing

After barrier alignment, the operator snapshots its state asynchronously. For heap state, Flink uses copy-on-write snapshots. For RocksDB, it starts uploading SST files in the background. The blocking part is the alignment phase, not the storage I/O itself. That is why well-tuned Flink jobs can checkpoint large state without pausing the pipeline for the full upload duration.

The analogy I use is a toll booth on a multi-lane highway. The barrier is like a gate that closes one lane at a time. Cars behind the gate are counted in the current batch. The gate only opens after all lanes have been blocked and the count is saved.

The Cost of Barrier Alignment: Backpressure

Barrier alignment has a cost. If one input partition is much faster than another, the fast side gets blocked waiting for the slow barrier, and its records pile up in memory.

To put numbers on it: imagine a join operator with two input streams. Stream A processes at 100K events/second and Stream B at 10K events/second. When Stream A's barrier arrives, the operator blocks Stream A and waits for Stream B. If Stream B is 5 seconds behind, the operator buffers 500K records from Stream A during the wait. That is 500K records sitting in memory, not being processed, while upstream operators back up.

This is why checkpointAlignmentDuration is one of the most important metrics to monitor. If it consistently exceeds a few seconds, you likely have partition skew, and unaligned checkpoints may be the right tradeoff.

Unaligned Checkpoints: Trading Storage for Latency

Flink 1.11 introduced unaligned checkpoints to solve the backpressure problem. Instead of blocking channels and waiting for barriers to align, unaligned checkpoints let the barrier overtake buffered data.

The diagram above shows the core difference. In aligned mode, the operator waits for all channels before snapshotting. In unaligned mode, it snapshots immediately on the first barrier and includes buffered records as part of the checkpoint state.

When a barrier arrives on one channel in unaligned mode, the operator does not block. It immediately snapshots its state and the in-flight buffered records on the other channels. On recovery, Flink restores the operator state and replays those buffered records too.

Use aligned checkpoints unless you have evidence of barrier-alignment backpressure. That is the right default for most production jobs.

State TTL: Preventing Unbounded State Growth

Without TTL (time-to-live), keyed state grows forever. If you keyBy(userId) and your user base grows to 100 million, you can end up with 100 million state entries in RocksDB even if most of those users have not been active in months.

Flink's State TTL lets you attach an expiration time to state entries. After the TTL expires, the entry is cleaned up during a later state access or during RocksDB compaction.

// Configure 24-hour TTL on keyed state
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)
    .build();

ValueStateDescriptor<UserSession> descriptor =
    new ValueStateDescriptor<>("user-session", UserSession.class);
descriptor.enableTimeToLive(ttlConfig);

The .cleanupInRocksdbCompactFilter(1000) line matters. It tells Flink to piggyback on RocksDB compaction to clean up expired entries in bulk. Without this, expired entries are only removed when accessed, so inactive keys can linger indefinitely.

I configure TTL on every keyed state descriptor in production. The number of times I have seen Flink jobs grow to terabytes of state because someone forgot TTL is higher than I would like to admit.

State Backends: Where Operator State Lives

Flink operators maintain state (counters, windows, key-value maps) that survives across records. The state backend determines where this state is stored during normal processing (between checkpoints) and how it gets snapshotted.

HashMapStateBackend (Heap)

State lives as Java objects on the JVM heap. Fast access (no serialization for reads/writes), but limited by available heap memory. A TaskManager with 8GB heap can hold roughly 4-6GB of state before GC pressure becomes a problem.

Checkpointing with HashMap backend serializes the entire state to bytes and writes it to checkpoint storage. This is a full snapshot every time, no incremental option.

EmbeddedRocksDBStateBackend

State lives in RocksDB, an embedded key-value store backed by local SSD. State size is not limited by heap memory (RocksDB uses off-heap memory and spills to disk). I have seen production jobs with 500GB+ of state running on RocksDB.

Checkpointing with RocksDB can be incremental: Flink tracks which RocksDB SST (Sorted String Table) files changed since the last checkpoint and uploads only the new files to checkpoint storage.

Aspect	HashMapStateBackend	EmbeddedRocksDBStateBackend
State location	JVM heap	Local SSD (off-heap)
Max state size	~4-8 GB (heap limited)	100s of GB (disk limited)
Read latency	~10 ns (object access)	~1-10 μs (deserialization)
Write latency	~10 ns (object mutation)	~1-5 μs (serialization + LSM write)
Checkpoint type	Full snapshot only	Full or incremental
Checkpoint size	Equal to total state size	Delta only (with incremental)
Best for	Small state, low latency	Large state, production workloads

For anything beyond a toy workload, I default to RocksDB with incremental checkpointing. That combination is what makes large-state Flink jobs viable in production.

Incremental Checkpointing: How RocksDB Deltas Work

The checkpoint interval tradeoff

A shorter checkpoint interval (e.g., 10 seconds) means less data to replay on recovery but higher I/O overhead from frequent snapshots. A longer interval (e.g., 5 minutes) reduces I/O but means potentially 5 minutes of data replay on failure. For most production jobs, 30-60 seconds is the sweet spot. I only go shorter for financial or real-time bidding workloads where even 30 seconds of replay is unacceptable.

Another critical metric is checkpointDuration / checkpointInterval. If this ratio exceeds 0.5, your checkpoints are taking more than half the interval to complete, which means the next checkpoint triggers before the previous one fully settles. This leads to checkpoint queueing, increased state backend load, and eventually checkpoint timeouts. Keep this ratio below 0.3 for healthy operation.

Incremental checkpointing is the reason large-state Flink jobs are viable in production. Without it, a job with 100GB of state would need to write 100GB to S3 every checkpoint interval. With it, each checkpoint writes only the SST files that changed, typically a few hundred megabytes.

Here is how it works:

RocksDB stores data in SST files on local disk (like any LSM-tree engine).
Between checkpoints, writes create new SST files and compaction may merge existing ones.
When a checkpoint triggers, Flink compares the current set of SST files to the set from the previous checkpoint.
New SST files (created since the last checkpoint) are uploaded to checkpoint storage.
Unchanged SST files (already in storage from a previous checkpoint) are referenced by pointer, not re-uploaded.
The checkpoint metadata records the full set of SST files that comprise this checkpoint's state.

Incremental checkpoint gotcha: cleaning up old checkpoints

Because incremental checkpoints reference SST files from previous checkpoints, you cannot delete old checkpoint directories arbitrarily. Flink's retention policy handles this, but if you manually clean S3 paths, you can break the reference chain and corrupt newer checkpoints. Let Flink manage checkpoint lifecycle.

Why Kafka Streams doesn't need checkpoint barriers

Kafka Streams stores state in RocksDB and continuously replicates changes to a Kafka changelog topic. If a consumer crashes, the new instance rebuilds state by replaying the changelog. There are no explicit checkpoint barriers because the changelog IS the checkpoint. The tradeoff is higher Kafka broker I/O (every state mutation produces a changelog record) and dependency on Kafka for both data and state storage.

// Incremental checkpoint structure (simplified)
Checkpoint N:
  - sst_001.sst (uploaded at checkpoint N-3, still valid)
  - sst_002.sst (uploaded at checkpoint N-1, still valid)
  - sst_005.sst (NEW, uploaded now)
  - sst_006.sst (NEW, uploaded now)
  Total uploaded: 2 files (~200 MB)
  Total state size: 4 files (~50 GB)

The tradeoff is that incremental checkpoints create a dependency chain. Checkpoint N depends on SST files from checkpoints N-1, N-2, and so on. If an old checkpoint's files are deleted, newer checkpoints that reference them become invalid. Flink manages this automatically, but it means you cannot delete individual checkpoints from storage arbitrarily.

My rule of thumb: enable incremental checkpointing for any RocksDB state backend job. The write amplification reduction is dramatic, and the dependency chain complexity is handled by Flink transparently.

End-to-End Exactly-Once: The Two-Phase Commit with Kafka

Flink's internal checkpointing gives exactly-once semantics within the Flink job (no record is processed twice, no record is lost). But what about the output? If Flink writes results to Kafka, how do you prevent duplicate writes after a recovery?

The answer is the two-phase commit (2PC) protocol integrated into Flink's Kafka sink.

The process works like this:

During normal processing, the Kafka sink writes records to Kafka inside a transaction (Kafka's transactional producer API).
When a checkpoint barrier arrives, the sink pre-commits: it flushes all pending writes and saves the Kafka transaction ID as part of its checkpoint state. It does not commit the transaction yet.
After all operators ACK the checkpoint, the JobManager notifies the sink.
The sink commits the Kafka transaction, making the records visible to downstream consumers.
If the job crashes before the commit, the uncommitted transaction times out and Kafka discards the records. On recovery, Flink replays from the checkpoint and re-writes them.

The Kafka Consumer Side: `isolation.level`

There is a subtle but critical detail on the consumer side. By default, Kafka consumers have isolation.level=read_uncommitted, which means they see records from uncommitted transactions. If your downstream consumer uses the default setting, it reads records that Flink has pre-committed but not yet committed. On a crash, those records get discarded by Kafka, but the consumer already processed them. You end up with phantom reads.

To get true end-to-end exactly-once, downstream consumers must set isolation.level=read_committed. This makes Kafka hold back records from uncommitted transactions until the transaction is committed or aborted.

The `transaction.timeout.ms` Gotcha

Kafka's default transaction timeout is 15 minutes (900,000ms). Flink's Kafka sink opens a transaction inside each checkpoint interval. If a checkpoint takes longer than the transaction timeout (because of barrier alignment delays on skewed partitions, or slow state backend I/O), Kafka aborts the transaction, and the sink fails.

The fix is to set transaction.timeout.ms on the Kafka broker to be larger than your maximum expected checkpoint duration. I typically set it to 2x the checkpoint timeout to provide safety margin. On the broker side, transaction.max.timeout.ms must be at least as large as the producer's transaction.timeout.ms.

End-to-end exactly-once requires both sides

Flink's internal exactly-once (via barriers) only prevents duplicate processing inside Flink. Without the 2PC integration at the sink, duplicate outputs will appear in Kafka after a recovery. The sink must support transactions (Kafka does, simple file appends do not). For sinks that do not support 2PC, you get at-least-once semantics and need idempotent writes on the consumer side.

Savepoints vs Checkpoints

Both are snapshots of job state, but they serve different purposes.

Checkpoints are automatic, periodic, and managed by Flink. They are optimized for fast recovery and may be incremental. Flink automatically cleans up old checkpoints (retaining only the last N configured). You should never need to manage checkpoints manually.

Savepoints are manually triggered, always full (not incremental), and designed for operational tasks: upgrading the Flink version, changing the job graph, migrating between clusters, or A/B testing pipeline changes. A savepoint is a portable snapshot that you can restore on a completely different Flink cluster.

Aspect	Checkpoint	Savepoint
Triggered by	Automatic (periodic)	Manual (operator command)
Purpose	Failure recovery	Planned operations (upgrades, migrations)
Incremental	Yes (with RocksDB)	No (always full)
Lifecycle	Managed by Flink (auto-cleanup)	Managed by operator (manual cleanup)
Portability	Same job graph only	Any compatible job graph
Speed	Fast (incremental)	Slow (full state write)

I think of checkpoints as "autosave" in a video game and savepoints as "manual save before the boss fight." Use checkpoints for reliability, savepoints for planned changes.

What Happens When Things Break

Failure	What Happens	How to Detect	How to Fix
TaskManager crashes	Flink restarts the task on a new slot. State restores from the last checkpoint. Records between the last checkpoint and the crash are replayed from source.	TaskManager heartbeat timeout in JobManager.	Configure restart strategy (fixed-delay, failure-rate). Ensure checkpoint storage is durable and accessible.
Checkpoint timeout	Checkpoint N is aborted. Flink continues processing and tries checkpoint N+1. The last successful checkpoint is still valid.	`numberOfFailedCheckpoints` metric increases.	Increase checkpoint timeout. Reduce state size. Switch to incremental checkpointing. Check for slow state backend I/O.
Checkpoint storage unavailable	Checkpoint cannot be written. Similar to timeout, the checkpoint fails. If this persists, Flink has no recovery point.	`lastCheckpointDuration` increases. Storage client errors in logs.	Use highly available storage (S3, HDFS with replication). Monitor storage health independently.
Barrier alignment backpressure	Slow partitions cause long barrier alignment, increasing checkpoint duration and end-to-end latency.	`checkpointAlignmentDuration` metric spikes. Backpressure indicators on operator UI.	Switch to unaligned checkpoints. Fix the root cause of partition skew.
State too large for heap	HashMapStateBackend runs out of heap, causing OOM.	JVM OOM error. TaskManager killed.	Switch to RocksDB backend. Increase heap (temporary fix). Reduce state by tuning TTL or window size.
RocksDB compaction stall	Heavy writes cause RocksDB compaction to fall behind, slowing reads and checkpoints.	`rocksdb.compaction-pending` metric. Increased read latency.	Tune RocksDB compaction settings. Use faster SSDs. Increase write buffer size.

Performance Characteristics

Metric	Typical Value	Notes
Checkpoint interval	30s - 5min	Shorter intervals mean less data to replay on failure, but more I/O overhead.
Checkpoint duration (small state)	100ms - 5s	HashMapStateBackend with < 1GB state.
Checkpoint duration (large state, incremental)	1s - 30s	RocksDB with 10-100GB state, incremental enabled.
Checkpoint duration (large state, full)	30s - 10min	RocksDB full snapshot with 50GB+ state. Avoid this.
Barrier alignment overhead	0 - 500ms	Depends on partition skew. Unaligned checkpoints eliminate this.
Recovery time	5s - 2min	Depends on state size (download from storage) and parallelism.
Exactly-once overhead vs at-least-once	~5-15% throughput	Cost of barrier alignment and 2PC commits.
Incremental checkpoint size	1-5% of total state	Only changed SST files. Varies with write volume.

The most impactful tuning knobs for checkpoint performance are: checkpoint interval (balance freshness vs overhead), state backend choice (heap vs RocksDB), incremental vs full snapshots, and aligned vs unaligned barriers. I start with 60-second intervals, RocksDB with incremental checkpointing, and aligned barriers, then adjust based on monitoring.

How This Compares to Alternatives

Feature	Flink Checkpointing	Kafka Streams (Changelog)	Spark Structured Streaming (WAL)	Storm (Acker)
Algorithm	Chandy-Lamport barriers	Changelog topic replication	Write-ahead log + micro-batch	Tuple tree acking
Exactly-once	Yes (barrier alignment + 2PC)	Yes (changelog + transactions)	Yes (idempotent writes + WAL)	At-least-once only
State size limit	100s of GB (RocksDB)	100s of GB (RocksDB + changelog)	Memory only per micro-batch	Minimal state
Incremental snapshots	Yes (RocksDB SST deltas)	Native (changelog is incremental)	No (WAL per batch)	N/A
Recovery time	Seconds to low minutes	Seconds (local state + changelog)	Seconds (rerun micro-batch)	Seconds (replay tuples)
Latency	True streaming (ms)	True streaming (ms)	Micro-batch (100ms - seconds)	True streaming (ms)
Backpressure during checkpoint	Yes (barrier alignment) or No (unaligned)	No (changelog is continuous)	No (checkpointed per batch boundary)	N/A

I reach for Flink when I need exactly-once processing with large state (10GB+), complex event processing (CEP), or event-time windowing with late data handling. Kafka Streams is my choice for simpler stream processing that stays within the Kafka ecosystem. Spark Structured Streaming works well when the team already uses Spark for batch and wants to unify.

Interview Cheat Sheet

When asked how Flink checkpoints: "Flink injects checkpoint barriers into data streams (Chandy-Lamport algorithm). Barriers flow through the operator DAG. When an operator receives barriers from all inputs, it snapshots its state to durable storage and forwards the barrier downstream."
When asked about exactly-once: "Barrier alignment ensures each record is counted in exactly one checkpoint. For end-to-end exactly-once with Kafka, Flink uses a two-phase commit: pre-commit when the barrier arrives, commit when the checkpoint succeeds."
When asked about large state: "Use EmbeddedRocksDBStateBackend with incremental checkpointing. RocksDB stores state on local SSD (not limited by heap). Incremental checkpoints upload only changed SST files, reducing I/O from the full state size to just the delta."
When asked about checkpoint performance: "The main bottleneck is barrier alignment (blocked channels on skewed partitions) and state snapshot I/O. Unaligned checkpoints fix the barrier problem at the cost of larger snapshots. Incremental checkpointing fixes the I/O problem."
When asked about savepoints vs checkpoints: "Checkpoints are automatic and optimized for fast recovery. Savepoints are manual, always full, and designed for operational tasks like upgrades and migrations."
When asked about recovery: "Flink restores every operator's state from the last successful checkpoint, resets source offsets to the checkpointed position, and replays data. With Kafka, this means re-reading from saved consumer group offsets."
When asked about at-least-once: "Disable barrier alignment. Operators snapshot immediately when the first barrier arrives (without waiting for other channels), so some records may be counted in two consecutive checkpoints. This is faster but can cause duplicates."
When asked about alternatives: "Kafka Streams uses changelog topic replication instead of barriers. Spark uses WAL with micro-batching. Flink's barrier approach gives true streaming latency with exactly-once, while Spark trades latency for simplicity."

Test Your Understanding

Quick Recap

Flink checkpoints use the Chandy-Lamport algorithm, injecting barrier markers into data streams to define consistent snapshot boundaries.
Barrier alignment ensures each record belongs to exactly one checkpoint, enabling exactly-once processing semantics.
The EmbeddedRocksDBStateBackend supports incremental checkpointing, writing only changed SST file deltas instead of the full state.
Unaligned checkpoints eliminate barrier-alignment backpressure by storing in-flight buffered records as part of the snapshot.
End-to-end exactly-once with Kafka uses a two-phase commit: pre-commit on barrier arrival, full commit on checkpoint completion.
Savepoints are manual, full snapshots for operational tasks (upgrades, migrations), while checkpoints are automatic and optimized for fast recovery.
Recovery restores operator state from the last successful checkpoint and replays source data from the saved offsets.
The most impactful tuning choices are: RocksDB backend, incremental checkpointing, checkpoint interval, and aligned vs unaligned barriers.

Chandy-Lamport algorithm: The theoretical foundation for Flink's distributed snapshots. Understanding the original paper helps explain why barriers work and what "consistent cut" means formally.
RocksDB and LSM-tree storage: Flink's large-state capability depends on RocksDB internals (SST files, compaction, write-ahead log). Understanding how LSM trees work explains incremental checkpoint performance.
Kafka exactly-once semantics: Flink's end-to-end exactly-once relies on Kafka's transactional producer API. Understanding Kafka transactions explains why the 2PC protocol works.
Stream processing architectures: Comparing Flink's barrier approach to Kafka Streams' changelog and Spark's micro-batch model clarifies when each system is the right choice.
Backpressure in streaming systems: Barrier alignment is a specific form of flow control. Understanding backpressure mechanics explains why unaligned checkpoints were invented.