From f71b66eb658be383ae3e3abc14a098e91d564e48 Mon Sep 17 00:00:00 2001 From: Francis Altomare Date: Wed, 22 Jul 2026 12:57:45 +0200 Subject: [PATCH] FLINK-37686 - Do not fail on NOT_OWNED deleted files --- .../state/forst/fs/ForStFlinkFileSystem.java | 29 ++++++- .../forst/fs/ForStFlinkFileSystemTest.java | 76 +++++++++++++++++++ 2 files changed, 104 insertions(+), 1 deletion(-) diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java index 6c7acd8b270772..faae71bd26e927 100644 --- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java +++ b/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]); diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java index b6f61afdb66f6a..3ecefb61dce650 100644 --- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java +++ b/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 =