Checkpointing and Recovering Stream State After a Crash #

A stateful stream processor that cannot recover its state exactly is a batch job with extra latency. The window that was half-built when the process died holds twenty minutes of readings, and if recovery approximates it, the published hourly mean is wrong in a way nothing downstream will ever detect. This guide covers the mechanism that makes recovery exact — a changelog plus periodic checkpoints — the ordering rule that keeps state and offsets consistent, and how to measure whether your recovery time is acceptable. It is the durability half of stateful IoT stream processing patterns.

Changelog First, Checkpoint Second #

The two mechanisms answer different questions and neither replaces the other.

The changelog is a compacted Kafka topic to which every state update is written before it is considered applied. Because it is compacted, it retains at least the latest value for every key forever, so state can be rebuilt from scratch by replaying it. That is what makes recovery exact: the rebuilt state is the same state, entry for entry.

The checkpoint is a periodic snapshot of the local store plus the changelog offset it corresponds to. It contributes nothing to correctness; it exists so recovery replays thirty seconds of changelog instead of three months. A processor with checkpoints and no changelog cannot recover the updates made since the last snapshot, which is exactly the data an incident destroys.

The consistency rule binding them is simple and unforgiving: the changelog write happens before the state update, and the offset commit happens after both. Any other order produces a recovery that is either missing updates or replaying them onto state that already contains them — the second of which is harmless only if every update is idempotent, which for a running sum it is not.

Production-Ready Implementation #

# python 3.11 · confluent-kafka==2.4.0
from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass


@dataclass
class WindowAgg:
    """A windowed aggregate that is fully described by its own fields.

    Every field is order-independent — count, sum, min, max — so replaying the
    changelog in log order rebuilds an identical aggregate regardless of the
    order the readings originally arrived in.
    """

    count: int = 0
    total: float = 0.0
    minimum: float = float("inf")
    maximum: float = float("-inf")

    def add(self, value: float) -> None:
        self.count += 1
        self.total += value
        self.minimum = min(self.minimum, value)
        self.maximum = max(self.maximum, value)

    @property
    def mean(self) -> float:
        return self.total / self.count if self.count else float("nan")


class CheckpointedStore:
    """Local state with a changelog for correctness and snapshots for speed."""

    def __init__(self, producer, changelog_topic: str, snapshot_path, *,
                 checkpoint_every_s: float = 30.0) -> None:
        self.producer = producer
        self.changelog_topic = changelog_topic
        self.snapshot_path = snapshot_path
        self.checkpoint_every_s = checkpoint_every_s
        self.state: dict[str, WindowAgg] = {}
        self._last_checkpoint = time.monotonic()

    def update(self, key: str, value: float) -> None:
        """Changelog first, then local state. Never the other way round."""
        agg = self.state.get(key) or WindowAgg()
        agg.add(value)
        self.producer.produce(
            self.changelog_topic,
            key=key.encode(),
            value=json.dumps(asdict(agg)).encode(),
        )
        self.state[key] = agg              # local update only after the durable write

    def maybe_checkpoint(self, changelog_offset: int) -> bool:
        """Snapshot on a timer. The offset is what makes the snapshot resumable."""
        if time.monotonic() - self._last_checkpoint < self.checkpoint_every_s:
            return False
        self.producer.flush(10)            # every changelog record durable before the snapshot
        payload = {
            "offset": changelog_offset,
            "state": {k: asdict(v) for k, v in self.state.items()},
        }
        tmp = self.snapshot_path.with_suffix(".tmp")
        tmp.write_text(json.dumps(payload))
        tmp.replace(self.snapshot_path)    # atomic: a torn snapshot is never visible
        self._last_checkpoint = time.monotonic()
        return True

Two details make this survive a crash rather than merely describing one. producer.flush() before writing the snapshot guarantees that everything the snapshot claims is in the changelog actually is. And writing to a temporary file then renaming makes the snapshot atomic — a process killed mid-write leaves the previous snapshot intact, where a partial in-place write leaves a file that parses as JSON right up to the point it does not.

Recovery reverses the order:

