Streaming State
Where the state physically lives, what makes it grow, how it is snapshotted and restored — and why state size, not throughput, is the number that decides whether a streaming job can be operated.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
How large will this job's state get, where does it live, and how long does it take to restore after a failure?
The on-call engineer, first — restore time is their recovery time. Then everyone downstream, because a job whose state cannot be restored quickly has an availability characteristic they will experience as a stale dashboard, a silent feature store and an alert that stopped firing.
One state entry: a key, a value and an expiry. The whole lesson is about counting these and multiplying by their size, because that product is the number that predicts everything operational about the job.
Treat state as an implementation detail of the engine. Set a checkpoint interval, point it at a bucket, and consider the matter handled — the framework does exactly-once state, after all.
The job has been running for three months and the checkpoint now takes longer than the checkpoint interval. Every snapshot overlaps the next, throughput falls, and no component has failed (Pipeline Reliability).
- The job has been running for three months and the checkpoint now takes longer than the checkpoint interval. Every snapshot overlaps the next, throughput falls, and no component has failed (Pipeline Reliability).
- A node dies. The restore reads the whole state from object storage before the job processes a single new record, and the recovery takes long enough that the lag accumulated during it is itself a second incident.
- A traffic spike needs more parallelism. Rescaling redistributes state across the new instance count, which is a bulk transfer proportional to state size — so the job is least able to scale at exactly the moment it needs to (Worker Scaling).
- The deduplication set holds every
event_idseen and has no TTL. Memory grows linearly with total events processed, the job is killed, restarts, refills, and is killed again (Memory Leaks: Growth That Does Not Come Back). - A code change alters the state's value type. The restored checkpoint no longer deserialises, and the only path forward is to start with empty state — which for a counting job means the counts restart from zero in production.
What is actually happening
- State lives local to the operator instance that owns its keys, because it is read and written on every record. Engines back it either with process memory (fast, bounded by heap) or with an embedded key-value store on local disk (larger, with its own compaction behaviour) (LSM Trees: Why Some Engines Favour Writes).
- Local storage is ephemeral, so durability comes from periodic snapshots to durable storage — object storage in nearly every deployment. A snapshot captures state plus the input positions that produced it, as one unit (Object Storage as Data Infrastructure).
- The snapshot must be *consistent across operators*: the state of a downstream operator and the state of an upstream one have to correspond to the same set of consumed input. Engines achieve this by flowing markers through the dataflow rather than by pausing everything, which is why snapshotting does not stop the job (Checkpointing).
- Incremental snapshots write only what changed since the last one, which decouples snapshot *duration* from total state size — but not restore duration, which still has to read enough to reconstruct the whole thing.
- State size is driven by distinct live keys × value size × retained time range. Throughput appears nowhere in that product, which is why a low-volume job can have a much larger state problem than a high-volume one.
- Expiry is the only mechanism that bounds state. It comes as a time-to-live on entries, as a window that eventually closes and purges, or as a key space that is finite by construction — and a job relying on none of these is unbounded whatever its current size suggests.
Where the bytes actually are
State has to be local. An operator reads and writes it on every single record, so a network round trip per access would set the throughput ceiling of the entire job — which is why no engine keeps working state in a remote database, however convenient that would be for recovery.
Local means ephemeral, so the durability story is a separate mechanism: periodically copy state, together with the input positions that produced it, to durable storage. The pairing is the whole trick. A snapshot of state without offsets is useless; a snapshot of offsets without state is worse than useless, because it looks valid.
The layout below is what that looks like on one instance. Two things are worth reading off it. First, the working set is on the instance and the durable copy is remote, so restore is a bulk read across the network before any record is processed. Second, the snapshot directory accumulates: retaining many historical checkpoints is a storage decision that is made once and revisited never.
one operator instance (owns a hash range of keys)
├── working state (memory, or an embedded KV store on local disk)
│ ├── agg/ key -> running value
│ ├── window/ (key, window) -> partial aggregate
│ ├── join/ key -> buffered records awaiting a counterpart
│ └── dedup/ event_id -> seen_at <- usually the biggest
└── input positions per assigned partition
durable snapshot storage (object storage)
└── checkpoints/
├── chk-0041/ state files + the offsets that produced them
├── chk-0042/
└── chk-0043/ <- restore reads this, then consumes from its offsets
restore = read chk-0043 fully -> resume from ITS offsets -> catch up on backlog
(no records are processed during the read)What actually makes state grow
The instinct is that state grows with traffic. It does not: it grows with distinct keys, with the size of each value, and with how long entries are kept. Throughput enters the formula only where the key is derived from the event itself — which is exactly the case for deduplication, and is why deduplication state is usually larger than everything else combined.
This has a practical consequence for capacity work. A job's record rate is on every dashboard and its key cardinality is on none, so the number that predicts operability is the one nobody is looking at. Adding an entry count metric per operator is a small change that turns a class of future outages into a slow-moving graph (Cardinality: The Label That Took Down Monitoring).
The overlap factor deserves its own mention. A sliding window of size ten sliding by one keeps ten live windows per key at all times, so the same data costs ten times the tumbling equivalent. That factor is a definition choice made in a single line of the job, and it multiplies memory, checkpoint size, restore time and rescaling cost together (Sliding Windows).
The dominant term everywhere. Entity keys (accounts, devices) grow slowly; event-derived keys (event ids, request ids) grow with total volume and are the usual cause of an unbounded job.
Multiplies the key term directly. Doubling a lateness allowance keeps roughly twice as many windows open per key, and doubling a dedup TTL roughly doubles the identifiers held.
A pure definition choice with a multiplicative effect. Tumbling windows have a factor of one; sliding windows have whatever the ratio says, and it is often chosen without noticing that it is a memory multiplier.
Storing an aggregate rather than the records that produced it is the highest-leverage single change available here, and it costs nothing except giving up the ability to explain the aggregate.
Affects durable storage rather than the job's memory, and accumulates silently because it is configured once at setup and never looked at again.
Deliberately near the bottom. Rate drives CPU and network; it drives state only through keys derived from the events themselves.
Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.
Relative weights for a typical keyed streaming job, given to establish an ordering rather than a magnitude. The ordering is the teaching: the thing everyone monitors (throughput) is last, and the thing nobody monitors (key cardinality) is first.
Snapshot and restore is the operation that decides everything
Every operational property of a stateful streaming job reduces to two durations: how long a snapshot takes, and how long a restore takes. The first decides whether steady-state throughput is achievable; the second is your recovery time, whether or not anyone has written a recovery objective.
Restores are where the surprises live, because restore duration is not measured in normal operation. The job reads the entire state from durable storage before processing a single record, and only then begins consuming the backlog that accumulated while it was down. If the job has little capacity headroom over its arrival rate, that second phase can be far longer than the first.
The failure table below is the sequence that actually happens, in the order it happens. Each row is survivable on its own; the reason streaming state incidents get large is that each one makes the next more likely — a slow checkpoint makes a restart more likely, a long restore makes the backlog larger, and a large backlog makes catch-up impossible without capacity that was sized for steady state.
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| Key cardinality grows past what the design assumed | Nothing. Throughput, lag and error rate are all normal for weeks. | State is sized by distinct keys, and no dashboard shows distinct keys. | Emit state entry count per operator and alert on a trend rather than a threshold. This is the only cheap intervention in the whole table (Cardinality: The Label That Took Down Monitoring). |
| Checkpoint duration approaches the checkpoint interval | Throughput sags gradually; occasional backpressure with no failing component. | Snapshots overlap, so the job spends an increasing share of its capacity copying state. | Reduce state (TTL, smaller values, fewer overlapping windows) rather than lengthening the interval — a longer interval only moves the cost into the next row (Backpressure). |
| An instance fails and the job restores | Output stops entirely for a period nobody predicted, then resumes. | Restore reads the entire state from object storage before any record is processed. | Measure restore duration in a drill and publish it as the job's recovery time. If it is unacceptable, the fix is less state, not a faster restore (RPO & RTO). |
| Backlog accumulated during the restore | Lag rises during the outage and then falls very slowly, or not at all. | Catch-up rate is capacity minus arrival rate. A job sized at its arrival rate has no catch-up rate at all (The Backlog Arithmetic: Four Levers and a Drain Time). | Hold explicit headroom, and be able to add parallelism — remembering that adding it redistributes state, which is itself proportional to state size (Headroom: The Capacity You Deliberately Do Not Use). |
| Someone adds a TTL to stop the growth | Memory stabilises; a month later a join stops matching a small share of records and a dedup starts letting a few duplicates through. | The TTL is shorter than the real tail of the distribution it is bounding — the mitigation traded a resource problem for a correctness problem. | Choose the TTL from the business question, route what it expires to a side output, and reconcile against batch so the loss is measured rather than assumed (Late Events). |
Whether a given engine supports incremental checkpoints, unaligned checkpoints, disk-backed state or state migration between versions has changed repeatedly across releases. Treat the capability list as something to verify in the current documentation for your exact version; the underlying trade — local speed against durable recoverability, snapshot cost against restore cost — does not change.
How to build it
Most important first.
- Estimate state before you write the job: count the distinct keys, multiply by the value size, multiply by how long entries live. If the answer is uncomfortable, the design is wrong now rather than in three months.
- Put a TTL on every state entry that is not bounded by a window, and choose it from the business question rather than from the memory budget — then check whether the memory budget can afford it, and treat a mismatch as a design problem rather than a tuning one (Data Retention).
- Keep the value small. Store the aggregate, not the records that produced it; store an identifier, not the payload. State is read and written per record, so value size costs on every access as well as in the snapshot (Projection Pushdown).
- Measure restore time deliberately, in a drill, before an incident makes you measure it accidentally. Restore time is your recovery time objective whether or not anyone has written it down (RPO & RTO).
- Prefer state that a replay can rebuild. If losing a checkpoint means replaying two days of log, that is a bad afternoon; if it means the counts are gone forever, that is data loss disguised as a configuration choice (Replay from the Log).
- Watch checkpoint duration as a first-class SLI. It is the earliest signal of state growth and it degrades smoothly for weeks before it degrades suddenly (Pipeline SLOs).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- A restored checkpoint restores state and input positions together, so the job resumes as though the records after that position had never been processed. That is what "exactly-once state update" means and it is the strongest guarantee in this module.
- That guarantee covers state only. Any effect already written to an external system before the failure has happened, and will happen again on reprocessing, unless that sink is transactional or idempotent (Exactly-Once: Input Consumption, State Update, Output Write).
- Snapshots are consistent across operators within one job. They are not consistent with anything outside it — not with another job, not with a database you read, not with a topic another consumer group is reading.
- What is not promised: that a snapshot is readable by a different version of the job; that restore is fast; that rescaling preserves per-key ordering during the transition; or that state expires when you think it does, since TTL is typically enforced lazily on access or during compaction rather than at the instant of expiry.
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Assert on state entry count per operator against the cardinality the design assumed. This catches the unbounded key space — the single most common cause of a streaming job becoming unoperable — months before it causes an outage.
- It misses state that is bounded in count but growing in value size, such as a session window accumulating records rather than an aggregate. Track bytes as well as entries.
- After every restore, reconcile the job's output for a closed period against a batch recomputation from the raw events. That is the only check that would notice a restore which quietly produced the wrong state (Reconciliation).
- Steady-state, snapshotting adds a small recurring cost and no freshness penalty worth reasoning about. The freshness event is the restore, which stops output entirely and then produces a catch-up burst.
- End-to-end freshness after a failure has a characteristic shape: flat while restoring, then a slope while the job consumes accumulated backlog at whatever margin its capacity has over arrivals, then normal. The slope's gradient is capacity minus arrival rate, and if that margin is thin the recovery is slow regardless of how fast the restore was (The Backlog Arithmetic: Four Levers and a Drain Time).
- This is why headroom is a freshness decision in streaming and not only a cost one: a job running at its capacity ceiling can never catch up from anything (Headroom: The Capacity You Deliberately Do Not Use).
- State serialisation is a schema, and it evolves under the same rules as any other: adding an optional field to a state value is usually compatible, removing or retyping one usually is not (Backward Compatibility).
- Changing the operator graph — adding a stage, changing a key, altering a window definition — generally invalidates the mapping between snapshot and job, so the snapshot cannot be restored into the new topology.
- The safe migration is a dual run from an earlier offset with separate state and a separate sink, compared before cutover. That makes log retention a hard bound on what changes are possible at all, which is an argument for longer retention that has nothing to do with disaster recovery (Retention and Replay).
- Restore state and offsets from the same checkpoint, always as a pair. Restoring one without the other silently doubles or silently drops everything between the two positions, and no check inside the job will notice.
- If the checkpoint is unusable, replay from the log with empty state. This is correct only for a job that is a deterministic function of the log — one that uses processing time or an external mutable lookup will produce a different answer than it did originally (Deterministic Replay: Making the Schedule Reproducible).
- If neither is available, the state is gone and the honest response is to say so: publish the gap, recompute what can be recomputed in batch from raw storage, and mark the affected range rather than letting a partially-rebuilt number stand (Data Incidents).
What can go wrong
- Checkpoint duration exceeding the checkpoint interval, so snapshots overlap and the job spends its capacity on snapshotting rather than on records.
- A restore that takes long enough for the accumulated backlog to exceed the job's ability to catch up, turning a short failure into an unbounded one (Cascading Failure: When the Response to Failure Causes More Failure).
- Skewed state: one instance holding most of the entries because one key holds most of the traffic, so that instance checkpoints slowly and restores slowly while the others idle (Data Skew).
- A snapshot written to storage the job cannot reach after a zone failure, which makes the recovery plan depend on a storage locality nobody documented (Multi-Zone Deployment).
- The mitigation failing: a TTL added to bound memory silently expires join buffers or dedup entries that were still needed, converting a resource problem into a correctness problem that produces no error at all.
- "Checkpointing makes the job exactly-once." It makes state updates exactly-once with respect to input consumption. Output written to an external system is a separate problem with a separate mechanism (Exactly-Once: Input Consumption, State Update, Output Write).
- "Incremental checkpoints mean state size stopped mattering." They decouple snapshot duration from total size. Restore duration, rescaling cost and memory pressure are all still proportional to it.
- "State is small because throughput is low." State is sized by distinct keys and retention, not by rate. A job seeing a handful of records a second about millions of distinct identifiers has a state problem and no throughput problem.
- "We can always replay to rebuild state." Only within retention, and only if the computation is deterministic given the log. Both conditions are commonly false and neither fails loudly (Replay from the Log).
- State is a copy of data with the same obligations as any other copy. A deduplication set holds identifiers, a session window holds behavioural records, and a join buffer can hold complete personal records — none of which appears in a data catalog or a retention policy (PII in Pipelines).
- A deletion request must reach state as well as storage, and there is usually no mechanism to do it: state is keyed by the job's key, not by subject, and snapshots are immutable blobs. The practical mitigation is a TTL short enough that the obligation expires with the entry, decided deliberately rather than discovered during an audit (Deletion Requests).
Operating it
- Checkpoint duration, size and failure count over time, per job — plotted over weeks, because the signal is a trend rather than a spike.
- State entry count and state bytes per operator and per instance, since the per-instance view is the only one in which skew is visible.
- Restore duration on every restart, recorded as an event so that the recovery time objective is a measured number rather than an aspiration (RPO & RTO).
- Local disk usage on the instances holding state, which is the resource that runs out first in disk-backed configurations and is rarely on a streaming dashboard (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- At 10x throughput with a fixed key space, state is roughly unchanged and the job scales like a stateless one. This is why the state problem is invisible during most growth.
- At 10x key cardinality, everything operational degrades at once: snapshot duration, restore duration, rescaling cost and memory pressure. Nothing about the record rate has changed, and every graph an operator normally watches looks fine (Cardinality: The Label That Took Down Monitoring).
- At 100x, the embedded state store becomes the dominant component and its behaviour — compaction, write amplification, random-read locality — sets the job's performance, which is a database problem wearing streaming clothes (Write, Read and Space Amplification).
- Four terms: memory or local disk to hold the state, storage to retain snapshots, network to write them, and compute spent on snapshotting and on the state store's own compaction. All four are driven by state size rather than by throughput.
- Snapshot retention is a quiet accumulator — keeping many historical checkpoints multiplies storage by their number and is usually configured once and never revisited (Storage Lifecycle).
- The largest indirect cost is the headroom you must hold so the job can catch up after a restore. A job sized exactly for its arrival rate cannot recover, so the recovery requirement sets the capacity, not the steady state (Capacity or Efficiency: Which Problem Are You Solving?).
- Frequent checkpoints reduce reprocessing after a failure and cost steady-state throughput and storage. Infrequent ones do the reverse. The right interval follows from how expensive your reprocessing is, and there is no setting that is good at both.
- Disk-backed state holds far more than memory-backed state and costs per-record access latency plus the store's own compaction overhead. Memory-backed state is faster and puts a hard ceiling on the key space.
- A short TTL buys bounded memory and costs correctness for anything slower than the TTL. A long one buys correctness and costs memory in direct proportion. This is the clearest example in the module of a knob where both directions are genuinely wrong for some case.
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALLocal state, durable snapshots taken with input positions, and expiry as the only bound are common to every stream processor. The product "distinct keys × value size × retention" predicts operational behaviour regardless of which engine computes it.
- ENGINE-SPECIFICFlink can back state with a disk-based store and supports incremental checkpoints, so it tolerates state larger than memory at the cost of that store's compaction behaviour; Kafka Streams backs local stores with compacted changelog topics, so its restore is a topic replay whose duration depends on the changelog rather than on a snapshot; Spark Structured Streaming checkpoints per micro-batch, which ties snapshot frequency to batch interval.
- CLOUD-SPECIFICSnapshots land in object storage, so restore throughput is bounded by that service's read behaviour and by whether the job restarts in the same zone as the data. A cross-zone restore reads over a different path than the one you benchmarked.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns the snapshot algorithm itself — how a set of communicating operators agrees on a cut of the computation without stopping, and why that is a genuinely hard problem rather than a scheduled copy.
- — DevOps / Production Engineering owns the drill: restore time is only a recovery objective if someone has actually restored, on a schedule, and recorded how long it took.