[core] Retry snapshot existence checks after I/O failures

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -67,6 +67,9 @@ public class SnapshotManager implements Serializable {
 
     public static final int EARLIEST_SNAPSHOT_DEFAULT_RETRY_NUM = 300;
 
+    private static final int SNAPSHOT_EXISTS_MAX_ATTEMPTS = 3;
+    private static final long SNAPSHOT_EXISTS_RETRY_INTERVAL_MILLIS = 1_000L;
+
     private final FileIO fileIO;
     private final Path tablePath;
     private final String branch;
@@ -147,13 +150,44 @@ public Snapshot tryGetSnapshot(long snapshotId) throws FileNotFoundException {
 
     public boolean snapshotExists(long snapshotId) {
         Path path = snapshotPath(snapshotId);
-        try {
-            return fileIO.exists(path);
-        } catch (IOException e) {
-            throw new RuntimeException(
-                    "Failed to determine if snapshot #" + snapshotId + " exists in path " + path,
-                    e);
+        IOException failure = null;
+        for (int attempt = 1; attempt <= SNAPSHOT_EXISTS_MAX_ATTEMPTS; attempt++) {
+            try {
+                return fileIO.exists(path);
+            } catch (IOException e) {
+                failure = e;
+                if (attempt == SNAPSHOT_EXISTS_MAX_ATTEMPTS) {
+                    break;
+                }
+                LOG.warn(
+                        "Failed to check whether snapshot #{} exists at {} (attempt {}/{}). Retrying.",
+                        snapshotId,
+                        path,
+                        attempt,
+                        SNAPSHOT_EXISTS_MAX_ATTEMPTS,
+                        e);
+                try {
+                    Thread.sleep(SNAPSHOT_EXISTS_RETRY_INTERVAL_MILLIS);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(
+                            "Interrupted while checking whether snapshot #"
+                                    + snapshotId
+                                    + " exists at "
+                                    + path,
+                            ie);
+                }
+            }
         }
+        throw new RuntimeException(
+                "Failed to check whether snapshot #"
+                        + snapshotId
+                        + " exists at "
+                        + path
+                        + " after "
+                        + SNAPSHOT_EXISTS_MAX_ATTEMPTS
+                        + " attempts.",
+                failure);
     }
 
     public void deleteSnapshot(long snapshotId) {

paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java

@@ -95,6 +95,32 @@ public void testSnapshotsWithIdSkipsExpiredSnapshot() throws Exception {
                 .isFalse();
     }
 
+    @Test
+    public void testSnapshotExistsRetriesAfterIOException() throws IOException {
+        FileIO fileIO = Mockito.mock(FileIO.class);
+        Mockito.when(fileIO.exists(Mockito.any(Path.class)))
+                .thenThrow(new IOException("Temporary failure"))
+                .thenReturn(true);
+        SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString()));
+
+        assertThat(snapshotManager.snapshotExists(2)).isTrue();
+        Mockito.verify(fileIO, Mockito.times(2)).exists(Mockito.any(Path.class));
+    }
+
+    @Test
+    public void testSnapshotExistsFailsAfterMaxAttempts() throws IOException {
+        FileIO fileIO = Mockito.mock(FileIO.class);
+        Mockito.when(fileIO.exists(Mockito.any(Path.class)))
+                .thenThrow(new IOException("Persistent failure"));
+        SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString()));
+
+        assertThatThrownBy(() -> snapshotManager.snapshotExists(2))
+                .hasMessageContaining("Failed to check whether snapshot #2 exists")
+                .hasMessageContaining("after 3 attempts")
+                .hasRootCauseMessage("Persistent failure");
+        Mockito.verify(fileIO, Mockito.times(3)).exists(Mockito.any(Path.class));
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testEarliestSnapshot(boolean isRaceCondition) throws IOException {