This is an automated email from the ASF dual-hosted git repository. RocMarshal pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit f40140e5da721159848dd3f9e490f636d6751226 Author: Mohsen Rezaei <[email protected]> AuthorDate: Tue Oct 13 11:24:23 2026 -0700 added validation framework to guard for Forst and object stores (cherry picked from commit f2ff56d443bf8e8c31856c0b6fc9c311b85e4027) --- .../flink/state/forst/ForStStateBackendV2Test.java | 41 +++++- .../state/forst/ObjectStoreLikeFileSystem.java | 139 +++++++++++++++++++++ ...tFileSystemTrackingCreatedDirDecoratorTest.java | 42 +++++++ .../org.apache.flink.core.fs.FileSystemFactory | 16 +++ 4 files changed, 236 insertions(+), 2 deletions(-) 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