def recover(store: CheckpointedStore, consumer, changelog_topic: str) -> int:
    """Load the snapshot, then replay the changelog from its offset. Returns records replayed."""
    start_offset = 0
    if store.snapshot_path.exists():
        payload = json.loads(store.snapshot_path.read_text())
        store.state = {k: WindowAgg(**v) for k, v in payload["state"].items()}
        start_offset = payload["offset"]

    consumer.assign([TopicPartition(changelog_topic, 0, start_offset)])
    replayed = 0
    while True:
        msg = consumer.poll(timeout=2.0)
        if msg is None:
            break                                   # caught up
        store.state[msg.key().decode()] = WindowAgg(**json.loads(msg.value()))
        replayed += 1
    return replayed

Note that replay overwrites rather than accumulating: each changelog record holds the full aggregate, not a delta. That choice costs a few more bytes per record and removes an entire class of recovery bug, because applying a full value twice is idempotent and applying a delta twice is not.

Parameter Tuning Guide #

The bottom row changes the shape of the problem. A standby replica consumes the changelog continuously on another node, so a failover promotes an already-warm store instead of rebuilding one. It costs a second copy of the state and turns a multi-minute recovery into a few seconds — the right trade for any pipeline where a stall is an incident, and the same reasoning as the rebalance costs discussion.

Verification and Testing #

The only convincing test kills the process at an arbitrary point and compares the recovered state against the state a clean run produced.

# python 3.11 · pytest==8.2.0
def test_recovery_reproduces_state_exactly(tmp_path, changelog, readings):
    """Crash at an arbitrary point; recovered state must equal the uninterrupted result."""
    clean = CheckpointedStore(producer(), changelog, tmp_path / "clean.json")
    for r in readings:
        clean.update(r["device_id"], r["value"])

    crashed = CheckpointedStore(producer(), changelog, tmp_path / "crashed.json")
    for i, r in enumerate(readings):
        crashed.update(r["device_id"], r["value"])
        crashed.maybe_checkpoint(changelog_offset=i)
        if i == len(readings) // 3:
            break                                    # simulate SIGKILL here

    recovered = CheckpointedStore(producer(), changelog, tmp_path / "crashed.json")
    recover(recovered, consumer(), changelog)
    for r in readings[len(readings) // 3 + 1:]:      # the rest is redelivered
        recovered.update(r["device_id"], r["value"])

    assert recovered.state == clean.state


def test_snapshot_write_is_atomic(tmp_path):
    """A truncated snapshot must never replace a good one."""
    store = CheckpointedStore(producer(), "cl", tmp_path / "s.json")
    store.update("s1", 20.0)
    store.maybe_checkpoint(changelog_offset=1)
    good = (tmp_path / "s.json").read_text()

    with mock.patch.object(pathlib.Path, "replace", side_effect=OSError("disk full")):
        with pytest.raises(OSError):
            store._last_checkpoint = 0
            store.maybe_checkpoint(changelog_offset=2)

    assert (tmp_path / "s.json").read_text() == good

Run the first test with the crash point swept across the input — a simple loop over a dozen positions — rather than at one fixed index. Recovery bugs cluster around specific moments, especially the instant just after a checkpoint.

Gotchas #

Updating local state before writing the changelog. The process then crashes holding state that no durable log records, and recovery quietly loses it. Changelog first, always.

A non-compacted changelog topic. With time-based retention, the oldest state updates are deleted and recovery rebuilds an incomplete store — silently, because there is no error to raise. Set cleanup.policy=compact.

Non-deterministic aggregates. first_value, last_value and anything reading datetime.now() produce different results on replay. If you need last-value semantics, make it deterministic by keying on the event time rather than on arrival order.

Snapshotting without flushing the producer. The snapshot records a changelog offset for records still sitting in the producer’s buffer. If the process dies before they are sent, recovery starts from an offset that never existed and skips them.

FAQ #

Related #

• Stateful IoT Stream Processing Patterns — the stateful processing stage this recovery protects
• Sizing RocksDB State Stores for Sensor Aggregates — the state size that decides how long recovery takes
• Exactly-Once Sensor Ingestion with Kafka Transactions — the transactional guarantee that makes state and offsets recover together