[core][rollback] Skip non-existent snapshots in rollback

[FangYongs]

Purpose

Linked issue: close #3732

Skip non-existent snapshots in rollback operation

Tests

Added FileStoreTableTestBase#testRollbackToSnapshotSkipNonExistentSnapshot

API and Format

no

Documentation

no

[JingsongLi]

Hi @FangYongs , you can show the exception stack first, and we can decide how to solve this problem.

[FangYongs]

@JingsongLi Our version is 0.6 and the root cause is as follows,:
Caused by: java.io.FileNotFoundException: No such file or directory: hdfs://xxxx
at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:899) ~[flink-shaded-hadoop-2-uber-3.2.1-1.5.2.26-16.0-byted-20240514.084713-1.jar:3.2.1-1.5.2.26-16.0-byted-SNAPSHOT]
at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.open(HadoopFileSystem.java:131) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.open(HadoopFileSystem.java:37) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
at org.apache.paimon.flink.FlinkFileIO.newInputStream(FlinkFileIO.java:63) ~[?:?]
at org.apache.paimon.fs.FileIO.readFileUtf8(FileIO.java:194) ~[?:?]
at org.apache.paimon.Snapshot.fromPath(Snapshot.java:419) ~[?:?]
at org.apache.paimon.utils.SnapshotManager.snapshot(SnapshotManager.java:90) ~[?:?]
at org.apache.paimon.table.RollbackHelper.cleanSnapshotsDataFiles(RollbackHelper.java:106) ~[?:?]
at org.apache.paimon.table.RollbackHelper.cleanLargerThan(RollbackHelper.java:70) ~[?:?]
at org.apache.paimon.table.AbstractFileStoreTable.rollbackTo(AbstractFileStoreTable.java:396) ~[?:?]
at org.apache.paimon.flink.action.RollbackToAction.run(RollbackToAction.java:54) ~[?:?]
at org.apache.paimon.flink.action.FlinkActions.main(FlinkActions.java:41) ~[?:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_362]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_362]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_362]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_362]
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:392) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:235) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:301) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT]
... 12 more

[JingsongLi]

@JingsongLi Our version is 0.6 and the root cause is as follows,: Caused by: java.io.FileNotFoundException: No such file or directory: hdfs://xxxx at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:899) ~[flink-shaded-hadoop-2-uber-3.2.1-1.5.2.26-16.0-byted-20240514.084713-1.jar:3.2.1-1.5.2.26-16.0-byted-SNAPSHOT] at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.open(HadoopFileSystem.java:131) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.open(HadoopFileSystem.java:37) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] at org.apache.paimon.flink.FlinkFileIO.newInputStream(FlinkFileIO.java:63) ~[?:?] at org.apache.paimon.fs.FileIO.readFileUtf8(FileIO.java:194) ~[?:?] at org.apache.paimon.Snapshot.fromPath(Snapshot.java:419) ~[?:?] at org.apache.paimon.utils.SnapshotManager.snapshot(SnapshotManager.java:90) ~[?:?] at org.apache.paimon.table.RollbackHelper.cleanSnapshotsDataFiles(RollbackHelper.java:106) ~[?:?] at org.apache.paimon.table.RollbackHelper.cleanLargerThan(RollbackHelper.java:70) ~[?:?] at org.apache.paimon.table.AbstractFileStoreTable.rollbackTo(AbstractFileStoreTable.java:396) ~[?:?] at org.apache.paimon.flink.action.RollbackToAction.run(RollbackToAction.java:54) ~[?:?] at org.apache.paimon.flink.action.FlinkActions.main(FlinkActions.java:41) ~[?:?] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_362] at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_362] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_362] at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_362] at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:392) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:235) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:301) ~[flink-dist-1.16-byted-SNAPSHOT.jar:1.16-byted-SNAPSHOT] ... 12 more

Thanks, I see

[JingsongLi]

+1