[core] Treat a snapshot rename that throws but succeeded as committed

paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java

@@ -63,7 +63,17 @@ public boolean commit(
 
         Callable<Boolean> callable =
                 () -> {
-                    boolean committed = fileIO.tryToWriteAtomic(newSnapshotPath, snapshot.toJson());
+                    boolean committed;
+                    try {
+                        committed = fileIO.tryToWriteAtomic(newSnapshotPath, snapshot.toJson());
+                    } catch (IOException e) {
+                        // the rename may have succeeded even if the file system reports an
+                        // error, check the snapshot file below
+                        if (!fileIO.exists(newSnapshotPath)) {
+                            throw e;
+                        }
+                        committed = false;
+                    }
                     if (!committed) {
                         if (!fileIO.exists(newSnapshotPath)) {
                             throw new IOException(

paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java

@@ -101,6 +101,51 @@ public void testCommitTargetSnapshotMissing(@TempDir java.nio.file.Path tmp) thr
         assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse();
     }
 
+    /**
+     * Simulate the object storage scenario where rename actually succeeds but throws, for example
+     * when the response of a successful copy cannot be parsed. The commit should succeed and write
+     * the LATEST hint.
+     */
+    @Test
+    public void testCommitSucceedsAndRenameThrows(@TempDir java.nio.file.Path tmp)
+            throws Exception {
+        FileIO fileIO = new ThrowingRenameLocalFileIO(true);
+        Path tablePath = new Path(tmp.toUri());
+        SnapshotManager snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null);
+
+        RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
+
+        Snapshot snapshot = createSnapshot(3L);
+
+        boolean committed = commit.commit(null, snapshot, "main", Collections.emptyList());
+
+        assertThat(committed).isTrue();
+        Path latestHint = new Path(snapshotManager.snapshotDirectory(), HintFileUtils.LATEST);
+        assertThat(fileIO.readOverwrittenFileUtf8(latestHint).orElse(null))
+                .isEqualTo(String.valueOf(snapshot.id()));
+    }
+
+    /** When rename throws and the target snapshot file is not created, the error is rethrown. */
+    @Test
+    public void testCommitRenameThrowsAndTargetMissing(@TempDir java.nio.file.Path tmp)
+            throws Exception {
+        FileIO fileIO = new ThrowingRenameLocalFileIO(false);
+        Path tablePath = new Path(tmp.toUri());
+        SnapshotManager snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null);
+
+        RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
+
+        Snapshot snapshot = createSnapshot(4L);
+
+        IOException ex =
+                assertThrows(
+                        IOException.class,
+                        () -> commit.commit(null, snapshot, "main", Collections.emptyList()));
+
+        assertThat(ex).hasMessageContaining(ThrowingRenameLocalFileIO.MESSAGE);
+        assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse();
+    }
+
     private static Snapshot createSnapshot(long id) throws IOException {
         long schemaId = 1L;
         String baseManifestList = "manifest-list-base";
@@ -171,4 +216,29 @@ public boolean rename(Path src, Path dst) throws IOException {
             }
         }
     }
+
+    /**
+     * A FileIO based on LocalFileIO that always throws from rename().
+     *
+     * <p>When {@code actuallyMove} is true, it performs the move then throws; otherwise it throws
+     * without moving.
+     */
+    private static class ThrowingRenameLocalFileIO extends LocalFileIO {
+
+        private static final String MESSAGE = "Failed to parse the rename response";
+
+        private final boolean actuallyMove;
+
+        ThrowingRenameLocalFileIO(boolean actuallyMove) {
+            this.actuallyMove = actuallyMove;
+        }
+
+        @Override
+        public boolean rename(Path src, Path dst) throws IOException {
+            if (actuallyMove) {
+                super.rename(src, dst);
+            }
+            throw new IOException(MESSAGE);
+        }
+    }
 }