(paimon) branch master updated: [core] Fix latest snapshot race with concurrent expiration

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new feabf2c6d7 [core] Fix latest snapshot race with concurrent expiration
feabf2c6d7 is described below

commit feabf2c6d74ca3e1ae30e0ee8776d05ad33e1b98
Author: JingsongLi <jingsongl...@gmail.com>
AuthorDate: Thu Aug 27 00:29:33 2026 +0800

    [core] Fix latest snapshot race with concurrent expiration
---
 .../org/apache/paimon/utils/SnapshotManager.java   | 15 ++++++-
 .../apache/paimon/utils/SnapshotManagerTest.java   | 51 ++++++++++++++++++++++
 2 files changed, 65 insertions(+), 1 deletion(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java 
b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index 75be8fe921..8cd1599978 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -185,7 +185,20 @@ public class SnapshotManager implements Serializable {
 
     public @Nullable Snapshot latestSnapshotFromFileSystem() {
         Long snapshotId = latestSnapshotIdFromFileSystem();
-        return snapshotId == null ? null : snapshot(snapshotId);
+        while (snapshotId != null) {
+            try {
+                return tryGetSnapshot(snapshotId);
+            } catch (FileNotFoundException e) {
+                Long newSnapshotId = latestSnapshotIdFromFileSystem();
+                if (snapshotId.equals(newSnapshotId)) {
+                    // Retry once to preserve the existing exception when the 
latest snapshot is
+                    // genuinely missing instead of being concurrently expired.
+                    return snapshot(snapshotId);
+                }
+                snapshotId = newSnapshotId;
+            }
+        }
+        return null;
     }
 
     public @Nullable Long latestSnapshotId() {
diff --git 
a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java 
b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
index 5adbf4129f..bf1a005878 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -111,6 +111,29 @@ public class SnapshotManagerTest {
         
assertThat(snapshotManager.earliestSnapshot().id()).isEqualTo(isRaceCondition ? 
1 : 0);
     }
 
+    @Test
+    public void testLatestSnapshotWithConcurrentExpiration() throws 
IOException {
+        long millis = 1684726826L;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                new LatestSnapshotRaceManager(localFileIO, new 
Path(tempDir.toString()), millis);
+        Snapshot snapshot = createSnapshotWithMillis(0, millis);
+        localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(0), 
snapshot.toJson());
+
+        
assertThat(snapshotManager.latestSnapshotFromFileSystem().id()).isEqualTo(1);
+    }
+
+    @Test
+    public void testLatestSnapshotStillFailsWhenNoNewerSnapshotExists() {
+        SnapshotManager snapshotManager =
+                Mockito.spy(newSnapshotManager(LocalFileIO.create(), new 
Path(tempDir.toString())));
+        
Mockito.doReturn(1L).when(snapshotManager).latestSnapshotIdFromFileSystem();
+
+        assertThatThrownBy(snapshotManager::latestSnapshotFromFileSystem)
+                .hasMessageContaining("Snapshot file")
+                .hasMessageContaining("does not exist");
+    }
+
     @Test
     public void testRepairEarliestSnapshot() throws IOException {
         FileIO fileIO = LocalFileIO.create();
@@ -904,4 +927,32 @@ public class SnapshotManagerTest {
             return snapshotId;
         }
     }
+
+    /** Simulates a commit and expiration between finding and reading the 
latest snapshot. */
+    private static class LatestSnapshotRaceManager extends SnapshotManager {
+        private final long millis;
+        private boolean expireLatestSnapshot = true;
+
+        private LatestSnapshotRaceManager(FileIO fileIO, Path tablePath, long 
millis) {
+            super(fileIO, tablePath, DEFAULT_MAIN_BRANCH, null, null);
+            this.millis = millis;
+        }
+
+        @Override
+        public @Nullable Long latestSnapshotIdFromFileSystem() {
+            Long snapshotId = super.latestSnapshotIdFromFileSystem();
+            if (snapshotId != null && expireLatestSnapshot) {
+                Snapshot nextSnapshot = createSnapshotWithMillis(snapshotId + 
1, millis + 1000);
+                try {
+                    fileIO().tryToWriteAtomic(
+                                    snapshotPath(nextSnapshot.id()), 
nextSnapshot.toJson());
+                    fileIO().delete(snapshotPath(snapshotId), true);
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+                expireLatestSnapshot = false;
+            }
+            return snapshotId;
+        }
+    }
 }