Every non-trivial stream operator remembers something — window counts, unjoined events, seen IDs, open sessions. When the process crashes at 3 AM, all that state is at stake. Checkpoints make it survive. State backends decide where it lives. Savepoints let it move. Exactly-once is the correctness discipline that ties it all together. Where the "always-on" property from M.39 meets the production reality that makes it usable.
M.40 established that stream operators maintain state — window aggregates, session groupings, join buffers, deduplication tables. M.41 is the specific problem of that state under production conditions: what happens when the process crashes and the state is only in memory? The naive answer is "the state is lost, we rebuild it from source" — technically true, but if the state represents an hour's worth of window aggregations across 10 million keys, "rebuild from source" means reading and replaying an hour of upstream data, which takes an hour, during which the pipeline is producing no output. A streaming system that recovers slowly is only slightly better than a streaming system that never recovers, because both fail the "always-on" property M.39 promised. The whole discipline of stateful stream processing exists to make the state survive crashes fast enough that the recovery gap is measured in seconds, not hours.
The reason it's harder than "just snapshot the state" comes from the shape of a distributed streaming system. A streaming job is a DAG of operators running on many machines — sources reading from Kafka, several intermediate transformations doing joins, groups, and windowing, and sinks writing to downstream systems. Each operator has its own state; each state is being read and written millions of times per second; the operators are processing different events at any given moment. A "snapshot" of this system has to capture a consistent point across all operators — not a moment where operator A is at event #10,000 and operator B is at event #4,700, but one where all operators reflect the state as of some coherent point in the event stream. Without that consistency, recovery produces state that's simultaneously ahead of some events and behind others; the aggregations end up double-counting some records and missing others; exactly-once semantics is broken. The whole snapshot machinery exists to produce that consistent point without stopping the pipeline.
Each attempt fails at a specific level of the problem: durability (attempt 1 has none), availability (attempt 2 pauses the pipeline), consistency (attempt 3 snapshots at inconsistent points), and only attempt 4 — coordinated barrier snapshotting — achieves all three. This is the mechanism at the heart of every modern stateful stream processor: Flink's async barrier snapshotting, Kafka Streams' changelog-topic pattern, MillWheel's persistent state — all variants of the same idea. §02 makes the mechanism concrete; §03 shows where the snapshotted state actually lives; §04 lets you feel the tradeoffs.
The historical grounding matters. Chandy-Lamport (1985) is the theoretical foundation — a distributed algorithm for recording a globally consistent snapshot of a distributed system, without pausing the system. The paper was written for general distributed systems (bank transactions, distributed databases), not for streaming, but the algorithm is a perfect fit: streaming operators send messages (events) along channels (the DAG edges); the algorithm records each operator's state and each channel's in-flight messages at a coherent point. MillWheel (Google, 2013) was the first industrial-scale streaming system to build on this foundation — persistent per-key state, low-watermarks, exactly-once semantics. Flink's async barrier snapshotting (Carbone et al., 2015) is the mainstream open-source adaptation: barriers as the coordination mechanism, per-operator state backends, incremental checkpoints on RocksDB. The vocabulary in this module — barriers, checkpoints, state backends, savepoints — is Flink's, but the concepts apply across all major stream processors (Kafka Streams, Beam/Dataflow, Spark Streaming's structured streaming). The next section builds up the checkpoint mechanism concretely; §03 addresses the "where does the state actually live?" question with the three canonical state backends.
The specific machinery that produces globally consistent snapshots without stopping the pipeline is Flink's async barrier snapshotting — a practical adaptation of the Chandy-Lamport algorithm (1985) for streaming systems. The mechanism is simple to describe once you've seen it, but the details matter because the failure modes in §05 all come from misunderstanding some piece of it. This section walks through the algorithm concretely: what a barrier is, how it flows through the DAG, what each operator does when it sees one, and how the whole thing produces a snapshot that's both consistent and non-blocking.
Job Manager triggers checkpoint N; source injects barrier-N into every output partition simultaneously. Business events continue flowing before, between, and after. The barrier is a small structured record — a few bytes — with a monotonically increasing sequence number.
Each operator, on seeing barrier-N, (a) initiates an async snapshot of its current state, (b) forwards barrier-N downstream, and (c) resumes processing. The snapshot write happens in a background thread; event processing doesn't wait for it.
Operators with multiple inputs (like joins) wait for barrier-N on ALL input channels before snapshotting. Faster channels are temporarily blocked; slower ones catch up; then all events after barrier-N flow into the new checkpoint's state.
When every operator acknowledges its snapshot for checkpoint N, the Job Manager atomically marks checkpoint N as the new latest complete checkpoint. On any subsequent crash, recovery starts from checkpoint N — all operators + source offset restored to a coherent point.
The alignment step (iii) is where the algorithm gets subtle. An operator with two input channels — say, a join operator receiving both the clicks stream and the impressions stream — has to wait for the barrier on BOTH channels before it can snapshot. If clicks are flowing faster than impressions, the operator temporarily blocks the clicks channel after seeing barrier-N on it, waiting for impressions to catch up. Events on the clicks channel that arrive during this alignment window belong to the next checkpoint, not the current one, and must not be applied to the state being snapshotted. Getting this exactly right is what makes exactly-once semantics possible: the snapshot for checkpoint N reflects exactly "the state after processing every event that arrived before barrier-N on every input." Flink also supports an "unaligned checkpoint" mode that avoids the blocking at the cost of larger snapshots — the in-flight events during alignment are also snapshotted. Which mode fits depends on latency requirements vs snapshot-size costs, and modern Flink deployments often use unaligned checkpoints for low-latency workloads.
The global commit step (iv) is what makes the algorithm safe under partial failures. Between the moment a checkpoint starts and the moment it's marked complete, the system is in an intermediate state — some operators have snapshotted, others haven't. If the process crashes during this window, the checkpoint is aborted and never referenced; recovery falls back to the previous complete checkpoint. Only when EVERY operator acknowledges does the Job Manager atomically flip the "latest complete checkpoint" pointer. This atomicity property — either all operators have snapshotted or none is considered — is what makes recovery deterministic. There's no "partial checkpoint" state that could produce inconsistent recovery. The pattern is the same at heart as two-phase commit for distributed databases, adapted for the streaming context where the participants are operators and the transaction is a coordinated snapshot.
The historical arc here is worth internalizing. Chandy-Lamport (1985) proved that distributed snapshots without stopping the system were possible — a foundational contribution that seemed abstract in 1985 and turned out to be exactly what streaming systems needed 30 years later. MillWheel (Google, 2013) built on Chandy-Lamport for the specific streaming case — persistent per-key state, low-watermarks for progress tracking, exactly-once at scale. Flink (2015+) made async barrier snapshotting the industry standard implementation, refining the algorithm with alignment, incremental checkpoints, and unaligned checkpoint modes. The paper trail — Chandy-Lamport → MillWheel → Flink — is one of the cleaner examples of "old theory becoming new practice" in distributed systems. The next section addresses the "where does the snapshotted state actually live?" question, which is where operational choices really bite.
The checkpoint machinery in §02 assumes there's a "state" to snapshot — but where does that state physically live between checkpoints? The answer isn't obvious, because state at streaming volumes has to satisfy contradictory requirements: read/write at millions of ops per second (streaming throughput), grow to arbitrary sizes (potentially terabytes for large user-tracking pipelines), survive checkpoints without long pauses (incremental snapshots), and integrate cleanly with the checkpoint alignment protocol. Flink's answer — and the industry's — is to make the "state backend" pluggable: three canonical implementations, each optimized for a different point in the capacity-vs-latency space. Choosing the right one is the specific decision every stateful streaming pipeline has to make.
State lives as plain Java objects in the JVM heap: HashMaps, ArrayLists, POJOs. Reads and writes are ordinary object accesses — nanoseconds per operation, no serialization overhead during normal processing. Fastest possible state backend by a wide margin. On checkpoint, the entire state is serialized and written to durable storage (S3, HDFS); there's no incremental support — every checkpoint writes the full state. State size is bounded by JVM heap size (typically 4-32GB); beyond that, GC pressure destroys throughput and OOMs become inevitable. Fits small state, low-latency workloads, and testing/development.
Historically a distinct backend, now (in modern Flink) the "checkpoint storage" component paired with the HashMap state backend. State lives in memory during operation (identical to backend 1); the difference is where checkpoints are written — to a distributed filesystem (S3, HDFS, Azure Blob) rather than local disk or in-process. The specific role is durability for medium-sized state that fits in memory but needs to survive cluster failures. Same read/write performance as memory backend during normal operation. Same size limit (JVM heap). But checkpoints are stored durably and can be used for recovery across cluster restarts, not just individual operator failures. The right default for many production Flink deployments with moderate state.
State lives in a RocksDB LSM tree on local disk — accessed via JNI from the JVM. RocksDB is an embedded key-value store (Facebook's fork of LevelDB) optimized for high-throughput writes and reasonable-latency reads. Reads and writes go through a JNI boundary, then hit either RocksDB's in-memory memtable (recent writes) or on-disk SST files (older data) — microseconds per operation instead of nanoseconds. Trade: state size is bounded by local disk, not RAM — production Flink pipelines routinely run with 100GB+ of RocksDB state per operator, TBs in aggregate. The critical property is incremental checkpoints: RocksDB's LSM structure means most SST files don't change between checkpoints; Flink can upload only the new SSTs to durable storage, skipping the ones already there. Checkpoint cost scales with the change rate, not the total state size — a 100GB state with 1% churn between checkpoints only uploads ~1GB.
The tension between the three backends collapses to a specific choice at design time: match the backend to the state's operational profile. Small, latency-critical, bounded state → Memory. Medium, durable-recovery-needed, bounded-heap state → Filesystem (with HashMap). Large, growing, incremental-checkpoint-needed state → RocksDB. The specific decision is often "when does the pipeline outgrow the current backend?" — teams start with Memory in development, move to Filesystem for early production (moderate state, cluster recovery needed), and switch to RocksDB when the state grows beyond what heap can hold or when full-state checkpoint costs become prohibitive. Flink's design deliberately makes the backend swap a configuration change, not a code change — the state APIs (ValueState, MapState, ListState) work identically across backends. This is important because getting the initial backend choice wrong is normal; the operational maturity is in recognizing when to switch and being able to do so with a savepoint restore.
The tooling landscape mirrors this. Apache Flink exposes all three backends as first-class configuration options, with RocksDB as the recommended default for production. Kafka Streams uses RocksDB by default for its state stores, with an in-memory option for testing; changelog topics act as the "checkpoint storage" equivalent, storing all state changes for recovery. Spark Structured Streaming uses HDFS-backed state stores by default, with a RocksDB backend added in Spark 3.2+ specifically to handle the "state larger than heap" case that mirrors Flink's rationale. Across the mainstream systems, the pattern is the same: memory-backed state for speed, disk-backed state for capacity, incremental checkpoints for large state. The vocabulary and configuration surfaces differ; the underlying tradeoffs don't.
Below: each of the three state backends (memory · filesystem · rocksdb) evaluated against three different workload profiles (small state · large state · bursty growth). Watch the read/write latency shift, the checkpoint duration explode or stay bounded, the recovery time trade against operational simplicity. The 9 combinations illustrate why "which state backend?" is a workload-specific decision, not a technology preference.
The failure modes of stateful stream processing are especially unforgiving because they usually surface at 3 AM under load, not during development. The pipeline runs green in staging; the checkpoint metrics look fine; the backend is "the right one for production." Then a spike arrives, or a code deploy runs, or a checkpoint interval passes with too much state churn — and the whole system either OOMs, falls behind, or silently violates exactly-once semantics. Recognizing these five patterns before they page you is what mature stream operations looks like.
State grew (unbounded key cardinality, session windows without upper bound, per-user counters without eviction), and the checkpoint duration now exceeds the checkpoint interval. The next checkpoint starts before the current one finishes; barriers back up; watermarks lag; the pipeline can't keep up. Symptoms: increasing checkpoint duration over weeks, growing "state size on checkpoint" metric, eventually alerts on "checkpoint failed" or "back-pressure." The fix: (a) switch to RocksDB with incremental checkpoints if not already there (§05 root cause is often "we're still on FS backend for state that outgrew it"); (b) add state TTL — Flink's StateTtlConfig automatically evicts state older than N minutes; (c) audit state growth by key — often 1% of keys drive 90% of state (bots, hot users, misconfigured integrations from M.40.iii); (d) reduce checkpoint interval to smaller, faster checkpoints. Checkpoint duration is a leading indicator of state health; it must be monitored as a primary SLI.
The barrier alignment step (§02.iii) blocks fast input channels while waiting for slow ones. When two input streams have very different arrival rates or watermark advancement, the fast one spends significant time blocked during every checkpoint. State on the alignment side grows because in-flight events accumulate; latency of the fast side degrades; the whole pipeline runs at the pace of the slowest input. The fix: (a) enable unaligned checkpoints (execution.checkpointing.unaligned=true in Flink) — in-flight events are snapshotted rather than aligned, larger checkpoints but no blocking; (b) rebalance upstream rates so watermarks advance similarly; (c) if streams are fundamentally asymmetric, redesign so they don't join within a windowed context — sometimes an unwindowed left join with buffered state is more appropriate. This is the §02 alignment mechanism failing under real-world asymmetric loads.
Deploying a code change that alters state schema (adding fields, renaming fields, changing types) without using savepoints for controlled migration invalidates the checkpoint on restart. The new operator can't deserialize the old checkpointed state; the recovery fails; the pipeline either restarts from empty state (losing hours of aggregations) or repeatedly crash-loops. The fix: (a) use savepoints for every schema-changing deploy — trigger a savepoint before deployment, restart from that savepoint with the new code, and Flink's state migration tools handle the schema evolution; (b) design state schemas with forward-compatibility in mind — use Avro or protobuf serialization with schema evolution rules; (c) test schema changes in staging with production-like state before deploying. Savepoints exist specifically for this problem — they're the "release engineering" primitive for stateful streaming.
Memory-backed state (HashMap in JVM heap) scales linearly with key cardinality; the moment cardinality grows beyond what heap can hold, the JVM either OOMs or enters a GC death spiral. Staging with small cardinality gives no signal — the code runs fine; the checkpoint duration is small; the metrics look green. Production with real cardinality causes catastrophic failure. The fix: (a) provision the state backend for peak cardinality, not average — RocksDB for anything with unbounded or hard-to-estimate cardinality; (b) load-test with production-scale state, not staging-scale; (c) set explicit heap limits and watch GC pauses as an SLI — if GC time exceeds 5% of processing time, you're already in trouble. Memory backend's failure mode is binary: it works until it doesn't. RocksDB degrades gracefully.
Exactly-once in stream processing requires three cooperating components: (a) replayable sources (Kafka is; unreplayable custom sources aren't); (b) coordinated checkpoints for state (§02's async barrier snapshotting handles this); (c) idempotent or transactional sinks (2PC for Kafka; UPSERT semantics for Postgres; overwrite semantics for object stores). If any of the three is missing, the pipeline is at-least-once at best. A common trap: configuring Flink for exactly-once but using a sink that only supports at-least-once (a plain Postgres INSERT, or an HTTP webhook without idempotency keys). The pipeline restarts, replays the last checkpoint interval of events, and the sink writes them again. The fix: (a) audit the sink's idempotency guarantees — INSERT INTO ... ON CONFLICT UPDATE, or use Kafka sink with EOS mode, or write to object storage with content-addressable filenames; (b) verify empirically with a test that intentionally kills the operator between checkpoints and checks for duplicates downstream; (c) treat "configured for EOS" and "actually EOS" as separate assertions — the first is trivial, the second requires end-to-end proof. EOS is a property of the entire pipeline, not just the checkpoint machinery.
The composite pattern across all five is that stateful streaming's correctness failures are almost always operational, not algorithmic. The Chandy-Lamport → Flink barrier algorithm is proven correct; RocksDB is a mature key-value store; the individual components work. The failures come from misconfigured intervals, wrong backend choices for the actual state size, savepoint discipline that gets skipped under deploy pressure, sink implementations that don't hold up their end of the exactly-once contract. This is why the discipline of stateful streaming is so operational: designing pipelines with realistic state estimates, monitoring checkpoint metrics as primary SLIs, using savepoints for every schema-changing deploy, verifying end-to-end EOS with dedicated tests. The mature streaming teams are the ones with this operational muscle memory built in.
The Phase H roadmap from here is where the state discipline meets its two most important upstream/downstream contexts. M.42 (Change Data Capture) covers how to source streams from operational databases — the "how do we get consistent, ordered events into the streaming system?" question. CDC streams have their own state (the log-sequence-number of the last consumed change, the schema of the current source table), and getting that state discipline right is what makes CDC pipelines survive database failovers, schema changes, and full-table snapshots. M.43 (Real-Time Analytics) covers how the state produced by pipelines like M.41 becomes queryable at analytical latencies — Flink's queryable state, Materialize/RisingWave's incremental view maintenance, the interactive-query layer over streaming state. Together, M.39 (batch vs stream) + M.40 (windowing) + M.41 (state) + M.42 (sourcing) + M.43 (querying) is the complete Phase H — the full production discipline of streaming as a first-class data platform.
The terms every "why is our checkpoint duration climbing?" incident review, every Flink savepoint RFC, every "is our sink really idempotent?" audit assumes you already know.
Test the vocabulary. Click an answer; explanation drops in instantly.
Perfect. Checkpoints, backends, barriers, savepoints, exactly-once — the discipline that makes streaming state trustworthy under failure. Next up: change data capture (M.42), where the streams start meeting operational databases and the sourcing problem takes over.
The discipline that makes stream state trustworthy under failure.
Chandy-Lamport 1985 → Flink async barrier snapshotting 2015. Barriers flow through the DAG on the same channels as events. Each operator snapshots on seeing the barrier, forwards it, resumes processing. Non-blocking, consistent, exactly-once-capable — the specific mechanism at the heart of modern stateful streaming.
Memory for small state and low latency. Filesystem for medium state with durability. RocksDB for large state, growing cardinality, and incremental checkpoints. The choice is workload-specific; getting it wrong once produces the specific outages in §05. The backend is a first-class design decision, not a default.
Exactly-once semantics requires replayable sources AND coordinated checkpoints AND idempotent/transactional sinks. "Configured for EOS" is trivial; "actually EOS" requires end-to-end proof. The failure mode is silent duplicates in downstream systems. Verify with tests, not configuration flags.