Operating Streams: Backpressure and Replay

A stateful streaming job is rarely lost to a single dramatic failure. More often a downstream database slows down, the job falls behind, checkpoints start timing out, the job restarts from an older checkpoint and falls further behind, and nobody notices until the input topic’s retention window is closing on the oldest unprocessed record. A few weeks later someone needs to change a field or fix a bug in an aggregation, and the question is whether that can be done without discarding weeks of window state. This article is about that operating life: how to read lag and saturation, how to replay history without producing duplicate effects, how schema changes reach a running job, how state is migrated, and how recovery and rollouts are carried out.

It builds on Stateful Stream Processing and Checkpoint Recovery, which established that a checkpoint pairs operator state with source positions, that a restart replays input since the last completed checkpoint, that sinks need either a commit tied to the checkpoint or idempotent writes, and that exactly-once guarantees end at the boundary of what the engine can see. The earlier article Event Time, Watermarks, and Late Data supplies the notions of watermark and allowed lateness used throughout. Apache Flink is the main example because its documentation is explicit about operations; Apache Spark Structured Streaming and Apache Kafka are compared where they differ. Documentation was read in October 2026: Apache Flink stable (2.3 line), Apache Spark 4.2.0, and Apache Kafka 4.3. All rates, durations, and sizes in worked examples are assumptions chosen for arithmetic, not measurements.

Lag and saturation

Four different lags

The word lag covers several quantities that move independently, and an alert on the wrong one misleads. The following table separates the ones that matter for a Kafka-fed job.

Quantity	Definition	Where to read it	What it tells you
Offset lag	Log end offset minus the consumer’s position, per partition	kafka-consumer-groups.sh --describe (columns CURRENT-OFFSET, LOG-END-OFFSET, LAG); consumer metric records-lag-max; Flink Kafka source pendingRecords	How many records are waiting; not how old they are, and not how long they will take
Committed-offset lag	Log end minus the offset the group last committed	Same tool; see committed offset and consumer lag	For Flink, steps forward only at checkpoint completion, so it overstates true lag between checkpoints
Event-time lag	Current time minus event time of what is being emitted or of the watermark	Flink Kafka source currentEmitEventTimeLag and watermarkLag	How stale results are in the clock consumers care about; also grows when the source is idle
Processing state	Share of time busy, idle, or backpressured per subtask	Flink busyTimeMsPerSecond, idleTimeMsPerSecond, backPressuredTimeMsPerSecond	Where the bottleneck is, not whether lag exists

Offset lag is a count, and a count converts to a delay only through a rate. A lag of 1.2 million records is a few seconds behind a pipeline processing 400,000 records per second and nearly an hour behind one processing 350 per second. Offset distance also need not equal the number of readable records when compaction or transactions leave gaps, a point made in the Kafka article. Event-time lag is closer to what consumers experience, but it has its own blind spot, since a watermark that stops advancing reports a growing lag even though the job is healthy and the input is simply empty. Use both, and read them together with the checkpoint metrics.

The fluid model: lag is an integral

The simplest useful model treats records as a fluid. If the arrival rate is \(\lambda\) records per second and the processing capacity is \(\mu\), the backlog changes at \(\lambda – \mu\) while it is positive, and a backlog of size \(B\) clears in \(B / (\mu – \lambda)\) seconds when \(\mu > \lambda\). The formula says three things. Capacity equal to arrival rate stops the growth and never drains the backlog. The drain time is inversely proportional to the headroom, so a job at 95 percent utilization needs four times as long to catch up as one at 80 percent, for the same backlog. And a backlog accumulated at the failure of a few minutes can take many times that long to clear. The following script, written for this article, evaluates the formula for several situations; every input is an assumption.

"""Arithmetic models for lag, saturation, retention, and replay effects. All rates are assumptions."""

def catch_up(arrival, capacity, backlog):
    """Seconds to drain `backlog` records when arrival and capacity are constant (records/s)."""
    net = capacity - arrival
    return float("inf") if net <= 0 else backlog / net

print("== 1. Outage then catch-up (assumed: 8,000 rec/s arrival, 20 min outage)")
arrival, outage = 8000, 20 * 60
backlog = arrival * outage
print(f"backlog after outage: {backlog:,} records")
for cap in (8000, 9000, 10000, 12000, 16000):
    t = catch_up(arrival, cap, backlog)
    print(f"  capacity {cap:>6,} rec/s -> net drain {cap-arrival:>6,}/s, catch-up {'never' if t==float('inf') else f'{t/60:6.1f} min'}")

