This is an automated email from the ASF dual-hosted git repository.
Zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9542791071b [FLINK-37686] Do not fail on NOT_OWNED deleted files
(#28801)
9542791071b is described below
commit 9542791071b73881c59966102c9d5a17e824b427
Author: Francis <[email protected]>
AuthorDate: Wed Jul 29 17:56:46 2026 +0200
[FLINK-37686] Do not fail on NOT_OWNED deleted files (#28801)
---
.../flink/state/forst/fs/ForStFlinkFileSystem.java | 29 ++++++++-
.../state/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 6c7acd8b270..faae71bd26e 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 javax.annotation.Nullable;
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 class ForStFlinkFileSystem extends FileSystem
implements Closeable {
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 b6f61afdb66..b41d0e548f1 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.FileBasedCache;
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.extension.ExtendWith;
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 java.util.concurrent.atomic.AtomicReference;
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 class ForStFlinkFileSystemTest {
.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 =
