Spark Kafka stream fails with failOnDataLoss: offsets out of range, aged-out data and recreated topics

#Symptom

A streaming query that reads from Kafka stops, and the driver log shows one of these messages (exact wording varies by Spark version):

• Some data may have been lost because they are not available in Kafka any more; either the data was aged out by Kafka or the topic may have been deleted before all the data in the topic was processed. If you don't want your streaming query to fail on such cases, set the source option "failOnDataLoss" to "false".
• Partition orders-3's offset was changed from 81234 to 0, some data may have been missed.
• Cannot fetch offset 81234 (GroupId: spark-kafka-source-..., TopicPartition: orders-3).
• Set(orders-3) are gone. Some data may have been missed.

Restarting the job gives the same error. The checkpoint still holds the old offsets, and every restart resumes from them.

#What the error means

Structured Streaming keeps its Kafka position in the checkpoint, not in Kafka consumer groups. On each micro-batch it compares the offsets it plans to read with what the brokers still have. If the next offset it needs is older than the partition's earliest offset, or a partition's offsets went backwards, Spark can't prove nothing was skipped, so by default (failOnDataLoss=true) it stops the query.

Warning

Setting failOnDataLoss to false and restarting makes the error go away. It does not bring the data back. Do it only after you know which offset range is gone and the owners of the data have accepted the gap or agreed to backfill it.

#Checks

#1. Read the offsets the checkpoint expects

The offset log is plain text. The last line of the newest file in offsets/ is a JSON map of topic → partition → next offset.

CKPT=/mnt/checkpoints/orders_bronze      # or abfss://..., s3://...
LATEST=$(ls "$CKPT/offsets" | grep -E '^[0-9]+$' | sort -n | tail -1)
tail -1 "$CKPT/offsets/$LATEST"
# {"orders":{"0":80211,"1":79987,"2":80544,"3":81234}}

On Databricks, read it from a notebook:

latest = max(int(f.name) for f in dbutils.fs.ls(f"{ckpt}/offsets") if f.name.isdigit())
print(dbutils.fs.head(f"{ckpt}/offsets/{latest}").splitlines()[-1])

#2. Compare with what the brokers still have

kafka-get-offsets.sh ships with Kafka 3.0 and later. --time -2 returns the earliest offset per partition and --time -1 the latest.

kafka-get-offsets.sh --bootstrap-server broker1:9092 --command-config client.properties \
  --topic orders --time -2      # earliest
kafka-get-offsets.sh --bootstrap-server broker1:9092 --command-config client.properties \
  --topic orders --time -1      # latest

On older clients, use kafka-run-class.sh kafka.tools.GetOffsetShell with the same options (the class name and flags vary by version; check --help).

How to read the result:

• Checkpoint offset < earliest: retention removed the data. The gap is earliest - checkpoint messages per partition.
• Checkpoint offset > latest: the topic was recreated or you're connected to a different cluster. Everything written to the new topic since it was recreated has not been read yet.
• Partition missing: the partition count went down.

#3. Confirm topic retention and history

kafka-configs.sh --bootstrap-server broker1:9092 --command-config client.properties \
  --entity-type topics --entity-name orders --describe --all | grep -E 'retention|cleanup.policy'
kafka-topics.sh --bootstrap-server broker1:9092 --command-config client.properties \
  --describe --topic orders

Look at the Spark streaming metrics too: the last successful batch time in the Spark UI Structured Streaming tab (or query.lastProgress). If that is older than retention.ms, aged-out data is your answer.

#Fixes

Pick one, depending on what is acceptable for the data.

#Option A: accept the gap and continue from the same checkpoint

Use this when the data is gone for good and the owners accept the gap. The checkpoint keeps the state store and sink progress, so this is the least disruptive option.

df = (spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9092")
      .option("subscribe", "orders")
      .option("failOnDataLoss", "false")      # temporary
      .option("maxOffsetsPerTrigger", 500000)  # don't swallow the whole backlog in one batch
      .load())

After restart, Spark logs a warning instead of failing and moves to the earliest offset still available. Once it has caught up, set failOnDataLoss back to true and redeploy, so the next incident fails loudly.

#Option B: new checkpoint with explicit starting offsets

Use this after a topic recreate, or when you want to choose exactly where to restart. startingOffsets is ignored when a checkpoint exists. It only applies to a new query, so it needs a new checkpoint path.

starting = '{"orders":{"0":0,"1":0,"2":0,"3":0}}'   # -2 = earliest, -1 = latest per partition

df = (spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9092")
      .option("subscribe", "orders")
      .option("startingOffsets", starting)
      .load())

(df.writeStream
   .option("checkpointLocation", "/mnt/checkpoints/orders_bronze_v2")  # new path
   .toTable("bronze.orders"))

Caution

A new checkpoint loses the streaming state (aggregations, dropDuplicates, stream-stream joins) and the sink's exactly-once bookkeeping. If the topic still holds messages that were already written, replaying them creates duplicates unless the sink is idempotent (MERGE on a business key, or Delta txnAppId/txnVersion in foreachBatch). Keep the old checkpoint, renamed rather than deleted, until the new stream has been checked.

To restart from a point in time instead of an offset, Spark 3.2+ supports startingOffsetsByTimestamp (a JSON map of partition → epoch milliseconds) and Spark 3.3+ adds startingTimestamp. Check the Kafka integration guide for your Spark version before you use them.

#Option C: backfill the gap from another source

If the lost range matters (orders, payments), restart with option A or B and backfill the missing window from the producer's database, a CDC snapshot or an archive (for example, a Kafka Connect S3/ADLS sink). Load it with a batch read into the same table, and dedupe on the business key.

#Verify

q = spark.streams.get(query_id)          # or query handle from writeStream.start()
print(q.status)
print(q.lastProgress["sources"][0]["endOffset"])

• endOffset moves forward on every batch.
• inputRowsPerSecond > processedRowsPerSecond only while you are catching up.
• Row counts per hour in the sink match the producer for the window around the incident.

#Prevent it next time

• Retention longer than your worst outage. If the stream can be down over a weekend, 24-hour retention will lose data. Set retention.ms on critical topics to at least 3–7 days, or turn on tiered storage.
• Alert on stream lag, not just failures. Alert when the last successful batch is older than a threshold well below retention.ms. Watch the avgOffsetsBehindLatest / maxOffsetsBehindLatest source metrics on Databricks (names vary by runtime; check lastProgress).
• Protect topics from recreation. Restrict DeleteTopics with ACLs, and make IaC fail on a topic diff instead of recreating the topic. If a topic has to be recreated, the runbook should include "start the consumers on a new checkpoint".
• Keep failOnDataLoss=true in production. A silent gap is worse than a failed job.
• Use idempotent sinks, so a replay from an earlier offset is always safe.
• Version checkpoint paths (..._v1, ..._v2) so a reset never overwrites the old checkpoint.