[core] Improve error message for concurrent snapshot expiration

paimon-core/src/main/java/org/apache/paimon/Snapshot.java

@@ -29,8 +29,12 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
@@ -61,6 +65,7 @@
 @Public
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class Snapshot {
+    private static final Logger LOG = LoggerFactory.getLogger(Snapshot.class);
 
     public static final long FIRST_SNAPSHOT_ID = 1;
 
@@ -357,6 +362,16 @@ public static Snapshot fromJson(String json) {
     public static Snapshot fromPath(FileIO fileIO, Path path) {
         try {
             return Snapshot.fromJson(fileIO.readFileUtf8(path));
+        } catch (FileNotFoundException e) {
+            String errorMessage =
+                    String.format(
+                            "Snapshot file %s does not exist. "
+                                    + "It might have been expired by other jobs operating on this table. "
+                                    + "In this case, you can avoid concurrent modification issues by configuring "
+                                    + "write-only = true and use a dedicated compaction job, or configuring "
+                                    + "different expiration thresholds for different jobs.",
+                            path);
+            throw new RuntimeException(errorMessage, e);
         } catch (IOException e) {
             throw new RuntimeException("Fails to read snapshot from path " + path, e);
         }

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

@@ -369,7 +369,9 @@ public void testTraversalSnapshotsFromLatestSafely() throws IOException, Interru
         localFileIO.deleteQuietly(snapshotManager.snapshotPath(3));
         thread.join();
 
-        assertThat(exception.get()).hasMessageContaining("Fails to read snapshot from path");
+        assertThat(exception.get())
+                .hasMessageFindingMatch("Snapshot file .* does not exist")
+                .hasMessageContaining("dedicated compaction job");
     }
 
     @Test