How do I fix Kafka "offsets out of range" in a stream?

Summary

• The error means the offsets your query remembers no longer exist in Kafka. The Spark documentation lists deleted topics, offsets out of range, and offsets removed after the retention period, and it says the offsets are not reset, so the application sees data loss.
• failOnDataLoss is true by default, so the query fails with a KAFKA_DATA_LOSS error. The option can raise a false alarm, so read the reason before you change anything.
• If the gap is acceptable, set failOnDataLoss to false. If it is not, restore the Kafka data or start a new run from a new checkpoint location with startingOffsets.
• auto.offset.reset cannot be set. startingOffsets applies only when a new query starts, and a resumed query picks up where it left off.
• Prevent a repeat by keeping topic retention longer than your longest outage. Track maxOffsetsBehindLatest in the query progress, because the Kafka source does not commit offsets to Kafka.

How do I fix Kafka "offsets out of range" in a stream?

A Structured Streaming query stores the Kafka offsets it has processed in its checkpoint. When those offsets no longer exist in Kafka, the Spark Kafka guide says the offsets will not be reset and the application will see data loss. With the default failOnDataLoss value of true, the query fails. First decide whether the missing data matters, then pick one of the two fixes below.

What causes the error?

The Kafka source reports a KAFKA_DATA_LOSS error: "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." The reason that follows tells you which case you have:

Reason in the error	What it says	Likely cause
KAFKA_DATA_LOSS.START_OFFSET_RESET	Starting offset for a partition was X but the consumer reset to Y	The offset the batch asked for is no longer in the log
KAFKA_DATA_LOSS.PARTITION_OFFSET_CHANGED	The offset changed from one value to another	The start offset is beyond what is available, or the topic-partition was deleted and re-created
KAFKA_DATA_LOSS.PARTITIONS_DELETED	Partitions have been deleted	The topic or its partitions were removed
KAFKA_DATA_LOSS.COULD_NOT_READ_OFFSET_RANGE	Could not read records in an offset range	The records were aged out, or the topic was deleted

The option's own description adds a warning: "This may be a false alarm."

How do I let the stream continue if the gap is acceptable?

Set failOnDataLoss to false. The query then keeps running, but the skipped records are lost for good:

df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:9092,host2:9092")
    .option("subscribe", "my_topic")
    .option("failOnDataLoss", "false")
    .load())

How do I restart cleanly if I cannot accept the gap?

Spark cannot bring back records that Kafka already deleted, so restore them in Kafka if you can. Otherwise start a new run from a new checkpoint location and choose where it begins. The next run of a query begins fresh when it gets a new checkpoint location, and startingOffsets applies only to a new query:

df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:9092,host2:9092")
    .option("subscribe", "my_topic")
    .option("startingOffsets", "earliest")
    .load())

query = (df.writeStream
    .format("parquet")
    .option("checkpointLocation", "/path/to/new/checkpoint")
    .option("path", "/path/to/output")
    .start())

For a new streaming query the default is latest, which skips everything produced while the query was down. Reading from earliest replays whatever Kafka still retains, so use an idempotent sink or expect duplicates. Keep the old checkpoint until the new run is healthy.

How do I prevent it?

• Keep retention longer than your longest outage. The topic setting retention.ms defaults to seven days and, in the Kafka documentation's words, represents an SLA on how soon consumers must read their data. The broker setting log.retention.hours defaults to 168 hours.
• Watch the lag inside the query. The Kafka source reports minOffsetsBehindLatest, maxOffsetsBehindLatest and avgOffsetsBehindLatest in the source metrics of query.lastProgress. Kafka consumer-group tools will not reflect the query's progress, because the source creates a unique group id per query and does not commit offsets.
• Keep up with the topic. If throughput cannot catch up with Kafka's retention, the guide warns that a batch's input rows may shrink until they reach zero.

To change a topic's retention, use kafka-configs.sh:

bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my_topic_name --alter --add-config retention.ms=<milliseconds>

How does Databricks approach this?

The Databricks Kafka FAQ describes KAFKA_DATA_LOSS_ERROR as offsets in the checkpoint that are no longer in Kafka, for example after the stream was paused longer than the retention period. If data loss is acceptable, it says to set failOnDataLoss to false to continue from the earliest available offset. If not, reset the checkpoint and reprocess from the earliest offsets, or restore the missing data. Each query needs its own checkpoint location, and changing it makes the next run begin fresh. For a Lakeflow pipeline, the checkpoint recovery page covers a full refresh, backup and backfill, or a selective checkpoint reset.

FAQs

Can I use auto.offset.reset to fix this?

No. The Kafka source or sink throws an exception if you set it. Use startingOffsets, which applies only when a new query starts. A resumed query always picks up where it left off.

Can I change startingOffsets on an existing query?

Not usefully. It applies only to a new streaming query, so start one with a new checkpoint location.