Hi @vijsharm,
This is a common scenario when working with Structured Streaming checkpoints and Kafka. Here is what is happening and how to resolve it.
WHAT HAPPENED
When you switched to a new checkpoint path, the streaming query had no prior offset information, so startingOffsets = "earliest" kicked in and reprocessed everything from the beginning of the Kafka topic. That part is expected.
When you switched back to the original checkpoint path, the query resumed from the last committed offset stored in that original checkpoint. The key detail: startingOffsets is only used when a query starts for the very first time with no existing checkpoint. On all subsequent runs, the query always resumes from the offsets recorded in the checkpoint, ignoring startingOffsets entirely.
So the original checkpoint still has the old offsets from before you switched away. If new records arrived in Kafka while you were using the alternate checkpoint, the original checkpoint does not know about them unless those offsets are still within range.
WHY NEW RECORDS ARE NOT APPEARING
There are a few likely causes:
1. Kafka retention expired the messages. If your Kafka topic has a retention period (e.g., 7 days) and the original checkpoint was stale for longer than that, the offsets it recorded may now point to data that Kafka has already purged. Since you have failOnDataLoss = "false", Spark silently skips over the gap rather than throwing an error, and it may appear that no new data is coming in.
2. The checkpoint committed offsets that already cover the new data. If the alternate checkpoint processed data past the point the original checkpoint left off, and then you switched back, the original checkpoint's offsets are behind. The query should pick up from those older offsets, but if Kafka has deleted those segments, the stream jumps ahead and may land at the current end with nothing new to process.
3. Offset metadata mismatch. If the Kafka topic was reconfigured (partitions added, topic recreated, etc.) while the checkpoint was in use elsewhere, the stored partition/offset mapping may no longer align with the current topic state.
HOW TO FIX THIS
Option A: Start fresh with a new checkpoint (recommended)
Use a brand-new checkpoint location and set startingOffsets to "latest" so you only pick up new records going forward. This avoids reprocessing and gives you a clean slate:
checkpoint_new = get_checkpoint_path(source_kafka_topic + "/" + process_name + "_v2")
query = (
df_stream.writeStream.format("delta")
.outputMode("append")
.trigger(availableNow=True)
.option("checkpointLocation", checkpoint_new)
.option("startingOffsets", "latest")
.option("mergeSchema", "true")
.option("schemaCheck", "false")
.toTable(raw_table_name)
)
query.awaitTermination()
If you need to backfill the gap between when the original checkpoint stopped and now, you can run a separate batch read from Kafka using startingOffsets and endingOffsets with specific timestamps or offset values to fill in the missing window.
Option B: Start fresh with "earliest" if you can tolerate reprocessing
If your downstream pipeline is idempotent and can handle duplicate records, you can use a new checkpoint with startingOffsets = "earliest". This reprocesses everything but guarantees nothing is missed. To avoid duplicates in your Delta table, consider adding a MERGE or deduplication step downstream.
Option C: Inspect and adjust the checkpoint offsets (advanced)
You can examine the checkpoint's offsets directory to see what offsets were last committed:
%fs ls <your_checkpoint_path>/offsets/
Look at the latest offset file to see the recorded partition offsets. Compare these with the current Kafka topic offsets (earliest and latest) to understand the gap. If the committed offsets are still within Kafka's retention window, the query should resume correctly on the next trigger.
BEST PRACTICES FOR CHECKPOINT MANAGEMENT
- Never share a checkpoint location between different streaming queries.
- Never reuse a checkpoint after significant time gaps if Kafka retention might have expired the data.
- When you need to reset a stream, always create a new checkpoint path rather than deleting or reusing an old one.
- Use failOnDataLoss = "true" (the default) during development so you get clear errors when offsets fall out of Kafka's retention window, rather than silently skipping data.
- Consider versioning your checkpoint paths (e.g., appending _v1, _v2) so you can always trace which checkpoint belongs to which configuration.
DOCUMENTATION REFERENCES
Structured Streaming checkpoints:
https://docs.databricks.com/aws/en/structured-streaming/checkpoints.html
Kafka Structured Streaming configuration (startingOffsets behavior):
https://docs.databricks.com/aws/en/structured-streaming/kafka.html
* This reply used an agent system I built to research and draft this response based on the wide set of documentation I have available and previous memory. I personally review the draft for any obvious issues and for monitoring system reliability and update it when I detect any drift, but there is still a small chance that something is inaccurate, especially if you are experimenting with brand new features.
If this answer resolves your question, could you mark it as "Accept as Solution"? That helps other users quickly find the correct fix.
