FLINK-37686 - Do not fail on NOT_OWNED deleted files

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java

@@ -52,6 +52,7 @@
 
 import java.io.Closeable;
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
@@ -373,7 +374,33 @@ public synchronized FileStatus[] listStatus(Path path) throws IOException {
             String relativePath = mappingFile.substring(pathStr.length());
             int slashIndex = relativePath.indexOf('/');
             if (slashIndex == -1) { // direct child
-                fileStatuses.add(getFileStatus(new Path(mappingFile)));
+                try {
+                    fileStatuses.add(getFileStatus(new Path(mappingFile)));
+                } catch (FileNotFoundException e) {
+                    // The physical object backing this mapping entry is gone,
+                    // we handle this case by removing the mapping. If the entry is
+                    // NOT_OWNED, ownership of that object was handed to the JobManager.
+                    // The JM's SharedStateRegistry may have deleted it once the last
+                    // checkpoint referencing it was removed.
+                    // Since nothing else prunes the now-deleted mapping entry,
+                    // every subsequent full listing would hit this missing object and fail
+                    // This bug is documented in (FLINK-37686), the block attempts to
+                    // drop the stale entry and skip it rather than failing.
+                    MappingEntry staleEntry = fileMappingManager.mappingEntry(mappingFile);
+                    if (staleEntry != null
+                            && staleEntry.getFileOwnership() == FileOwnership.NOT_OWNED) {
+                        LOG.warn(
+                                "Source object for mapping file {} (resolved source: {}, ownership: NOT_OWNED) "
+                                        + "is missing. Its physical object was owned and deleted by the "
+                                        + "JobManager, the now-deleted mapping is being removed",
+                                mappingFile,
+                                staleEntry.getSourcePath());
+                        fileMappingManager.deleteFileOrDirectory(new Path(mappingFile), false);
+                    } else {
+                        // for all other cases, continue to bubble up the failure
+                        throw e;
+                    }
+                }
             }
         }
         return fileStatuses.toArray(new FileStatus[0]);

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java

@@ -36,6 +36,7 @@
 import org.apache.flink.state.forst.fs.cache.FileCacheEntry;
 import org.apache.flink.state.forst.fs.cache.SizeBasedCacheLimitPolicy;
 import org.apache.flink.state.forst.fs.cache.SpaceBasedCacheLimitPolicy;
+import org.apache.flink.state.forst.fs.filemapping.FileOwnership;
 import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
 import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
 import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
@@ -49,6 +50,7 @@
 import org.junit.jupiter.api.io.TempDir;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -68,6 +70,7 @@
 
 import static org.apache.flink.state.forst.ForStOptions.CACHE_LRU_ACCESS_BEFORE_PROMOTION;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link ForStFlinkFileSystem}. */
 @ExtendWith(ParameterizedTestExtension.class)
@@ -471,6 +474,79 @@ public FileStatus getFileStatus(org.apache.flink.core.fs.Path path) {
                                 .getLen());
     }
 
+    @Test
+    public void testListStatusDropsDeletedNotOwnedEntryAfterGiveUpOwnership() throws IOException {
+        org.apache.flink.core.fs.Path remotePath =
+                new org.apache.flink.core.fs.Path(tempDir.toString() + "/remote");
+        org.apache.flink.core.fs.Path localPath =
+                new org.apache.flink.core.fs.Path(tempDir.toString() + "/local");
+        ForStFlinkFileSystem fileSystem =
+                new ForStFlinkFileSystem(
+                        new ByteBufferReadableLocalFileSystem(),
+                        remotePath.toString(),
+                        localPath.toString(),
+                        null);
+        fileSystem.mkdirs(remotePath);
+        fileSystem.mkdirs(localPath);
+
+        org.apache.flink.core.fs.Path survivorPath =
+                new org.apache.flink.core.fs.Path(remotePath, "survivor.sst");
+        try (ByteBufferWritableFSDataOutputStream os = fileSystem.create(survivorPath)) {
+            os.write(1);
+        }
+
+        org.apache.flink.core.fs.Path notOwnedPath =
+                new org.apache.flink.core.fs.Path(remotePath, "not-owned.sst");
+        try (ByteBufferWritableFSDataOutputStream os = fileSystem.create(notOwnedPath)) {
+            os.write(2);
+        }
+        org.apache.flink.core.fs.Path notOwnedSource =
+                fileSystem.getMappingEntry(notOwnedPath).getSourcePath();
+        fileSystem.giveUpOwnership(notOwnedPath, new FileStateHandle(notOwnedSource, 1L));
+        assertThat(fileSystem.getMappingEntry(notOwnedPath).getFileOwnership())
+                .isEqualTo(FileOwnership.NOT_OWNED);
+
+        // Now the JM is deleting the physical object.
+        fileSystem.getDelegateFS().delete(notOwnedSource, false);
+
+        FileStatus[] statuses = fileSystem.listStatus(remotePath);
+        assertThat(statuses).hasSize(1);
+        assertThat(statuses[0].getPath()).isEqualTo(survivorPath);
+        assertThat(fileSystem.getMappingEntry(notOwnedPath)).isNull();
+        assertThat(fileSystem.getMappingEntry(survivorPath)).isNotNull();
+    }
+
+    @Test
+    public void testListStatusStillFailsForDeletedDbOwnedFile() throws IOException {
+        org.apache.flink.core.fs.Path remotePath =
+                new org.apache.flink.core.fs.Path(tempDir.toString() + "/remote");
+        org.apache.flink.core.fs.Path localPath =
+                new org.apache.flink.core.fs.Path(tempDir.toString() + "/local");
+        ForStFlinkFileSystem fileSystem =
+                new ForStFlinkFileSystem(
+                        new ByteBufferReadableLocalFileSystem(),
+                        remotePath.toString(),
+                        localPath.toString(),
+                        null);
+        fileSystem.mkdirs(remotePath);
+        fileSystem.mkdirs(localPath);
+
+        org.apache.flink.core.fs.Path dbOwnedPath =
+                new org.apache.flink.core.fs.Path(remotePath, "db-owned.sst");
+        try (ByteBufferWritableFSDataOutputStream os = fileSystem.create(dbOwnedPath)) {
+            os.write(1);
+        }
+        MappingEntry entry = fileSystem.getMappingEntry(dbOwnedPath);
+        assertThat(entry.getFileOwnership()).isEqualTo(FileOwnership.SHAREABLE_OWNED_BY_DB);
+
+        // Delete the physical object out from under the DB-owned entry.
+        fileSystem.getDelegateFS().delete(entry.getSourcePath(), false);
+
+        assertThatThrownBy(() -> fileSystem.listStatus(remotePath))
+                .isInstanceOf(FileNotFoundException.class);
+        assertThat(fileSystem.getMappingEntry(dbOwnedPath)).isNotNull();
+    }
+
     @Test
     public void testOverride() throws IOException {
         org.apache.flink.core.fs.Path remotePath =