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 474012a987c [FLINK-40931][state/forst] Track renamed directories in
`ForStFileSystemTrackingCreatedDirDecorator` (#29475)
474012a987c is described below
commit 474012a987c521f21cfb20847eba2b1ded4062cd
Author: Mohsen Rezaei <[email protected]>
AuthorDate: Sat Oct 17 04:50:28 2026 -0700
[FLINK-40931][state/forst] Track renamed directories in
`ForStFileSystemTrackingCreatedDirDecorator` (#29475)
---
...ForStFileSystemTrackingCreatedDirDecorator.java | 16 +++
.../flink/state/forst/ForStStateBackendV2Test.java | 41 +++++-
.../state/forst/ObjectStoreLikeFileSystem.java | 139 +++++++++++++++++++++
...tFileSystemTrackingCreatedDirDecoratorTest.java | 42 +++++++
.../org.apache.flink.core.fs.FileSystemFactory | 16 +++
5 files changed, 252 insertions(+), 2 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecorator.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecorator.java
index 7d7ba568b2d..5f205561f4b 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecorator.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecorator.java
@@ -56,4 +56,20 @@ public class ForStFileSystemTrackingCreatedDirDecorator
extends ForStFlinkFileSy
}
return super.exists(f);
}
+
+ /**
+ * Renaming a tracked directory moves its tracked existence to the
destination. A directory
+ * rename recreates the destination through the undecorated file system,
where mkdir() has no
+ * effect, so without this the destination is reported as missing. RocksDB
relies on it when it
+ * renames the export directory into place while restoring at a lower
parallelism with ingest-DB
+ * restore mode.
+ */
+ @Override
+ public synchronized boolean rename(Path src, Path dst) throws IOException {
+ boolean renamed = super.rename(src, dst);
+ if (renamed && createdDirPaths.remove(src)) {
+ createdDirPaths.add(dst);
+ }
+ return renamed;
+ }
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
index 60dff57672d..23100ae0288 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
@@ -111,7 +111,8 @@ class ForStStateBackendV2Test extends
StateBackendTestV2Base<ForStStateBackend>
+ " useIngestDbRestoreMode: {2},"
+ " useDeleteFileInRange: {3},"
+ " CheckpointStorage: {4}, "
- + " restoreMode: {5}")
+ + " restoreMode: {5},"
+ + " remoteOnObjectStore: {6}")
public static List<Object[]> modes() {
// ---------- base cases ----------
// combination of parameters: [hasLocalDir, hasRemoteDir,
useIngestDbRestoreMode,
@@ -137,6 +138,27 @@ class ForStStateBackendV2Test extends
StateBackendTestV2Base<ForStStateBackend>
// ---------- enumerate different cases for restoreMode ----------
cases = addAllCasesForParameter(cases, RestoreMode.values());
+ // ---------- the remote directory on an object-store-like file system
----------
+ // the cases above keep their remote directory on the plain local file
system
+ cases = addAllCasesForParameter(cases, new Object[] {false});
+ for (boolean useIngestDbRestoreMode : new boolean[] {true, false}) {
+ for (Object storage :
+ new Object[] {filesystemCheckpointStorage,
dbReusableCheckpointStorage}) {
+ for (RestoreMode restoreMode : RestoreMode.values()) {
+ cases.add(
+ new Object[] {
+ false,
+ true,
+ useIngestDbRestoreMode,
+ true,
+ storage,
+ restoreMode,
+ true
+ });
+ }
+ }
+ }
+
return cases;
}
@@ -157,11 +179,22 @@ class ForStStateBackendV2Test extends
StateBackendTestV2Base<ForStStateBackend>
@Parameter(5)
public RestoreMode restoreMode;
+ @Parameter(6)
+ public boolean remoteOnObjectStore;
+
@Override
protected CheckpointStorage getCheckpointStorage() throws Exception {
+ if (remoteOnObjectStore && storageSupplier ==
dbReusableCheckpointStorage) {
+ // the primary directory is the checkpoint directory, both on the
object store
+ return new FileSystemCheckpointStorage(new
Path(objectStoreRemoteDir()), 0, -1);
+ }
return storageSupplier.get();
}
+ private static String objectStoreRemoteDir() {
+ return ObjectStoreLikeFileSystem.SCHEME + "://" +
tempFolderForForstRemote;
+ }
+
@Override
protected ConfigurableStateBackend getStateBackend() throws Exception {
ForStStateBackend backend = new ForStStateBackend();
@@ -170,7 +203,11 @@ class ForStStateBackendV2Test extends
StateBackendTestV2Base<ForStStateBackend>
config.set(LOCAL_DIRECTORIES, tempFolderForForStLocal.toString());
}
if (hasRemoteDir) {
- config.set(PRIMARY_DIRECTORY, tempFolderForForstRemote.toString());
+ config.set(
+ PRIMARY_DIRECTORY,
+ remoteOnObjectStore
+ ? objectStoreRemoteDir()
+ : tempFolderForForstRemote.toString());
} else {
config.set(PRIMARY_DIRECTORY, LOCAL_DIR_AS_PRIMARY_SHORTCUT);
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ObjectStoreLikeFileSystem.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ObjectStoreLikeFileSystem.java
new file mode 100644
index 00000000000..4d083b1b1d8
--- /dev/null
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ObjectStoreLikeFileSystem.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst;
+
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.local.LocalFileStatus;
+import org.apache.flink.core.fs.local.LocalFileSystem;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A local file system that follows the directory semantics of S3-like object
stores (the Presto and
+ * the native Flink S3 file systems): directories are not real entities,
{@link #mkdirs} is a no-op
+ * that reports success, writing a file implicitly creates its parent
"prefix", and a directory only
+ * exists while at least one file exists somewhere below it.
+ */
+public class ObjectStoreLikeFileSystem extends LocalFileSystem {
+
+ public static final String SCHEME = "objstore";
+
+ @Override
+ public URI getUri() {
+ return URI.create(SCHEME + ":///");
+ }
+
+ @Override
+ public boolean mkdirs(Path f) {
+ return true;
+ }
+
+ @Override
+ public FSDataOutputStream create(Path filePath, WriteMode overwrite)
throws IOException {
+ createPhysicalParent(filePath);
+ return super.create(filePath, overwrite);
+ }
+
+ @Override
+ public boolean rename(Path src, Path dst) throws IOException {
+ createPhysicalParent(dst);
+ return super.rename(src, dst);
+ }
+
+ @Override
+ public boolean exists(Path f) {
+ return existsAsObjectOrPrefix(pathToFile(f));
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path f) throws IOException {
+ File file = pathToFile(f);
+ if (!existsAsObjectOrPrefix(file)) {
+ throw new FileNotFoundException("No object or prefix for " + f);
+ }
+ return new LocalFileStatus(file, this);
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path f) {
+ File dir = pathToFile(f);
+ if (!existsAsObjectOrPrefix(dir)) {
+ return new FileStatus[0];
+ }
+ if (dir.isFile()) {
+ return new FileStatus[] {new LocalFileStatus(dir, this)};
+ }
+ List<FileStatus> result = new ArrayList<>();
+ File[] children = dir.listFiles();
+ if (children != null) {
+ for (File child : children) {
+ if (existsAsObjectOrPrefix(child)) {
+ result.add(new LocalFileStatus(child, this));
+ }
+ }
+ }
+ return result.toArray(new FileStatus[0]);
+ }
+
+ private void createPhysicalParent(Path path) throws IOException {
+ Path parent = path.getParent();
+ if (parent != null) {
+ super.mkdirs(parent);
+ }
+ }
+
+ private static boolean existsAsObjectOrPrefix(File file) {
+ if (file.isFile()) {
+ return true;
+ }
+ File[] children = file.listFiles();
+ if (children == null) {
+ return false;
+ }
+ for (File child : children) {
+ if (existsAsObjectOrPrefix(child)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Factory for {@link ObjectStoreLikeFileSystem}, registered for tests
via ServiceLoader. */
+ public static final class Factory implements FileSystemFactory {
+
+ @Override
+ public String getScheme() {
+ return SCHEME;
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) {
+ return new ObjectStoreLikeFileSystem();
+ }
+ }
+}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecoratorTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecoratorTest.java
index 20b3bee3afe..9c35d3041e4 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecoratorTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFileSystemTrackingCreatedDirDecoratorTest.java
@@ -61,6 +61,48 @@ public class ForStFileSystemTrackingCreatedDirDecoratorTest {
mkdirAndCheck(true);
}
+ @Test
+ public void testRenameDirAndCheck() throws IOException {
+ renameDirAndCheck(false);
+ }
+
+ @Test
+ public void testDummyMkdirRenameDirAndCheck() throws IOException {
+ renameDirAndCheck(true);
+ }
+
+ void renameDirAndCheck(boolean enableDummyMkdir) 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 =
+ ForStFileSystemUtils.tryDecorate(
+ new ForStFlinkFileSystem(
+ new MockLocalFileSystem(enableDummyMkdir),
+ remotePath.toString(),
+ localPath.toString(),
+ null));
+
+ // mirrors RocksDB's export: create "<dir>.tmp", then rename it into
place as "<dir>"
+ String dirPathStr = genRandomFilePathStr();
+ org.apache.flink.core.fs.Path tmpDir =
+ new org.apache.flink.core.fs.Path(dirPathStr + ".tmp");
+ org.apache.flink.core.fs.Path dstDir = new
org.apache.flink.core.fs.Path(dirPathStr);
+ fileSystem.mkdirs(tmpDir);
+ assertThat(fileSystem.exists(tmpDir)).isTrue();
+ assertThat(fileSystem.exists(dstDir)).isFalse();
+
+ assertThat(fileSystem.rename(tmpDir, dstDir)).isTrue();
+ assertThat(fileSystem.exists(dstDir)).isTrue();
+ if (enableDummyMkdir) {
+ // the source was never materialized, so it is gone once it is no
longer tracked
+ // (a materialized source directory is not deleted by a ForSt
directory rename)
+ assertThat(fileSystem.exists(tmpDir)).isFalse();
+ }
+ }
+
void mkdirAndCheck(boolean enableDummyMkdir) throws IOException {
org.apache.flink.core.fs.Path remotePath =
new org.apache.flink.core.fs.Path(tempDir.toString() +
"/remote");
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory
b/flink-state-backends/flink-statebackend-forst/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory
new file mode 100644
index 00000000000..00f5293b9bf
--- /dev/null
+++
b/flink-state-backends/flink-statebackend-forst/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.state.forst.ObjectStoreLikeFileSystem$Factory
