The offset that vanished: recover a dead Kafka stream without silently losing data

The error message is the whole diagnosis: the checkpoint recorded offset 48,210 for the partition, and the topic now offers something lower — or nothing at that position at all. Spark can't tell a recreated topic from aged-out retention, so it assumes the worst: data may have been lost. Failing loudly is the entire point of the default.

Three things cause it, and they all mean the same thing — the checkpoint and the topic describe different worlds: (1) the topic was deleted and recreated; (2) Kafka's retention aged the offsets out while the stream was stopped; (3) the topic was reconfigured — partitions added or moved — while the checkpoint pointed at the old layout. In every case, the offsets the stream wants no longer exist in Kafka.

import json

ckpt = "/Volumes/catalog/schema/vol/checkpoints/events_stream"

# The offsets/ directory holds one JSON file per committed batch:
# {"events": {"0": 48210, "1": 48177}}
offset_files = sorted(
    dbutils.fs.ls(ckpt + "/offsets/"), key=lambda f: int(f.name)
)
expected = json.loads(dbutils.fs.head(offset_files[-1].path))
print(json.dumps(expected, indent=2))

# Illustrative expected output:
# {"events": {"0": 48210, "1": 48177}}

Recommended learning: Databricks docs: Structured Streaming checkpoints

The failure

Illustrative scenario: A team streams click events from a Kafka topic into a Delta bronze table — append mode, hourly trigger, checkpoint on a Unity Catalog volume. Friday evening they stop the job for a runtime upgrade. The topic's retention is 48 hours — fine for a normal weekend, but the upgrade slips to Monday morning. The restarted query dies immediately with the IllegalStateException. The checkpoint's last committed offsets (illustrative: partition 0 → 48,210) sit below the topic's new earliest offset (illustrative: 51,003) — about 2,800 offsets of events are simply gone from Kafka. Restarting again changes nothing: the offsets don't exist to resume from.

1. Diagnose before you reset

Run the offsets-file read from the tip and establish two numbers: what the checkpoint expects, and the topic's current earliest offset. If the expected offsets are still within the topic's range, restart as-is — the query resumes from the checkpoint. If they sit below the earliest offset, the data is gone from Kafka: reset, don't retry.

2. Reset with a versioned checkpoint — never reuse or delete the old path

# Illustrative values: replace KAFKA, catalog, and paths with your own
KAFKA = "kafka-broker1:9092"
CKPT  = "/Volumes/catalog/schema/vol/checkpoints/events_stream"

recovered = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", KAFKA)
    .option("subscribe", "events")
    .option("startingOffsets", "latest")  # new query: only new data from here
    .load()
    .selectExpr("CAST(key AS STRING) AS event_key",
                "CAST(value AS STRING) AS payload",
                "topic", "partition", "offset",
                "timestamp AS kafka_ts"))

q = (recovered.writeStream.format("delta")
    .outputMode("append")
    .option("checkpointLocation", CKPT + "_v2")  # versioned: never reuse _v1
    .trigger(processingTime="1 hour")
    .toTable("catalog.bronze.events"))

Two things matter here. startingOffsets=latest on the new query means no replay — it picks up only records arriving from here on. And the _v2 path keeps the reset auditable: _v1 stays behind as the forensic record of what the old query committed. Per the docs: “When you delete the files in a checkpoint directory or change to a new checkpoint location, the next run of the query begins fresh.”

3. Backfill the gap from a replayable source — never from Kafka — then dedupe before writing

# Illustrative independent replay source. Replace the path, format, columns,
# and timestamps with your landing-zone contract and observed missing window.
RAW_REPLAY = "/Volumes/catalog/raw/landing/events"
MISSING_FROM = "2026-09-25T18:00:00Z"
MISSING_TO   = "2026-09-28T08:00:00Z"

gap = (spark.read.format("json")
    .load(RAW_REPLAY)
    .where(f"event_time >= timestamp '{MISSING_FROM}' "
           f"AND event_time < timestamp '{MISSING_TO}'")
    .select("event_key", "payload", "event_time"))

# Dedupe by event key before writing. Valid only when event_key is unique
# per event; duplicate keys would collapse distinct events.
(gap.dropDuplicates(["event_key"])
    .write.format("delta").mode("append")
    .saveAsTable("catalog.bronze.events_backfill"))

The expired offsets (illustrative: 48,210 and 48,177) are below Kafka's earliest available offset (illustrative: 51,003) and cannot be read from Kafka. This batch read uses an explicitly illustrative independent raw landing-zone source, filtered to the missing event-time window. Dedupe by event key before writing because the backfill can overlap the new stream's first micro-batch; the dropDuplicates is valid only when the event key is unique per event — duplicate keys would collapse distinct events. Validate the backfilled row count against the gap size (illustrative: ~2,800) before unioning it into the bronze table.

Streaming recovery you can't afford to get wrong? I run checkpoint-recovery reviews — the offsets audit, the retention math, and the backfill drill, on your pipeline. Get in touch through the contact in About.

Recommended learning: Data engineering workflows — pipelines, orchestration, and streaming practice