print("\n== 2. Retention race (assumed: consumer stopped for D minutes, then drains at capacity)")
# While stopped, the oldest unread record ages 1 s per s. After restart it ages (1 - cap/arrival) s per s
# (negative when cap > arrival, i.e. the backlog shrinks). Data is lost if that age ever exceeds retention.
for retention_h in (24, 1, 0.25):
    for D in (20, 90):
        peak_age_min = D                       # age at the moment the consumer restarts (cap >= arrival)
        status = "records lost" if peak_age_min > retention_h * 60 else f"margin {retention_h*60-peak_age_min:6.1f} min"
        print(f"  retention {retention_h:>5} h, outage {D:>2} min -> oldest unread age at restart {peak_age_min} min: {status}")
cap = 7000
rate = 1 - cap / arrival
t_h = (24 * 3600 - 20 * 60) / rate / 3600
print(f"  capacity {cap:,} < arrival {arrival:,}: after a 20 min outage the oldest unread record still ages {rate:.3f} s per s; it reaches 24 h after about {t_h:,.0f} more hours")

print("\n== 3. Hot partition (assumed: 12 partitions, 8,000 rec/s total, one partition carries 40%, one consumer thread handles 1,000 rec/s of one partition)")
parts = [0.40] + [0.60 / 11] * 11
for i, share in enumerate(parts[:3]):
    rate = 8000 * share
    print(f"  partition {i}: arrival {rate:7.1f}/s vs 1000/s -> lag change {rate-1000:+8.1f}/s")
print(f"  total arrival 8,000 vs total capacity {12*1000:,}: aggregate looks healthy ({8000/12000:.0%} utilised), hot partition does not")

print("\n== 4. Replay: sink behaviour when an old pane arrives after a newer one")
panes_live = [("w0", 4, 1), ("w0", 5, 2)]            # (window, count, pane_version)
replay_after = [("w0", 4, 1)]                       # a backfill/restarted job re-emits the older pane later
def apply(policy, seq):
    table, log = {}, []
    for w, c, v in seq:
        if policy == "append": log.append((w, c, v))
        elif policy == "upsert_last_write_wins": table[w] = (c, v)
        elif policy == "upsert_version_guard":
            if w not in table or v > table[w][1]: table[w] = (c, v)
    return log if policy == "append" else table
for p in ("append", "upsert_last_write_wins", "upsert_version_guard"):
    print(f"  {p:24s}", apply(p, panes_live + replay_after))

print("\n== 5. Burst against fixed capacity (assumed: 5 minute burst at 16,000 rec/s, capacity 10,000, baseline 8,000)")
burst_backlog = (16000 - 10000) * 5 * 60
print(f"  backlog at end of burst: {burst_backlog:,} records; drains at {10000-8000:,}/s -> {burst_backlog/2000/60:.0f} min")

print("\n== 6. Rebuilding state by replaying retained input (assumed: 7 days retained at 8,000 rec/s)")
history = 8000 * 7 * 24 * 3600
print(f"  retained records: {history:,}")
for cap in (10000, 16000, 40000, 80000):
    t = history / (cap - 8000)          # live traffic keeps arriving while history is replayed
    print(f"  capacity {cap:>6,} rec/s -> {t/3600:8.1f} h ({t/86400:5.1f} days)")

The output is:

== 1. Outage then catch-up (assumed: 8,000 rec/s arrival, 20 min outage)
backlog after outage: 9,600,000 records
  capacity  8,000 rec/s -> net drain      0/s, catch-up never
  capacity  9,000 rec/s -> net drain  1,000/s, catch-up  160.0 min
  capacity 10,000 rec/s -> net drain  2,000/s, catch-up   80.0 min
  capacity 12,000 rec/s -> net drain  4,000/s, catch-up   40.0 min
  capacity 16,000 rec/s -> net drain  8,000/s, catch-up   20.0 min

