Stateful Stream Processing

Stateful stream processing is processing in which the result for one record depends on records seen earlier, so the job must keep something between records. A stateless transformation — parse this field, drop that one, route by type — remembers nothing and can be restarted anywhere. Almost everything else remembers.

The distinction matters because state is the part you operate.

What a job remembers

• Window aggregates. A count or sum per key for every window that is still open — including windows held open by an allowed lateness horizon.
• Last-value state. The current status of each order, the latest reading per device — one entry per key, for as many keys as exist.
• Join buffers. One side of a stream-to-stream join held while waiting for the other, bounded by whatever time window the join allows.
• Deduplication sets. Identifiers already seen, retained for the replay horizon the pipeline supports.
• Source positions. Where each input had been read to, which must agree with the rest of the state for a restart to be correct.

Read that list and a pattern appears: every item has a retention rule, explicit or not, and the rule decides how the state behaves. Last-value state grows with the number of keys and levels off once the keys are known — a hundred sensors keep a hundred entries however often they report. Window state stays bounded while the watermark expires windows as fast as new ones open. A deduplication set with no horizon grows while new distinct identifiers keep arriving; a key space grows while new keys are retained without retiring old ones. Values that accumulate a history can grow even when the key count stays fixed. Lack of expiry alone does not imply continued growth. The questions are how many keys, how many entries per key, how long each entry lives, and whether the watermark is actually advancing.

How it survives failure

State has to outlive the process holding it, which is what checkpoints provide. Flink’s description of the purpose is exact: checkpoints “make state in Flink fault tolerant by allowing state and the corresponding stream positions to be recovered, thereby giving the application the same semantics as a failure-free execution.” The two halves — state and the positions it corresponds to — are what make a restart consistent rather than merely possible.

Taking a consistent picture without stopping the stream uses barriers. When a checkpoint begins, the sources record their offsets and “insert numbered checkpoint barriers into their streams,” which then “flow through the job graph, indicating the part of the stream before and after each checkpoint.” An operator with two inputs performs barrier alignment so the snapshot reflects consumption “up to (but not past) both barriers.” Writing the state out need not block processing either: 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.” That describes an aligned checkpoint. Flink also supports unaligned checkpoints, where a barrier overtakes buffered records and the in-flight records are stored in the checkpoint instead, so a backpressured job can still complete checkpoints at the cost of larger ones.

Three operational consequences

• State size is a main term in recovery cost. A restart restores state before processing resumes, and with a local state backend a job holding tens of gigabytes per instance has to download and load it first. The duration depends on the backend and the recovery path — Flink’s disaggregated state management can resume without a full download, paying in remote reads afterward — and on the bytes actually moved, the I/O and parallelism available, and how long it takes to get from the first processed record back to normal throughput. Convert the figure into a duration by timing an actual restore rather than by arithmetic on size, and compare it with the recovery target the service has promised — the two are often set independently and discovered to disagree during an incident.
• Checkpoint duration is a symptom. With aligned checkpoints, one slow or skewed input holds up the whole checkpoint, so rising times often indicate backpressure upstream of wherever the alert is configured. Slow synchronous or asynchronous snapshot phases and slow checkpoint storage raise the same metric without any input being late, so read the breakdown — start delay, alignment, sync and async durations — before deciding where the problem is. Either way the metric is an early signal rather than a constant to ignore.
• State grows from decisions, not traffic. A wider window or lateness horizon can raise retained state for the same input rate. Unbounded growth requires continued accumulation: new keys or entries retained without removal, or values that themselves keep growing. A fixed set of fixed-size values can remain bounded without expiry. When memory keeps increasing, inspect key and entry counts, value sizes, arrival rates, retention rules, and watermark progress.

What the guarantee covers

The strongest claim a stateful engine makes is narrower than it sounds. Flink’s phrasing for exactly-once is that “every event will affect the state being managed by Flink exactly once” — state, not the outside world. A record may be read twice after a restart; what the engine promises is that its own counters end up as if it had been read once. Anything written outward needs its own mechanism, either idempotent writes or a transaction committed with the checkpoint, as covered in two-phase commit sink.

How state fits with windows, watermarks, and end-to-end correctness is worked through in Stream Processing Semantics.

Further reading

• Apache Flink Documentation, Learn Flink: Fault Tolerance via State Snapshots
• Apache Flink Documentation, Checkpoints
• Apache Flink Documentation, Checkpointing under backpressure
• Apache Flink Documentation, Disaggregated State Management