Hi All,
We've been fighting checkpoint corruption on stateful Structured Streaming with S3 checkpoints for a few months now (Spark 3.5.x). The error looks like this: Error reading delta file .../state/0/506/1.delta ... does not exist 1.delta, on a query sitting at version xxx . The provider can't find a snapshot, falls back to replaying from version zero, and asks for the first delta, which expired weeks ago by S3 lifecyle policy. Snapshots only get written by the background maintenance task. If that stops keeping up, you're now relying on every single delta since version 1 still being there. Then the S3 bucket's lifecycle rule expires some of those deltas, and you encounter the error above upon the next restart. As for why maintenance falls behind: the trigger finishes too fast for a small streaming profile. The maintenance task is low priority, it doesn't hold the app up, so it just gets killed before it finishes. We see it mostly in small regions where shuffle partitions are way oversized for the cores available. So Right-sizing spark.sql.shuffle.partitions per query is probably a mitigation, we are not sure. Problem is we've got a lot of queries with very different state sizes and we currently just derive one default from cluster size and apply it everywhere. Curious how others deal with that, if you do it per query or something smarter. Anyway, the actual question. We're planning to drop S3 expiry from the checkpoint prefixes entirely, and handle size with a cleanup that checks whether a file is still referenced instead of how old it is. Reasoning being that any rule which can delete a file the live checkpoint still needs is just a correctness bug, and picking a bigger number doesn't fix it since there's no upper bound on how old a referenced file can be. Is that what everyone else does? Or is anyone running bounded retention on checkpoint prefixes without getting bitten, and if so what makes it safe? Thanks, Dibyendu