== 2. Retention race (assumed: consumer stopped for D minutes, then drains at capacity)
  retention    24 h, outage 20 min -> oldest unread age at restart 20 min: margin 1420.0 min
  retention    24 h, outage 90 min -> oldest unread age at restart 90 min: margin 1350.0 min
  retention     1 h, outage 20 min -> oldest unread age at restart 20 min: margin   40.0 min
  retention     1 h, outage 90 min -> oldest unread age at restart 90 min: records lost
  retention  0.25 h, outage 20 min -> oldest unread age at restart 20 min: records lost
  retention  0.25 h, outage 90 min -> oldest unread age at restart 90 min: records lost
  capacity 7,000 < arrival 8,000: after a 20 min outage the oldest unread record still ages 0.125 s per s; it reaches 24 h after about 189 more hours

== 3. Hot partition (assumed: 12 partitions, 8,000 rec/s total, one partition carries 40%, one consumer thread handles 1,000 rec/s of one partition)
  partition 0: arrival  3200.0/s vs 1000/s -> lag change  +2200.0/s
  partition 1: arrival   436.4/s vs 1000/s -> lag change   -563.6/s
  partition 2: arrival   436.4/s vs 1000/s -> lag change   -563.6/s
  total arrival 8,000 vs total capacity 12,000: aggregate looks healthy (67% utilised), hot partition does not

== 4. Replay: sink behaviour when an old pane arrives after a newer one
  append                   [('w0', 4, 1), ('w0', 5, 2), ('w0', 4, 1)]
  upsert_last_write_wins   {'w0': (4, 1)}
  upsert_version_guard     {'w0': (5, 2)}

== 5. Burst against fixed capacity (assumed: 5 minute burst at 16,000 rec/s, capacity 10,000, baseline 8,000)
  backlog at end of burst: 1,800,000 records; drains at 2,000/s -> 15 min

== 6. Rebuilding state by replaying retained input (assumed: 7 days retained at 8,000 rec/s)
  retained records: 4,838,400,000
  capacity 10,000 rec/s ->    672.0 h ( 28.0 days)
  capacity 16,000 rec/s ->    168.0 h (  7.0 days)
  capacity 40,000 rec/s ->     42.0 h (  1.8 days)
  capacity 80,000 rec/s ->     18.7 h (  0.8 days)

Section 1 shows an outage of 20 minutes at 8,000 records per second producing 9.6 million records of backlog. At 10,000 records per second the job needs 80 minutes to clear it, and at 9,000 it needs 160 minutes; a job with exactly the arrival rate in capacity never recovers. Section 2 compares the age of the oldest unread record with topic retention. Kafka deletes old log segments by time or size independently of consumers: the documented default for retention.ms is seven days, described as an SLA on how soon consumers must read their data. In the model, an outage of 90 minutes against a one-hour retention loses data, because the oldest unread record is already older than the log keeps. A consumer slower than the producers by a small margin ages the oldest unread record only slowly, as the last line of section 2 shows, which is why this failure usually appears as a slow decline rather than a sudden one, and why alerting on the age of the oldest unread record is more informative than alerting on offset lag alone. Section 5 shows a burst: five minutes at 16,000 records per second against 10,000 capacity leaves 1.8 million records that take 15 minutes to drain.

Saturation: finding the constraint

A job is saturated when some stage cannot keep up with what arrives at it. Flink’s monitoring documentation defines three per-subtask metrics that add up to about 1,000 ms per second: back-pressured time, idle time, and busy time. A task is considered back-pressured when it has no available output buffers, and idle when no input is available. The documentation states the direction plainly: if a task shows back pressure, it is producing faster than the operators after it can consume, so a source showing back pressure points at something downstream, and the constraint is found by following the signal forward. The three numbers are time averages over a few seconds: the documentation’s example is two subtasks with the same busy time of about 500 ms per second, one at constant half load and one alternating between fully loaded and idle. Backpressure as a general mechanism is described in the glossary; this section treats it operationally.

The mechanism behind the metric is bounded buffering. Flink’s network stack uses credit-based flow control: a receiver announces its free buffers to the sender as credits, a sender forwards a buffer only when it has credit, and a receiver that cannot keep up eventually reports zero credits, which stops the sender. Because every operator is a receiver for the one before it, the stall propagates upstream hop by hop until it reaches the source, which can then only read more slowly or not at all. That is the property that keeps the job from running out of memory, and it is also why backpressure is not itself a failure: it is the engine slowing the intake to the speed of its slowest stage. The failure is what accumulates behind the source, which for Kafka is offset lag, and what the stall does to checkpoint barriers.

A systematic reading goes through the following questions, in order.

