Your expired Kafka offsets are gone — stop trying to re-read them.

Expired means gone from Kafka

When a Structured Streaming job restarts from an existing checkpoint, it resumes from the source offsets committed there — not from fresh reader options. startingOffsets applies only when a new query starts. If the application was down longer than Kafka retention, or retention is shorter than a batch's processing time, the checkpoint can point below the broker's earliest available offset.

At that point the query fails with KAFKA_DATA_LOSS (SQLSTATE 22000). A read below the broker's earliest available offset cannot succeed because those records no longer exist on Kafka. Changing startingOffsets on the existing checkpoint does nothing. Setting failOnDataLoss=false skips missing data; it does not re-read it.

1. Start with a fail-loud bronze ingestion query

This baseline makes the two semantics explicit: startingOffsets="earliest" is honored only for a new query, while failOnDataLoss="true" is the default and stops the query when source data might be missing.

from pyspark.sql import functions as F

BOOTSTRAP_SERVERS = "kafka-1:9092,kafka-2:9092"
TOPIC = "events"
CHECKPOINT = "/Volumes/ops/streaming/checkpoints/events_bronze_v1"

kafka_events = (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
      .option("subscribe", TOPIC)
      .option("startingOffsets", "earliest")   # ONLY for a new query
      .option("failOnDataLoss", "true")       # default: fail loudly
      .load()
      .select(
          F.col("topic"), F.col("partition"), F.col("offset"),
          F.col("timestamp"), F.col("key").cast("string"),
          F.col("value").cast("string")
      )
)

query = (
    kafka_events.writeStream
      .format("delta")
      .outputMode("append")
      .option("checkpointLocation", CHECKPOINT)
      .toTable("ops.kafka_bronze")
)

2. GATE 1 — compare committed offsets with broker earliest offsets

Stop the query and inspect the newest numeric file under the checkpoint's offsets/ directory. dbutils.fs.head exposes the checkpoint record; do not modify it. Separately obtain the current earliest offset for every topic partition from your Kafka administration or monitoring tooling. A committed offset below its partition's earliest available offset proves that Kafka retention has created an unrecoverable broker-side gap.

# GATE 1 — READ ONLY. Never edit files in this directory.
offset_dir = f"{CHECKPOINT}/offsets/"
offset_files = [f for f in dbutils.fs.ls(offset_dir) if f.name.rstrip("/").isdigit()]
latest = max(offset_files, key=lambda f: int(f.name.rstrip("/")))

print(f"Latest checkpoint offset file: {latest.path}")
print(dbutils.fs.head(latest.path, 65536))

# Parse the Kafka offset JSON from that output according to your checkpoint format.
# Then compare, partition by partition:
#   checkpoint_committed_offset  vs.  broker_earliest_available_offset
# If committed < earliest, that Kafka range is gone.

3. Illustrative timeline — how the gap appears

ILLUSTRATIVE ERROR PATTERN — exact wording varies by runtime:

[KAFKA_DATA_LOSS.Could_Not_Read_Offset_Range] SQLSTATE: 22000
Reason: COULD_NOT_READ_OFFSET_RANGE
Could not read an offset range for topic partition events-2.
The checkpoint requested data below the broker's earliest available offset.

The class, SQLSTATE, and reason-code family are documented. The text above is an illustrative pattern, not a captured production stack trace.

4. Recovery A — new checkpoint, replay what Kafka still retains

Use a brand-new checkpoint path and startingOffsets="earliest". The query replays only the data that remains in Kafka. The expired range stays missed; record that gap before you proceed. Ensure the sink can tolerate replay or deduplicate by a stable event key.

CHECKPOINT_V2 = "/Volumes/ops/streaming/checkpoints/events_bronze_v2"

replay_retained = (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
      .option("subscribe", TOPIC)
      .option("startingOffsets", "earliest")  # honored: V2 is new
      .option("failOnDataLoss", "true")
      .load()
)

(replay_retained.writeStream
  .format("delta")
  .outputMode("append")
  .option("checkpointLocation", CHECKPOINT_V2)
  .toTable("ops.kafka_bronze"))

# The expired Kafka range is NOT recovered. Document it in ops.streaming_gaps.

5. Recovery B — deliberately advance past the loss

Use failOnDataLoss=false only as a one-time, declared gap advance when the business accepts the missing range. It tells Spark to skip unavailable data and continue. It must never become a standing configuration: leaving it off turns future retention failures into silent data loss.

# One controlled recovery run against the existing checkpoint.
advance_once = (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
      .option("subscribe", TOPIC)
      .option("failOnDataLoss", "false")  # skips; does NOT re-read
      .load()
)

(advance_once.writeStream
  .format("delta")
  .outputMode("append")
  .trigger(availableNow=True)
  .option("checkpointLocation", CHECKPOINT)
  .toTable("ops.kafka_bronze"))

