You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/04/08 07:21:11 UTC
[flink] branch release-1.11 updated: [FLINK-22081][core] Fix
entropy injection metadata path in pluggable HadoopS3FileSystem
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 3bd44e0 [FLINK-22081][core] Fix entropy injection metadata path in pluggable HadoopS3FileSystem
3bd44e0 is described below
commit 3bd44e083c86bb95bfc11a56d91e4458f6384ed7
Author: chen qin <qi...@gmail.com>
AuthorDate: Tue Mar 30 18:13:34 2021 -0700
[FLINK-22081][core] Fix entropy injection metadata path in pluggable HadoopS3FileSystem
---
.../org/apache/flink/core/fs/EntropyInjector.java | 24 ++++++++--------------
.../flink/core/fs/PluginFileSystemFactory.java | 7 +++++--
.../apache/flink/core/fs/EntropyInjectorTest.java | 22 +++++++++++++++++++-
3 files changed, 34 insertions(+), 19 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 bc63ae3..d8e22d2 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
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.WrappingProxy;
import javax.annotation.Nullable;
@@ -95,23 +96,14 @@ public class EntropyInjector {
private static EntropyInjectingFileSystem getEntropyFs(FileSystem fs) {
if (fs instanceof EntropyInjectingFileSystem) {
return (EntropyInjectingFileSystem) fs;
- } else if (fs instanceof SafetyNetWrapperFileSystem) {
- FileSystem delegate = ((SafetyNetWrapperFileSystem) fs).getWrappedDelegate();
- if (delegate instanceof EntropyInjectingFileSystem) {
- return (EntropyInjectingFileSystem) delegate;
- } else if (delegate instanceof PluginFileSystemFactory.ClassLoaderFixingFileSystem) {
- FileSystem innerFs =
- ((PluginFileSystemFactory.ClassLoaderFixingFileSystem) delegate).getInner();
- if (innerFs instanceof EntropyInjectingFileSystem) {
- return (EntropyInjectingFileSystem) innerFs;
- }
- return null;
- } else {
- return null;
- }
- } else {
- return null;
}
+
+ if (fs instanceof WrappingProxy) {
+ FileSystem delegate = ((WrappingProxy<FileSystem>) fs).getWrappedDelegate();
+ return getEntropyFs(delegate);
+ }
+
+ return null;
}
@VisibleForTesting
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
index 994062a..0d1e889 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
@@ -19,6 +19,7 @@ package org.apache.flink.core.fs;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.TemporaryClassLoaderContext;
+import org.apache.flink.util.WrappingProxy;
import java.io.IOException;
import java.net.URI;
@@ -67,7 +68,8 @@ public class PluginFileSystemFactory implements FileSystemFactory {
return String.format("Plugin %s", inner.getClass().getName());
}
- static class ClassLoaderFixingFileSystem extends FileSystem {
+ static class ClassLoaderFixingFileSystem extends FileSystem
+ implements WrappingProxy<FileSystem> {
private final FileSystem inner;
private final ClassLoader loader;
@@ -190,7 +192,8 @@ public class PluginFileSystemFactory implements FileSystemFactory {
}
}
- public FileSystem getInner() {
+ @Override
+ public FileSystem getWrappedDelegate() {
return inner;
}
}
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 5118a75..802f2ec 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
@@ -169,7 +169,7 @@ public class EntropyInjectorTest {
}
@Test
- public void testClassLoaderFixingFs() throws Exception {
+ public void testClassLoaderFixingFsWithSafeyNet() throws Exception {
final String entropyKey = "__ekey__";
final String entropyValue = "abc";
@@ -195,6 +195,26 @@ public class EntropyInjectorTest {
}
@Test
+ public void testClassLoaderFixingFsWithoutSafeyNet() throws Exception {
+ final String entropyKey = "__ekey__";
+ final String entropyValue = "abc";
+
+ final File folder = TMP_FOLDER.newFolder();
+
+ final Path path = new Path(Path.fromLocalFile(folder), entropyKey + "/path/");
+ final Path pathWithEntropy = new Path(Path.fromLocalFile(folder), entropyValue + "/path/");
+
+ PluginFileSystemFactory pluginFsFactory =
+ PluginFileSystemFactory.of(new TestFileSystemFactory(entropyKey, entropyValue));
+ FileSystem testFs = pluginFsFactory.create(URI.create("test"));
+
+ OutputStreamAndPath streamAndPath =
+ EntropyInjector.createEntropyAware(testFs, path, WriteMode.NO_OVERWRITE);
+
+ assertEquals(pathWithEntropy, streamAndPath.path());
+ }
+
+ @Test
public void testIsEntropyFs() {
final FileSystem efs = new TestEntropyInjectingFs("test", "ignored");