1. Is the source backpressured, and by what? Follow back-pressured time downstream until you reach the first subtask that is busy and not back-pressured; that stage is the bottleneck. A busy subtask is not told what it is busy with, so identify whether the time is CPU, a blocking call to a database, a rate-limited external API, or state access.
2. Is the load uniform across subtasks? One subtask much busier than its peers is skew, not under-provisioning. Flink’s metrics documentation notes that a long checkpoint start delay on only one task most likely points to data skew. On the Kafka side, one hot partition can dominate completion time, as in the data skew entry, and adding consumers cannot divide the partition: in a consumer group a partition is read by one member, and a job parallelism above the partition count leaves some readers idle.
3. Does throughput match the required rate? Compare records per second out of each stage with the arrival rate, not with the stage’s own history. A stage that is idle most of the time while backlog exists is waiting on something upstream, such as a source that fetches slowly.
4. Is the sink the constraint? A downstream database with a connection limit or a rate-limited API caps the whole job regardless of how much compute it has. Batching writes, increasing write parallelism within the destination’s limit, or buffering in a durable queue are the options; adding stream workers is not.

The next model makes the hot-partition case explicit. With 12 partitions, 8,000 records per second in total, one partition carrying 40 percent of the traffic, and a consumer thread able to process 1,000 records per second from one partition, the aggregate is 67 percent utilized (8,000 of 12,000), yet the hot partition receives 3,200 per second and its lag grows by 2,200 per second while the other eleven drain at 564 per second. The aggregate figure says healthy; the per-partition lag says otherwise. This is why lag alerts must be per partition, and why a remedy has to address the key distribution (a salted key or a pre-aggregation step, at the cost of the per-key ordering contract described in Designing Topics and Event Contracts) and not the consumer count.

What saturation does to the rest of the job

Backpressure delays checkpoint barriers, because a barrier travels with the records through the same full buffers. Flink’s documentation says that under heavy backpressure the end-to-end time of an aligned checkpoint is dominated by barrier propagation, visible as high start delay and alignment time. The consequences compound. A checkpoint that exceeds its timeout (ten minutes by default) is aborted. With the default of zero tolerated checkpoint failures, the first aborted checkpoint fails the job, which restarts from the previous completed checkpoint, now older, and replays more records into an already saturated pipeline. Three remedies are documented: remove the backpressure source, reduce the buffered in-flight data, or enable unaligned checkpoints. Buffer debloating, available since Flink 1.14 and enabled with taskmanager.network.memory.buffer-debloat.enabled, adjusts the amount of in-flight data so that its consumption time matches a configured target; it caps the used buffer size without reducing memory use, and the documentation lists cases it does not handle well, such as subtasks with several inputs of very different throughput. Unaligned checkpoints let barriers overtake buffered records and store them in the checkpoint, so checkpoint duration becomes independent of throughput, at the price of more I/O to checkpoint storage, no help if that storage is the bottleneck, and a behavior change for operators that rely on the latest watermark during recovery.

Two Kafka consumer settings matter when processing slows. max.poll.records (default 500) limits how many records one poll returns, and max.poll.interval.ms (default five minutes) is the longest allowed gap between polls: a consumer that exceeds it is considered failed and its partitions are reassigned in a rebalance. A slow sink inside a hand-written consumer loop can therefore trigger rebalances, which pause the group and worsen the lag; engines that fetch in a separate thread, or that decouple fetching from processing with bounded buffers, avoid the coupling. For Spark, the Kafka source offers rate limiting with maxOffsetsPerTrigger, which bounds the offsets processed per trigger and splits them proportionally across partitions; it keeps a catch-up batch from becoming a single huge one, at the price of a longer catch-up.

When the work cannot be deferred and capacity cannot grow, the remaining choices are to decide what not to process: shed load, throttle producers, or degrade a result. These are product decisions to make before an incident. The arithmetic above gives the numbers they depend on.

Replay and duplicate effects

Four reasons to replay

Replay means reading retained input again from an earlier position. It happens for four reasons with different safety needs. Automatic recovery restarts from the last checkpoint and replays a bounded span; the checkpoint article covered it. Bug repair rewinds to a point before a defect and recomputes. Backfill initializes or extends a derived dataset with history. Dead-letter reinjection replays records that were set aside after a failure, as in the dead letter queue entry. Only the first is handled by the engine; the others are operator actions whose effects on sinks must be designed.

