You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/05/04 08:49:53 UTC
[flink] branch release-1.17 updated: [FLINK-31984][fs][checkpoint] Savepoint should be relocatable if entropy injection is not effective
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 5c0a53534ed [FLINK-31984][fs][checkpoint] Savepoint should be relocatable if entropy injection is not effective
5c0a53534ed is described below
commit 5c0a53534edb34eeee4bbff09fc5ea4dcc5983c8
Author: Mingliang Liu <li...@apache.org>
AuthorDate: Tue May 2 16:18:53 2023 -0700
[FLINK-31984][fs][checkpoint] Savepoint should be relocatable if entropy injection is not effective
---
.../org/apache/flink/core/fs/EntropyInjector.java | 6 ++++--
.../apache/flink/core/fs/EntropyInjectorTest.java | 24 ++++++++++++++++------
.../filesystem/FsCheckpointStreamFactory.java | 6 ++----
.../filesystem/FsCheckpointStreamFactoryTest.java | 22 +++++++++++++++++---
4 files changed, 43 insertions(+), 15 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/EntropyInjector.java b/flink-core/src/main/java/org/apache/flink/core/fs/EntropyInjector.java
index 57680365388..93f6c13c36e 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/EntropyInjector.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/EntropyInjector.java
@@ -103,9 +103,11 @@ public class EntropyInjector {
// ------------------------------------------------------------------------
- public static boolean isEntropyInjecting(FileSystem fs) {
+ public static boolean isEntropyInjecting(FileSystem fs, Path target) {
final EntropyInjectingFileSystem entropyFs = getEntropyFs(fs);
- return entropyFs != null && entropyFs.getEntropyInjectionKey() != null;
+ return entropyFs != null
+ && entropyFs.getEntropyInjectionKey() != null
+ && target.getPath().contains(entropyFs.getEntropyInjectionKey());
}
@Nullable
diff --git a/flink-core/src/test/java/org/apache/flink/core/fs/EntropyInjectorTest.java b/flink-core/src/test/java/org/apache/flink/core/fs/EntropyInjectorTest.java
index eb1a5acf769..46fa84494d3 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/EntropyInjectorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/EntropyInjectorTest.java
@@ -216,17 +216,29 @@ public class EntropyInjectorTest {
}
@Test
- public void testIsEntropyFs() {
- final FileSystem efs = new TestEntropyInjectingFs("test", "ignored");
-
- assertTrue(EntropyInjector.isEntropyInjecting(efs));
+ public void testIsEntropyFs() throws Exception {
+ final String entropyKey = "_test_";
+ final FileSystem efs = new TestEntropyInjectingFs(entropyKey, "ignored");
+ final File folder = TMP_FOLDER.newFolder();
+ final Path path = new Path(Path.fromLocalFile(folder), entropyKey + "/path/");
+ assertTrue(EntropyInjector.isEntropyInjecting(efs, path));
}
@Test
- public void testIsEntropyFsWithNullEntropyKey() {
+ public void testIsEntropyFsWithNullEntropyKey() throws Exception {
final FileSystem efs = new TestEntropyInjectingFs(null, "ignored");
- assertFalse(EntropyInjector.isEntropyInjecting(efs));
+ final File folder = TMP_FOLDER.newFolder();
+ assertFalse(EntropyInjector.isEntropyInjecting(efs, Path.fromLocalFile(folder)));
+ }
+
+ @Test
+ public void testIsEntropyFsPathDoesNotIncludeEntropyKey() throws Exception {
+ final String entropyKey = "_test_";
+ final FileSystem efs = new TestEntropyInjectingFs(entropyKey, "ignored");
+ final File folder = TMP_FOLDER.newFolder();
+ final Path path = new Path(Path.fromLocalFile(folder), "path"); // no entropy key
+ assertFalse(EntropyInjector.isEntropyInjecting(efs, path));
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
index 8774848cb8c..dd687b95c88 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
@@ -86,9 +86,6 @@ public class FsCheckpointStreamFactory implements CheckpointStreamFactory {
/** Cached handle to the file system for file operations. */
private final FileSystem filesystem;
- /** Whether the file system dynamically injects entropy into the file paths. */
- private final boolean entropyInjecting;
-
private final FsCheckpointStateToolset privateStateToolset;
private final FsCheckpointStateToolset sharedStateToolset;
@@ -135,7 +132,6 @@ public class FsCheckpointStreamFactory implements CheckpointStreamFactory {
this.sharedStateDirectory = checkNotNull(sharedStateDirectory);
this.fileStateThreshold = fileStateSizeThreshold;
this.writeBufferSize = writeBufferSize;
- this.entropyInjecting = EntropyInjector.isEntropyInjecting(fileSystem);
if (fileSystem instanceof DuplicatingFileSystem) {
final DuplicatingFileSystem duplicatingFileSystem = (DuplicatingFileSystem) fileSystem;
this.privateStateToolset =
@@ -156,6 +152,8 @@ public class FsCheckpointStreamFactory implements CheckpointStreamFactory {
Path target = getTargetPath(scope);
int bufferSize = Math.max(writeBufferSize, fileStateThreshold);
+ // Whether the file system dynamically injects entropy into the file paths.
+ final boolean entropyInjecting = EntropyInjector.isEntropyInjecting(filesystem, target);
final boolean absolutePath = entropyInjecting || scope == CheckpointedStateScope.SHARED;
return new FsCheckpointStateOutputStream(
target, filesystem, bufferSize, fileStateThreshold, !absolutePath);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactoryTest.java
index d2ad53edc70..b9d85f98532 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactoryTest.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+import java.util.Objects;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
@@ -95,8 +96,9 @@ public class FsCheckpointStreamFactoryTest {
@Test
public void testEntropyMakesExclusiveStateAbsolutePaths() throws IOException {
- final FsCheckpointStreamFactory factory =
- createFactory(new FsStateBackendEntropyTest.TestEntropyAwareFs(), 0);
+ final FsStateBackendEntropyTest.TestEntropyAwareFs fs =
+ new FsStateBackendEntropyTest.TestEntropyAwareFs();
+ final FsCheckpointStreamFactory factory = createFactory(fs, 0);
final FsCheckpointStreamFactory.FsCheckpointStateOutputStream stream =
factory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
@@ -105,7 +107,9 @@ public class FsCheckpointStreamFactoryTest {
assertThat(handle, instanceOf(FileStateHandle.class));
assertThat(handle, not(instanceOf(RelativeFileStateHandle.class)));
- assertPathsEqual(exclusiveStateDir, ((FileStateHandle) handle).getFilePath().getParent());
+ assertPathsEqual(
+ exclusiveStateDir.resolve(fs.generateEntropy()),
+ ((FileStateHandle) handle).getFilePath().getParent());
}
@Test
@@ -158,6 +162,18 @@ public class FsCheckpointStreamFactoryTest {
return createFactory(fs, fileSizeThreshold, 4096);
}
+ private FsCheckpointStreamFactory createFactory(
+ FsStateBackendEntropyTest.TestEntropyAwareFs fs, int fileSizeThreshold) {
+ final Path exclusiveStateDirWithEntropy =
+ exclusiveStateDir.resolve(Objects.requireNonNull(fs.getEntropyInjectionKey()));
+ return new FsCheckpointStreamFactory(
+ fs,
+ new org.apache.flink.core.fs.Path(exclusiveStateDirWithEntropy.toUri()),
+ new org.apache.flink.core.fs.Path(sharedStateDir.toUri()),
+ fileSizeThreshold,
+ 4096);
+ }
+
private FsCheckpointStreamFactory createFactory(
FileSystem fs, int fileSizeThreshold, int bufferSize) {
return new FsCheckpointStreamFactory(