[Bug] Fix Flink Dedicated Streaming Compact Job stuck when checkpointed snapshot expired - #9535
Conversation
|
Spark test failed. |
|
The proposed fix addresses the expired nextSnapshotId case. Please verify whether pending splits and sink writer state can also reference the expired snapshot. If they cannot, this source-only fix may be sufficient for the current bug scope. If they can, additional cleanup or writer reinitialization is required for end-to-end recovery. |
| @@ -315,6 +315,19 @@ public Long watermark() { | |||
|
|
|||
| @Override | |||
| public void restore(@Nullable Long nextSnapshotId) { | |||
There was a problem hiding this comment.
Similar possible situations need to be taken into consideration, for example:
A previous asynchronous compaction may expire and delete the snapshot files that the source is about to read.
Possible sequence:
The source selects snapshot N for reading
-> a previous asynchronous compaction commits a newer snapshot
-> snapshot N or its manifest files are deleted
-> the source continues reading snapshot N
-> the read fails with OutOfRangeException or FileNotFoundException
This is a read-time race and can happen even without a Flink failover.
There was a problem hiding this comment.
I think we can treat it as a source-and-sink recovery problem, follows:
detect expired snapshot(include normal read or checkpoint restore)
-> discard expired pending splits
-> build a baseline from the latest valid snapshot
-> notify all compaction writers
-> rebuild writer state
-> ignore in-flight splits covered by the new baseline
-> continue from the next snapshot
3 participants
Fixes #9533: When Flink Dedicated Streaming Compact Job restores from a checkpoint whose snapshot has been expired, the job gets stuck in a loop because nextPlan() keeps returning SnapshotNotExistPlan without incrementing nextSnapshotId.
Fix
In restore(), when nextSnapshotId is less than earliestSnapshotId (snapshot expired), set nextSnapshotId to null so plan() falls back to tryFirstPlan() which uses ContinuousCompactorStartingScanner to find the latest compact snapshot.