How a rewind is expressed differs by engine. Flink’s Kafka source takes an offsets initializer for the starting position: the committed group offsets, the earliest or latest offset, or a timestamp, and the default if none is specified is the earliest offset. The starting position applies when the job starts without state; a job restored from a checkpoint or savepoint continues from the positions in that state. Rewinding a job therefore means starting a new job without the old state, or restoring from an earlier snapshot. Spark states the same rule: startingOffsets applies only when a new streaming query is started, and resuming always picks up where the query left off, so replay requires a new checkpoint location. Both rules protect a running job from accidentally replaying, and both mean that a rewind is a new job with new state to seed and new output to reconcile.

What a replay does to each kind of output

Output	What a replay produces	What makes it safe
Append-only table	Every replayed record appended again	Stage and merge on a stable event identity, or deduplicate within a horizon that covers the replay depth; see event deduplication
Upsert table keyed by group and window	Rows rewritten with the replayed values	A version guard so that an older result cannot overwrite a newer one; see upsert and merge
Kafka output topic	Duplicate records unless transactional	Transactional sink and read_committed readers, or consumers that deduplicate by event identity
Incremental window results (late firings)	Earlier panes re-emitted, possibly after later ones	Replace by key and window with a pane version; see trigger and accumulation mode
External effect (email, payment)	The effect repeated	Receiver-side idempotency key; disable the effect in a replay job

Two details cause most replay incidents. The first is ordering across writers. A backfill or replay job and the live job write to the same table at the same time, and the replay’s older value for a window arrives after the live job’s newer one. The last section of the script above models this with two panes for one window: the live job writes count 4 as version 1, then count 5 as version 2, and a replay re-delivers version 1 afterward. An append sink holds all three rows. An upsert that lets the last write win ends with count 4, which is wrong, while an upsert with a version guard keeps count 5. Only the guarded version tolerates concurrent writers, which is why the data backfill entry warns that publishing an old backfill last can overwrite fresher data even when every transaction is atomic. The second detail is that a version has to mean something: when replayed code changes the value for a window deliberately, the replay must carry a higher revision and a restatement notice, as in delivery revision and restatement.

The second detail is determinism of the replay itself. As the previous articles noted, a replay does not see the same arrival interleaving as the live run, so which records are classified as late can differ, and watermarks advance in large jumps as history is consumed quickly. A replay should be compared with an independent total from the source, with a reconciliation on identifiers, not with the live output. A good replay design also bounds its own cost: replaying seven days of history at 8,000 records per second is 4.84 billion records, and section 6 of the model shows that this takes 28 days at 10,000 records per second of capacity (the live traffic continues to arrive), 7 days at 16,000, about 42 hours at 40,000, and about 19 hours at 80,000. Replaying deep history is a capacity project, and the topic’s retention bounds what can be replayed at all.

Schema changes

A schema change reaches a streaming job in three places, with different rules for each.

The input records

The records a job reads were written under a contract (see Designing Topics and Event Contracts), and the topic holds records written under several versions of it. Compatibility direction decides the safe upgrade order: backward compatibility lets a new reader read old data, so readers can move first, whereas forward compatibility lets an old reader read new data, so producers can move first, as the schema evolution and compatibility entry explains. A streaming job adds a constraint that batch jobs rarely have: it is a reader of retained history. A replay or a restart from an old checkpoint reads old records, so the job must read every version still retained, which points to transitive backward compatibility and to keeping the old schemas available. A producer-side change that is structurally compatible can still change meaning, such as dollars becoming cents under the same type, and nothing in the registry catches it. In a job with window state, the damage is persistent: window aggregates now mix units and the mix is stored in state. The remedy is a new field or a new topic with an explicit cutover, plus a recomputation of the affected range.

A record the job cannot decode is a different problem. If deserialization throws, the task fails, the job restarts from the last checkpoint, reads the same record, and fails again; with the default exponential-delay restart strategy the loop slows down but does not end. The usual design catches the failure at the edge of the job, writes the raw record with its source position and error to a dead-letter destination, counts it, and continues. That choice must come with an alert and an owner, because a dead-letter destination nobody reads is indistinguishable from data loss.

The state

