Skip to content

[Bug] Fix Flink Dedicated Streaming Compact Job stuck when checkpointed snapshot expired #9535

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhang-arvin wants to merge 1 commit into
base: master
Choose a base branch
from
+13 −0
Open

[Bug] Fix Flink Dedicated Streaming Compact Job stuck when checkpointed snapshot expired #9535

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,19 @@ public Long watermark() {

@Override
public void restore(@Nullable Long nextSnapshotId) {

@lilei1128 lilei1128 Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@lilei1128 lilei1128 Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

if (nextSnapshotId != null) {
Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
if (earliestSnapshotId != null && earliestSnapshotId > nextSnapshotId) {
LOG.warn(
"The restored snapshot with id {} has expired. "
+ "The earliest snapshot is {}. "
+ "Falling back to starting scanner.",
nextSnapshotId,
earliestSnapshotId);
this.nextSnapshotId = null;
return;
}
}
this.nextSnapshotId = nextSnapshotId;
}

Expand Down
Loading