You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/06/19 11:17:12 UTC
[flink] 01/02: [FLINK-12143][runtime] Workaround for dynamic class
loading in FS creation: set thread context class loader to plugin's class
loader
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fe92f6a91798826c4e7d6796376d6ceae919498b
Author: Aleksey Pak <al...@ververica.com>
AuthorDate: Mon Jun 17 14:57:52 2019 +0200
[FLINK-12143][runtime] Workaround for dynamic class loading in FS creation: set thread context class loader to plugin's class loader
---
.../flink/core/fs/ConnectionLimitingFactory.java | 5 +++++
.../java/org/apache/flink/core/fs/FileSystem.java | 6 +++++-
.../java/org/apache/flink/core/plugin/Plugin.java | 13 +++++++++++++
.../org/apache/flink/core/plugin/PluginLoader.java | 20 ++++++++++++++------
.../org/apache/flink/core/plugin/PluginManager.java | 2 +-
.../apache/flink/test/plugin/PluginLoaderTest.java | 10 +++++++---
6 files changed, 45 insertions(+), 11 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java b/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
index b85a7d6..3b1141b 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
@@ -48,6 +48,11 @@ public class ConnectionLimitingFactory implements FileSystemFactory {
// ------------------------------------------------------------------------
@Override
+ public ClassLoader getClassLoader() {
+ return factory.getClassLoader();
+ }
+
+ @Override
public String getScheme() {
return factory.getScheme();
}
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index d159e70..b57f1f6 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -32,6 +32,7 @@ import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.fs.local.LocalFileSystemFactory;
import org.apache.flink.core.plugin.PluginManager;
+import org.apache.flink.core.plugin.TemporaryClassLoaderContext;
import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
@@ -432,7 +433,10 @@ public abstract class FileSystem {
final FileSystemFactory factory = FS_FACTORIES.get(uri.getScheme());
if (factory != null) {
- fs = factory.create(uri);
+ ClassLoader classLoader = factory.getClassLoader();
+ try (TemporaryClassLoaderContext classLoaderContext = new TemporaryClassLoaderContext(classLoader)) {
+ fs = factory.create(uri);
+ }
}
else {
try {
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/Plugin.java b/flink-core/src/main/java/org/apache/flink/core/plugin/Plugin.java
index 0edf1b1..59f5e2a 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/Plugin.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/Plugin.java
@@ -20,6 +20,7 @@ package org.apache.flink.core.plugin;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Preconditions;
/**
* Interface for plugins. Plugins typically extend this interface in their SPI and the concrete implementations of a
@@ -29,6 +30,18 @@ import org.apache.flink.configuration.Configuration;
public interface Plugin {
/**
+ * Helper method to get the class loader used to load the plugin. This may be needed for some plugins that use
+ * dynamic class loading afterwards the plugin was loaded.
+ *
+ * @return the class loader used to load the plugin.
+ */
+ default ClassLoader getClassLoader() {
+ return Preconditions.checkNotNull(
+ this.getClass().getClassLoader(),
+ "%s plugin with null class loader", this.getClass().getName());
+ }
+
+ /**
* Optional method for plugins to pick up settings from the configuration.
*
* @param config The configuration to apply to the plugin.
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
index f90bda8..eb7b1c5 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
@@ -40,12 +40,20 @@ public class PluginLoader {
private final ClassLoader pluginClassLoader;
@VisibleForTesting
- public PluginLoader(PluginDescriptor pluginDescriptor, ClassLoader parentClassLoader, String[] alwaysParentFirstPatterns) {
- this.pluginClassLoader =
- new ChildFirstClassLoader(
- pluginDescriptor.getPluginResourceURLs(),
- parentClassLoader,
- ArrayUtils.concat(alwaysParentFirstPatterns, pluginDescriptor.getLoaderExcludePatterns()));
+ public PluginLoader(ClassLoader pluginClassLoader) {
+ this.pluginClassLoader = pluginClassLoader;
+ }
+
+ @VisibleForTesting
+ public static ClassLoader createPluginClassLoader(PluginDescriptor pluginDescriptor, ClassLoader parentClassLoader, String[] alwaysParentFirstPatterns) {
+ return new ChildFirstClassLoader(
+ pluginDescriptor.getPluginResourceURLs(),
+ parentClassLoader,
+ ArrayUtils.concat(alwaysParentFirstPatterns, pluginDescriptor.getLoaderExcludePatterns()));
+ }
+
+ public static PluginLoader create(PluginDescriptor pluginDescriptor, ClassLoader parentClassLoader, String[] alwaysParentFirstPatterns) {
+ return new PluginLoader(createPluginClassLoader(pluginDescriptor, parentClassLoader, alwaysParentFirstPatterns));
}
/**
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
index c6835a2..2385732 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
@@ -66,7 +66,7 @@ public class PluginManager {
public <P extends Plugin> Iterator<P> load(Class<P> service) {
ArrayList<Iterator<P>> combinedIterators = new ArrayList<>(pluginDescriptors.size());
for (PluginDescriptor pluginDescriptor : pluginDescriptors) {
- PluginLoader pluginLoader = new PluginLoader(pluginDescriptor, parentClassLoader, alwaysParentFirstPatterns);
+ PluginLoader pluginLoader = PluginLoader.create(pluginDescriptor, parentClassLoader, alwaysParentFirstPatterns);
combinedIterators.add(pluginLoader.load(service));
}
return Iterators.concat(combinedIterators.iterator());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
index 52bad19..ec57acb 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
@@ -39,7 +39,9 @@ public class PluginLoaderTest extends PluginTestBase {
final URL classpathA = createPluginJarURLFromString(PLUGIN_A);
PluginDescriptor pluginDescriptorA = new PluginDescriptor("A", new URL[]{classpathA}, new String[0]);
- final PluginLoader pluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
+ ClassLoader pluginClassLoaderA = PluginLoader.createPluginClassLoader(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
+ Assert.assertNotEquals(PARENT_CLASS_LOADER, pluginClassLoaderA);
+ final PluginLoader pluginLoaderA = new PluginLoader(pluginClassLoaderA);
Iterator<TestSpi> testSpiIteratorA = pluginLoaderA.load(TestSpi.class);
@@ -52,14 +54,16 @@ public class PluginLoaderTest extends PluginTestBase {
Assert.assertNotNull(testSpiA.testMethod());
Assert.assertEquals(TestServiceA.class.getCanonicalName(), testSpiA.getClass().getCanonicalName());
- Assert.assertNotEquals(PARENT_CLASS_LOADER, testSpiA.getClass().getClassLoader());
+ // The plugin must return the same class loader as the one used to load it.
+ Assert.assertEquals(pluginClassLoaderA, testSpiA.getClassLoader());
+ Assert.assertEquals(pluginClassLoaderA, testSpiA.getClass().getClassLoader());
// Looks strange, but we want to ensure that those classes are not instance of each other because they were
// loaded by different classloader instances because the plugin loader uses child-before-parent order.
Assert.assertFalse(testSpiA instanceof TestServiceA);
// In the following we check for isolation of classes between different plugin loaders.
- final PluginLoader secondPluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
+ final PluginLoader secondPluginLoaderA = PluginLoader.create(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
TestSpi secondTestSpiA = secondPluginLoaderA.load(TestSpi.class).next();
Assert.assertNotNull(secondTestSpiA.testMethod());