State is persisted data with its own schema, and it is checked at restore. Flink’s documentation lists what can evolve: POJO types may gain or lose fields and Avro types may change as Avro’s schema-resolution rules permit, while declared field types and the class name of a POJO cannot change, and Kryo-serialized types cannot evolve at all because the framework cannot verify compatibility. The key of keyed state cannot change: if a key field were dropped, previously distinct keys could collapse, and the framework has no way to merge their values, while RocksDB compares keys by their serialized bytes. The procedure is to take a savepoint, change the state type in the application, and restore; the framework migrates each state when it is first accessed. Internal state of built-in operators follows the operator’s input and output types, so changing the type of the stream entering a window operator changes the type of its state, a point the upgrade documentation tabulates for reduce, window, join, and aggregation functions.

Spark states the rule more strictly. Between restarts from the same checkpoint location, additions, deletions, and schema modifications to the stateful operations of a query (streaming aggregation, deduplication, stream-stream joins, and arbitrary stateful operations) are not allowed. Its documented workaround is to encode the state yourself in bytes with a format that supports schema migration, for example Avro, so that the binary state always restores. In both engines the lesson is the same: choose state types that can evolve before the first deployment, because the first change arrives when the state is largest.

The output

Consumers of the job’s output have their own readers. Adding a column is compatible for consumers that name columns and breaks consumers that read by position; changing the meaning of a measure breaks everyone while passing every structural check. The data contract entry covers what a producer promises. For a job, the practical rule is to treat output changes as a versioned contract: introduce the new shape next to the old one, move consumers, and retire the old one when their owners confirm, with a restatement notice if values that were published change.

State migration

Several situations require changing state itself, not only the code that uses it. The table lists them with the options the documentation supports.

Change	Does the existing state survive?	Options
Logic change, same state types, operators have stable identifiers	Yes	Savepoint, deploy, restore
Operators have no explicit identifiers	Only if the generated identifiers did not change	Set uid on every stateful operator before the first deployment; a topology change otherwise breaks matching
New stateful operator	The new operator starts empty	Accept a warm-up period, bootstrap it by replay, or write initial state with the State Processor API
Removed stateful operator	Its state cannot be matched and is discarded only if you allow it	Restore with --allowNonRestoredState after confirming nothing else needs it
State type changed	Only for POJO or Avro types within their rules	Evolve the type, or define a second state and migrate in code, or use the State Processor API
Key changed	No, keys cannot be migrated in place	Rebuild from retained input, or rewrite state offline
Parallelism changed	Yes, within the maximum parallelism	Restore with the new parallelism
Maximum parallelism changed	Not through a normal restore	State Processor API, which can adjust maximum parallelism, or rebuild
Engine version changed	Usually, through a savepoint; not guaranteed for SQL and Table API plans	Take a savepoint on the old version and restore on the new; test restore first

Flink’s savepoint documentation fixes the central rule: a savepoint maps operator identifiers to state, and the identifiers are generated from the program’s structure unless set explicitly, so an unchanged program restores from its own savepoint, while a changed one restores only if its stateful operators carry the same explicit uid values. Adding a stateful operator initializes it empty, removing one fails the restore unless you pass --allowNonRestoredState, and reordering operators is safe only with identifiers. Changing the parallelism is a restore with a new parallelism. For Table API and SQL jobs, the documentation warns that a change to either the query or the Flink version can make state incompatible, because the planner may choose different operators; patch versions are kept state-compatible by intent.

The State Processor API reads, writes, and modifies savepoints and checkpoints with batch programs. The documentation lists uses that a normal restore cannot do: bootstrapping a new job’s state from a batch dataset, correcting inconsistent state entries, changing the data type of state, adjusting the maximum parallelism of operators, splitting or merging operator state, and reassigning identifiers. It operates on savepoints. Flink’s comparison table shows that only canonical savepoints support changing state backend or writing with this API, while the faster native format does not.

The alternative to migrating state is rebuilding it from the log. That works only if the input still exists for as long as the state’s lifetime, and it costs the capacity calculation shown earlier: rebuilding state that spans seven days of input takes days even at fair headroom. State that spans longer than the topic’s retention cannot be rebuilt this way at all. This is the argument for treating savepoints as assets, for keeping retention comfortably above the longest state lifetime when rebuild is the planned fallback, and for writing down which of the two paths each job relies on.

Recovery and rollout

An in-place upgrade with savepoints

The documented path for upgrading a job, or moving it to a newer engine version, is built on savepoints. The following sequence is Flink’s documented procedure, annotated with the checks that make it safe.

