You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/05/04 11:22:08 UTC
[flink] branch release-1.17 updated: Revert "[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.
mapohl 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 8dd0b770936 Revert "[FLINK-31984][fs][checkpoint] Savepoint should be relocatable if entropy injection is not effective"
8dd0b770936 is described below
commit 8dd0b770936b14dcdc23a392e3a00db5ffa2d50f
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Thu May 4 13:15:16 2023 +0200
Revert "[FLINK-31984][fs][checkpoint] Savepoint should be relocatable if entropy injection is not effective"
This reverts commit 5c0a53534edb34eeee4bbff09fc5ea4dcc5983c8.
---
.../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, 15 insertions(+), 43 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 93f6c13c36e..57680365388 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,11 +103,9 @@ public class EntropyInjector {
// ------------------------------------------------------------------------
- public static boolean isEntropyInjecting(FileSystem fs, Path target) {
+ public static boolean isEntropyInjecting(FileSystem fs) {
final EntropyInjectingFileSystem entropyFs = getEntropyFs(fs);
- return entropyFs != null
- && entropyFs.getEntropyInjectionKey() != null
- && target.getPath().contains(entropyFs.getEntropyInjectionKey());
+ return entropyFs != null && entropyFs.getEntropyInjectionKey() != null;
}
@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 46fa84494d3..eb1a5acf769 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,29 +216,17 @@ public class EntropyInjectorTest {
}
@Test
- 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));
+ public void testIsEntropyFs() {
+ final FileSystem efs = new TestEntropyInjectingFs("test", "ignored");
+
+ assertTrue(EntropyInjector.isEntropyInjecting(efs));
}
@Test
- public void testIsEntropyFsWithNullEntropyKey() throws Exception {
+ public void testIsEntropyFsWithNullEntropyKey() {
final FileSystem efs = new TestEntropyInjectingFs(null, "ignored");
- 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));
+ assertFalse(EntropyInjector.isEntropyInjecting(efs));
}
// ------------------------------------------------------------------------
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 dd687b95c88..8774848cb8c 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,6 +86,9 @@ 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;
@@ -132,6 +135,7 @@ 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 =
@@ -152,8 +156,6 @@ 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 b9d85f98532..d2ad53edc70 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,7 +30,6 @@ 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;
@@ -96,9 +95,8 @@ public class FsCheckpointStreamFactoryTest {
@Test
public void testEntropyMakesExclusiveStateAbsolutePaths() throws IOException {
- final FsStateBackendEntropyTest.TestEntropyAwareFs fs =
- new FsStateBackendEntropyTest.TestEntropyAwareFs();
- final FsCheckpointStreamFactory factory = createFactory(fs, 0);
+ final FsCheckpointStreamFactory factory =
+ createFactory(new FsStateBackendEntropyTest.TestEntropyAwareFs(), 0);
final FsCheckpointStreamFactory.FsCheckpointStateOutputStream stream =
factory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
@@ -107,9 +105,7 @@ public class FsCheckpointStreamFactoryTest {
assertThat(handle, instanceOf(FileStateHandle.class));
assertThat(handle, not(instanceOf(RelativeFileStateHandle.class)));
- assertPathsEqual(
- exclusiveStateDir.resolve(fs.generateEntropy()),
- ((FileStateHandle) handle).getFilePath().getParent());
+ assertPathsEqual(exclusiveStateDir, ((FileStateHandle) handle).getFilePath().getParent());
}
@Test
@@ -162,18 +158,6 @@ 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(