# After the declared advance, restore failOnDataLoss="true".

6. Recovery C — backfill from an independent bronze Delta table

If an independently retained bronze table already contains the missing events, rebuild the affected downstream range from that table. This is the only path here that can fill data Kafka no longer retains. Record the incident and the exact recovery status in a durable operations table.

CREATE TABLE IF NOT EXISTS ops.streaming_gaps (
  stream_name STRING NOT NULL,
  topic STRING NOT NULL,
  partition INT NOT NULL,
  missing_from_offset BIGINT NOT NULL,
  missing_to_offset_exclusive BIGINT NOT NULL,
  detected_at TIMESTAMP NOT NULL,
  recovery_path STRING NOT NULL,
  recovery_status STRING NOT NULL,
  evidence STRING,
  recorded_by STRING
) USING DELTA;

-- ILLUSTRATIVE gap record only — replace every figure with Gate 1 evidence.
INSERT INTO ops.streaming_gaps VALUES (
  'events_bronze', 'events', 2, 120000, 126500,
  current_timestamp(), 'independent_bronze_backfill', 'backfilled',
  'checkpoint committed=120000; broker earliest=126500', current_user()
);

-- Backfill from the independent retained table using your idempotent sink logic.
-- ILLUSTRATIVE filter shape; table and offset figures are environment-specific.
MERGE INTO silver.events AS target
USING (
  SELECT * FROM archive.kafka_bronze
  WHERE topic = 'events' AND partition = 2
    AND offset >= 120000 AND offset < 126500
) AS source
ON target.event_id = source.event_id
WHEN NOT MATCHED THEN INSERT *;

7. GATE 2 — prove continuity before closing the incident

Check the per-partition maximum offsets actually present in bronze, verify the gap record exists, and reconcile the recovered or declared range with the Gate 1 evidence. Do not call the stream recovered because it is merely running.

-- GATE 2A: latest persisted offset per topic partition.
SELECT topic, partition, MAX(offset) AS max_persisted_offset
FROM ops.kafka_bronze
GROUP BY topic, partition
ORDER BY topic, partition;

-- GATE 2B: the incident must have a durable, reviewable gap record.
SELECT stream_name, topic, partition,
       missing_from_offset, missing_to_offset_exclusive,
       recovery_path, recovery_status, detected_at
FROM ops.streaming_gaps
WHERE stream_name = 'events_bronze'
ORDER BY detected_at DESC;

-- GATE 2C (outside SQL): compare these results with current broker earliest/latest
-- offsets and the checkpoint's next committed offsets, partition by partition.

Prevention — make the checkpoint-to-broker distance observable

• Size both log.retention.ms and size-based retention beyond the maximum plausible downtime and processing lag; the more restrictive retention limit can win.
• Alert when a checkpoint's committed offset approaches a partition's earliest available broker offset.
• Keep an independently retained, replayable bronze Delta table when the business cannot accept an unrecoverable Kafka-only gap.
• Keep failOnDataLoss=true for normal operation so the next loss is loud.

1. Name the failure. Capture KAFKA_DATA_LOSS, SQLSTATE 22000, and the documented reason code such as COULD_NOT_READ_OFFSET_RANGE.
2. Stop and run GATE 1. Read the newest checkpoint offsets/ file with dbutils.fs.head; compare committed with broker-earliest for every partition.
3. Choose exactly one recovery. Backfill from independent bronze, replay retained Kafka data under a new checkpoint, or deliberately advance with failOnDataLoss=false.
4. Declare the gap. Record topic, partition, missing offset range, evidence, recovery path, owner, and status in ops.streaming_gaps.
5. Never hand-edit checkpoint JSON. It is an undocumented mutation with no guaranteed outcome.
6. Run GATE 2. Verify per-partition persisted offsets, checkpoint progress, broker bounds, and the gap record before calling recovery complete.
7. Prevent the repeat. Extend time and size retention, alert on committed-versus-earliest distance, and keep replayable bronze.

Use this when

1. A Structured Streaming Kafka query fails with KAFKA_DATA_LOSS after downtime or prolonged processing lag.
2. GATE 1 shows a checkpointed offset below the broker's earliest available offset for one or more partitions.
3. You must choose transparently among retained-data replay, a declared skip, or backfill from an independent replayable bronze table.
4. Your incident process requires partition-level evidence and a durable record of every missing or backfilled range.

Avoid this when

1. The issue is transient connectivity, authentication, authorization, or broker availability; fix that cause without resetting checkpoint semantics.
2. All checkpointed offsets are still inside Kafka's retained range; restart from the existing checkpoint rather than manufacturing a reset.
3. You plan to leave failOnDataLoss=false enabled permanently; that converts future data loss into silent skipping.
4. You have no independent archive and cannot accept missing data; a new checkpoint cannot recreate records Kafka has deleted.