1. Take a savepoint and stop the job. Flink’s upgrade documentation describes stopping a job together with a savepoint as a single action (bin/flink stop with a savepoint path), after which no other checkpoints are taken. Taking a savepoint and continuing leaves the job emitting output that a restore from that savepoint will emit again, which is the same replay problem as before. Record the savepoint path and the source positions it contains.
2. Check the sink’s behavior at stop. For a transactional sink, know what happens to the pending transaction when the job stops: whether it is committed with the final snapshot or aborted and re-created on restart. The documentation does not allow assuming; test it.
3. Deploy the new version with the same operator identifiers. Do not change the engine version and the connector version in the same step: Flink’s Kafka connector upgrade notes advise upgrading them separately.
4. Restore with the default claim mode. bin/flink run -s restores; in the default NO_CLAIM mode Flink leaves the savepoint files in your control and forces the first successful checkpoint to be a full one, after which you can delete the savepoint. In CLAIM mode Flink owns the files and may delete them, so it is not safe to start two jobs from the same snapshot.
5. Watch recovery, not just startup. Restart count, checkpoint success and duration, offset lag falling at the expected net rate, the watermark advancing, late-record drops, and sink commits.

Keep the savepoint until the new version has run long enough to be trusted, because it is the way back. Going back has a cost that restoring state does not remove: the input processed since the savepoint was taken is processed again, and output already committed stays committed, so the sink must tolerate the re-application. The rollback is therefore a replay, with the properties of the previous section.

Running two versions side by side

A savepoint can also start a second job. The savepoint documentation lists forking among its purposes: stop-and-resume, fork, and update. A forked job gets its own outputs; it must not share the original’s sinks or transactional identifiers. For the Kafka sink the documentation requires a transactionalIdPrefix unique across applications on the cluster, and a job with an idempotent sink would overwrite the original’s rows if it wrote the same keys. In Spark a new query needs its own checkpoint location, for the reasons given in the replay section. A side-by-side run is the streaming counterpart of a shadow deployment: both versions read the same input, the new one writes to a separate destination, and a reconciliation of the two outputs by key and window shows where they differ before any consumer is moved. A canary deployment differs in that it exposes real consumers to the candidate. For streams that can mean routing a subset of keys to the new version, which is only valid if the subset is stable and each key’s history is entirely in one version.

Engine and connector versions

Flink’s documentation distinguishes API compatibility by stability annotation: for public APIs, patch releases are binary compatible, minor releases are source compatible, and major releases guarantee neither; evolving and experimental APIs carry weaker guarantees. State compatibility is a separate question answered by testing a restore. The same reasoning appears more generally in the version skew entry: any system whose parts deploy independently has a range of working combinations, and the question is whether it is written down. For a stream job the parts are the engine, the connector, the broker version, the schema of the records, and the schema of the state. Write down the supported combinations, and upgrade one dimension at a time.

Spark and the restart rules

For Spark Structured Streaming the guide lists which changes are allowed between restarts from the same checkpoint location. Changing the number or type of input sources is not allowed, and changing subscribed topics is generally not allowed because the results are unpredictable. Rate limits such as maxOffsetsPerTrigger can be added, changed, or removed. Some sink changes are allowed (file sink to Kafka sink) and others are not (Kafka sink to file sink), and changing a file sink’s output directory is not allowed. Filters can be added or removed, projections with the same output schema can change, and changes to stateful operations are not allowed. Anything outside the list means a new checkpoint location, which means a new query, state rebuilt by replay, and the cutover described above.

A hypothetical cascade, with numbers

The following scenario is hypothetical and uses the assumptions of the model above: 8,000 records per second arrive, the job normally processes 10,000 per second, checkpoints run every minute with the default ten-minute timeout and zero tolerated failures, and the topic retains 24 hours.

Step	What happens	Consequence
1	A database behind the sink slows, and the sink can absorb 5,000 records per second	The job’s throughput falls to 5,000 per second; the sink subtasks are busy and everything upstream becomes back-pressured
2	Offset lag grows by 3,000 per second	After 30 minutes the lag is 5.4 million records; event-time lag is climbing
3	Barriers queue behind full buffers; checkpoint start delay rises	Checkpoints take longer; the first that exceeds ten minutes is aborted
4	With zero tolerated failures, the job fails over	It restarts from the last completed checkpoint and replays records into the saturated pipeline; transactions open at failure may expire if the producer timeout is short
5	The database recovers and the sink returns to 10,000 per second	Net drain is 2,000 per second, so the 5.4 million backlog plus the replay needs about 45 minutes or more
6	An alert on the age of the oldest unread record	Compare that age with the 24-hour retention; with margin, no data is lost, and the cost is delayed results

