diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
index f7c1b9bf6b..2fe061f0bc 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
@@ -203,8 +203,7 @@ private CompletedSnapshotStore createCompletedSnapshotStore(
                 retrievedSnapshots.add(
                         checkNotNull(snapshotStateHandle.retrieveCompleteSnapshot()));
             } catch (Exception e) {
-                if (e.getMessage()
-                        .contains(CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)) {
+                if (CompletedSnapshot.isSnapshotDataNotExists(e)) {
                     LOG.error(
                             "Metadata not found for snapshot {} of table bucket {}, maybe snapshot already removed or broken.",
                             snapshotStateHandle.getSnapshotId(),
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java
index 86cd25858b..b09c715746 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java
@@ -88,6 +88,9 @@ public class CompletedSnapshot {
 
     public static final String SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE = "No such file or directory";
 
+    private static final String HADOOP_SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE =
+            "File does not exist";
+
     public CompletedSnapshot(
             TableBucket tableBucket,
             long snapshotID,
@@ -195,6 +198,24 @@ public static FsPath getMetadataFilePath(FsPath snapshotLocation) {
         return new FsPath(snapshotLocation, SNAPSHOT_METADATA_FILE_NAME);
     }
 
+    /**
+     * Returns whether the throwable or any of its causes indicates that snapshot data no longer
+     * exists in the remote storage.
+     */
+    public static boolean isSnapshotDataNotExists(Throwable throwable) {
+        Throwable cause = throwable;
+        while (cause != null) {
+            String message = cause.getMessage();
+            if (message != null
+                    && (message.contains(SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)
+                            || message.contains(HADOOP_SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE))) {
+                return true;
+            }
+            cause = cause.getCause();
+        }
+        return false;
+    }
+
     private void disposeMetadata() throws IOException {
         FsPath metadataFilePath = getMetadataFilePath();
         FileSystem fileSystem = metadataFilePath.getFileSystem();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index f3d1ff76af..e8fcb24ae9 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -854,7 +854,7 @@ private void downloadKvSnapshots(CompletedSnapshot completedSnapshot, Path kvTab
         try {
             kvSnapshotDataDownloader.transferAllDataToDirectory(downloadSpec, closeableRegistry);
         } catch (Exception e) {
-            if (e.getMessage().contains(CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)) {
+            if (CompletedSnapshot.isSnapshotDataNotExists(e)) {
                 try {
                     snapshotContext.handleSnapshotBroken(completedSnapshot);
                 } catch (Exception t) {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
index 2558be96f9..fbb60279af 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
@@ -186,6 +186,18 @@ void testRemoveCompletedSnapshotStoreFromManager() throws Exception {
 
     @Test
     void testMetadataInconsistencyWithMetadataNotExistsException() throws Exception {
+        testMetadataInconsistencyWithMetadataNotExistsException(
+                CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE);
+    }
+
+    @Test
+    void testMetadataInconsistencyWithHadoopFileDoesNotExistException() throws Exception {
+        testMetadataInconsistencyWithMetadataNotExistsException(
+                "File does not exist: /remote/snapshot/CURRENT");
+    }
+
+    private void testMetadataInconsistencyWithMetadataNotExistsException(String errorMessage)
+            throws Exception {
         // setup test data with mixed valid and invalid snapshots
         TableBucket tableBucket = new TableBucket(1, 1);
         CompletedSnapshot validSnapshot =
@@ -196,7 +208,7 @@ void testMetadataInconsistencyWithMetadataNotExistsException() throws Exception
         CompletedSnapshot invalidSnapshot =
                 KvTestUtils.mockCompletedSnapshot(tempDir, tableBucket, 2L);
         TestingCompletedSnapshotHandle invalidSnapshotHandle =
-                new TestingCompletedSnapshotHandleWithFileNotFound(invalidSnapshot);
+                new TestingCompletedSnapshotHandleWithFileNotFound(invalidSnapshot, errorMessage);
 
         // create CompletedSnapshotHandleStore with real implementations
         CompletedSnapshotHandleStore completedSnapshotHandleStore =
@@ -284,14 +296,17 @@ private Set<TableBucket> createTableBuckets(int tableNum, int bucketNum) {
     private static class TestingCompletedSnapshotHandleWithFileNotFound
             extends TestingCompletedSnapshotHandle {
 
-        public TestingCompletedSnapshotHandleWithFileNotFound(CompletedSnapshot snapshot) {
+        private final String errorMessage;
+
+        public TestingCompletedSnapshotHandleWithFileNotFound(
+                CompletedSnapshot snapshot, String errorMessage) {
             super(snapshot, false);
+            this.errorMessage = errorMessage;
         }
 
         @Override
         public CompletedSnapshot retrieveCompleteSnapshot() throws IOException {
-            throw new FileNotFoundException(
-                    CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE);
+            throw new FileNotFoundException(errorMessage);
         }
     }
 
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java
index 44b6d4f3b6..7518fd1fe2 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java
@@ -105,6 +105,28 @@ void testKvSnapshotSize(@TempDir Path tempDir) throws Exception {
         assertThat(snapshot.getKvSnapshotHandle().getSnapshotSize()).isEqualTo(400L);
     }
 
+    @Test
+    void testIsSnapshotDataNotExists() {
+        assertThat(
+                        CompletedSnapshot.isSnapshotDataNotExists(
+                                new IOException(
+                                        CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)))
+                .isTrue();
+        assertThat(
+                        CompletedSnapshot.isSnapshotDataNotExists(
+                                new IOException("File does not exist: /remote/snapshot/CURRENT")))
+                .isTrue();
+        assertThat(
+                        CompletedSnapshot.isSnapshotDataNotExists(
+                                new IOException(
+                                        "Failed to download snapshot.",
+                                        new IOException(
+                                                "File does not exist: /remote/snapshot/CURRENT"))))
+                .isTrue();
+        assertThat(CompletedSnapshot.isSnapshotDataNotExists(new IOException("Access denied")))
+                .isFalse();
+    }
+
     private void checkCompletedSnapshotCleanUp(
             Path snapshotPath, KvSnapshotHandle kvSnapshotHandle, boolean isShareFileShouldDelete) {
         // private should be deleted, but the local file should still remain