Each step corresponds to a setting or a metric that can be chosen in advance: sink capacity and headroom (step 1 and 5), per-partition lag alerts (2), checkpoint timeout and tolerated failures (3 and 4), transaction timeout (4), and retention (6). The scenario is not a prediction; its purpose is to show that the pieces interact, so tuning one in isolation (raising the checkpoint tolerance, for example) moves the failure elsewhere.

Runbook

A runbook for a stateful job is a short table of symptoms, checks, and owners. The table below is a starting structure; the thresholds belong to each service’s recovery targets.

Signal	Check first	Likely action	Owner decides
Lag growing for several intervals	Per-partition lag, backpressure location, sink latency	Fix the bottleneck; scale the constrained stage; throttle or shed if the product allows	On-call data engineer; product owner if shedding
Checkpoints failing or slow	Start delay versus alignment versus async duration; storage latency	Remove backpressure; consider unaligned checkpoints; check storage	On-call data engineer
Restart count rising	Exception in the job log; poison record; transaction expiry	Route bad records to dead-letter; raise transaction timeout above checkpoint plus restart	On-call data engineer
Watermark not advancing	Idle partitions; stalled source; future timestamp	Idleness setting; fix source; validate timestamps	Data engineer with source owner
Late-record drops rising	Disorder and delay histograms; new producer or region	Adjust bound or lateness; route late records for correction	Data engineer with consumers of the result
Planned upgrade	Savepoint, identifiers, state compatibility, sink behavior at stop	Savepoint, deploy, restore, verify; keep the savepoint	Change owner, with sign-off for output changes
Wrong results found	Range, cause, whether state is affected	Rewind in a new job to a separate output; reconcile; restate with notice	Data owner decides; consumers are notified

The tests from the previous article apply here too. A recovery drill that restores from a savepoint into the new version in a lower environment, reports the time taken, and reconciles the output is the only proof that a rollout plan works. The recovery time measured there, plus the catch-up time computed from the headroom, is the number that belongs next to the service’s recovery target.

Decisions to write down

• Lag definitions and alerts. Per-partition offset lag, age of the oldest unread record, event-time lag, and checkpoint health, each with a threshold tied to a stated recovery target.
• Headroom. The processing capacity above the arrival rate and the catch-up time it implies for the worst expected backlog.
• Replay procedure. Which sinks tolerate replay, which need version guards or a separate destination, and who may start a replay.
• Retention. At least as long as the longest outage plus catch-up, and as long as the state lifetime where rebuild is the fallback.
• Schema rules. Compatibility mode for input topics, evolvable state types chosen before the first deployment, versioned outputs, and a dead-letter path for unreadable records.
• State migration path. For each job, whether it relies on savepoints, the State Processor API, or rebuild from the log.
• Rollout procedure. Savepoint, identifiers, claim mode, side-by-side run, rollback by replay, and the versions upgraded one at a time.
• Ownership. Who detects, who decides, who executes, and who notifies consumers.

References

• Apache Flink documentation (stable, 2.3 line): Monitoring Back Pressure
• Apache Flink documentation: Checkpointing under backpressure
• Apache Flink documentation: Network memory tuning (buffer debloating)
• Apache Flink documentation: Metrics
• Apache Flink documentation: Kafka connector
• Apache Flink documentation: Savepoints
• Apache Flink documentation: Checkpoints vs. Savepoints
• Apache Flink documentation: Upgrading Applications and Flink Versions
• Apache Flink documentation: State Schema Evolution
• Apache Flink documentation: State Processor API
• Apache Flink documentation: Task Failure Recovery
• Apache Flink blog: A Deep-Dive into Flink’s Network Stack
• Apache Spark documentation: Structured Streaming, recovery semantics after changes in a query
• Apache Spark documentation: Structured Streaming and Kafka integration
• Apache Kafka 4.3 documentation: Basic Kafka operations (checking consumer position)
• Apache Kafka 4.3 documentation: Monitoring
• Apache Kafka 4.3 documentation: Consumer configs
• Apache Kafka 4.3 documentation: Topic configs