You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2020/08/16 21:28:27 UTC
[flink] 05/05: [FLINK-16917][runtime] Added option to disable
safety net user classloader.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fce82d7f56d5da3d3bf9ea6b66888d1350eb172f
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Fri Aug 7 17:56:00 2020 +0200
[FLINK-16917][runtime] Added option to disable safety net user classloader.
If users experience class loader leaks in libraries not under his control, they can force old behavior with the config option.
---
docs/_includes/generated/core_configuration.html | 8 +++++
.../generated/expert_class_loading_section.html | 8 +++++
.../java/org/apache/flink/client/ClientUtils.java | 9 ++++-
.../apache/flink/configuration/CoreOptions.java | 11 ++++++
.../formats/avro/AvroKryoClassloadingTest.java | 3 +-
.../librarycache/BlobLibraryCacheManager.java | 18 +++++++---
.../librarycache/FlinkUserCodeClassLoaders.java | 42 +++++++++++++++-------
.../jobmaster/JobManagerSharedServices.java | 4 ++-
.../runtime/taskexecutor/TaskManagerServices.java | 5 ++-
.../librarycache/BlobLibraryCacheManagerTest.java | 3 +-
.../BlobLibraryCacheRecoveryITCase.java | 3 +-
.../FlinkUserCodeClassLoadersTest.java | 9 +++--
.../state/RocksDbMultiClassLoaderTest.java | 14 ++++++--
.../client/gateway/local/ExecutionContextTest.java | 6 ++--
14 files changed, 114 insertions(+), 29 deletions(-)
diff --git a/docs/_includes/generated/core_configuration.html b/docs/_includes/generated/core_configuration.html
index 1ad12bc..d50c975 100644
--- a/docs/_includes/generated/core_configuration.html
+++ b/docs/_includes/generated/core_configuration.html
@@ -9,6 +9,14 @@
</thead>
<tbody>
<tr>
+ <td><h5>classloader.check-leaked-classloader</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Fails attempts at loading classes if the user classloader of a job is used after it has terminated.
+This is usually caused by the classloader being leaked by lingering threads or misbehaving libraries, which may also result in the classloader being used by other jobs.
+This check should only be disabled if such a leak prevents further jobs from running.</td>
+ </tr>
+ <tr>
<td><h5>classloader.fail-on-metaspace-oom-error</h5></td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
diff --git a/docs/_includes/generated/expert_class_loading_section.html b/docs/_includes/generated/expert_class_loading_section.html
index b2d85e4..97bc7e1 100644
--- a/docs/_includes/generated/expert_class_loading_section.html
+++ b/docs/_includes/generated/expert_class_loading_section.html
@@ -9,6 +9,14 @@
</thead>
<tbody>
<tr>
+ <td><h5>classloader.check-leaked-classloader</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Fails attempts at loading classes if the user classloader of a job is used after it has terminated.
+This is usually caused by the classloader being leaked by lingering threads or misbehaving libraries, which may also result in the classloader being used by other jobs.
+This check should only be disabled if such a leak prevents further jobs from running.</td>
+ </tr>
+ <tr>
<td><h5>classloader.fail-on-metaspace-oom-error</h5></td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 8e4f5e4..c94dbc2 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -72,7 +72,14 @@ public enum ClientUtils {
configuration.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER);
FlinkUserCodeClassLoaders.ResolveOrder resolveOrder =
FlinkUserCodeClassLoaders.ResolveOrder.fromString(classLoaderResolveOrder);
- return FlinkUserCodeClassLoaders.create(resolveOrder, urls, parent, alwaysParentFirstLoaderPatterns, NOOP_EXCEPTION_HANDLER);
+ final boolean checkClassloaderLeak = configuration.getBoolean(CoreOptions.CHECK_LEAKED_CLASSLOADER);
+ return FlinkUserCodeClassLoaders.create(
+ resolveOrder,
+ urls,
+ parent,
+ alwaysParentFirstLoaderPatterns,
+ NOOP_EXCEPTION_HANDLER,
+ checkClassloaderLeak);
}
public static JobExecutionResult submitJob(
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
index 6cc6190..0ae8e66 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
@@ -124,6 +124,17 @@ public class CoreOptions {
return parseParentFirstLoaderPatterns(base, append);
}
+ @Documentation.Section(Documentation.Sections.EXPERT_CLASS_LOADING)
+ public static final ConfigOption<Boolean> CHECK_LEAKED_CLASSLOADER = ConfigOptions
+ .key("classloader.check-leaked-classloader")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("Fails attempts at loading classes if the user classloader of a job is used after it has " +
+ "terminated.\n" +
+ "This is usually caused by the classloader being leaked by lingering threads or misbehaving libraries, " +
+ "which may also result in the classloader being used by other jobs.\n" +
+ "This check should only be disabled if such a leak prevents further jobs from running.");
+
/**
* Plugin-specific option of {@link #ALWAYS_PARENT_FIRST_LOADER_PATTERNS}. Plugins use this parent first list
* instead of the global version.
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroKryoClassloadingTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroKryoClassloadingTest.java
index b00e415..53c9b49 100644
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroKryoClassloadingTest.java
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroKryoClassloadingTest.java
@@ -75,7 +75,8 @@ public class AvroKryoClassloadingTest {
new URL[] { avroLocation, kryoLocation },
parentClassLoader,
CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS.defaultValue().split(";"),
- NOOP_EXCEPTION_HANDLER);
+ NOOP_EXCEPTION_HANDLER,
+ true);
final Class<?> userLoadedAvroClass = Class.forName(avroClass.getName(), false, userAppClassLoader);
assertNotEquals(avroClass, userLoadedAvroClass);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
index 624720b..9f0908b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
@@ -148,13 +148,20 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
*/
private final Consumer<Throwable> classLoadingExceptionHandler;
+ /**
+ * Test if classloader is used outside of job.
+ */
+ private final boolean checkClassLoaderLeak;
+
private DefaultClassLoaderFactory(
FlinkUserCodeClassLoaders.ResolveOrder classLoaderResolveOrder,
String[] alwaysParentFirstPatterns,
- Consumer<Throwable> classLoadingExceptionHandler) {
+ Consumer<Throwable> classLoadingExceptionHandler,
+ boolean checkClassLoaderLeak) {
this.classLoaderResolveOrder = classLoaderResolveOrder;
this.alwaysParentFirstPatterns = alwaysParentFirstPatterns;
this.classLoadingExceptionHandler = classLoadingExceptionHandler;
+ this.checkClassLoaderLeak = checkClassLoaderLeak;
}
@Override
@@ -164,18 +171,21 @@ public class BlobLibraryCacheManager implements LibraryCacheManager {
libraryURLs,
FlinkUserCodeClassLoaders.class.getClassLoader(),
alwaysParentFirstPatterns,
- classLoadingExceptionHandler);
+ classLoadingExceptionHandler,
+ checkClassLoaderLeak);
}
}
public static ClassLoaderFactory defaultClassLoaderFactory(
FlinkUserCodeClassLoaders.ResolveOrder classLoaderResolveOrder,
String[] alwaysParentFirstPatterns,
- @Nullable FatalErrorHandler fatalErrorHandlerJvmMetaspaceOomError) {
+ @Nullable FatalErrorHandler fatalErrorHandlerJvmMetaspaceOomError,
+ boolean checkClassLoaderLeak) {
return new DefaultClassLoaderFactory(
classLoaderResolveOrder,
alwaysParentFirstPatterns,
- createClassLoadingExceptionHandler(fatalErrorHandlerJvmMetaspaceOomError));
+ createClassLoadingExceptionHandler(fatalErrorHandlerJvmMetaspaceOomError),
+ checkClassLoaderLeak);
}
private static Consumer<Throwable> createClassLoadingExceptionHandler(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoaders.java
index 81f2535..4bbe00d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoaders.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.execution.librarycache;
+import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.util.ChildFirstClassLoader;
import org.apache.flink.util.FlinkUserCodeClassLoader;
@@ -43,20 +44,24 @@ public class FlinkUserCodeClassLoaders {
public static URLClassLoader parentFirst(
URL[] urls,
ClassLoader parent,
- Consumer<Throwable> classLoadingExceptionHandler) {
- return new SafetyNetWrapperClassLoader(
- new ParentFirstClassLoader(urls, parent, classLoadingExceptionHandler),
- parent);
+ Consumer<Throwable> classLoadingExceptionHandler,
+ boolean checkClassLoaderLeak) {
+ FlinkUserCodeClassLoader classLoader = new ParentFirstClassLoader(urls, parent, classLoadingExceptionHandler);
+ return wrapWithSafetyNet(classLoader, checkClassLoaderLeak);
}
public static URLClassLoader childFirst(
URL[] urls,
ClassLoader parent,
String[] alwaysParentFirstPatterns,
- Consumer<Throwable> classLoadingExceptionHandler) {
- return new SafetyNetWrapperClassLoader(
- new ChildFirstClassLoader(urls, parent, alwaysParentFirstPatterns, classLoadingExceptionHandler),
- parent);
+ Consumer<Throwable> classLoadingExceptionHandler,
+ boolean checkClassLoaderLeak) {
+ FlinkUserCodeClassLoader classLoader = new ChildFirstClassLoader(
+ urls,
+ parent,
+ alwaysParentFirstPatterns,
+ classLoadingExceptionHandler);
+ return wrapWithSafetyNet(classLoader, checkClassLoaderLeak);
}
public static URLClassLoader create(
@@ -64,18 +69,28 @@ public class FlinkUserCodeClassLoaders {
URL[] urls,
ClassLoader parent,
String[] alwaysParentFirstPatterns,
- Consumer<Throwable> classLoadingExceptionHandler) {
+ Consumer<Throwable> classLoadingExceptionHandler,
+ boolean checkClassLoaderLeak) {
switch (resolveOrder) {
case CHILD_FIRST:
- return childFirst(urls, parent, alwaysParentFirstPatterns, classLoadingExceptionHandler);
+ return childFirst(
+ urls,
+ parent,
+ alwaysParentFirstPatterns,
+ classLoadingExceptionHandler,
+ checkClassLoaderLeak);
case PARENT_FIRST:
- return parentFirst(urls, parent, classLoadingExceptionHandler);
+ return parentFirst(urls, parent, classLoadingExceptionHandler, checkClassLoaderLeak);
default:
throw new IllegalArgumentException("Unknown class resolution order: " + resolveOrder);
}
}
+ private static URLClassLoader wrapWithSafetyNet(FlinkUserCodeClassLoader classLoader, boolean check) {
+ return check ? new SafetyNetWrapperClassLoader(classLoader, classLoader.getParent()) : classLoader;
+ }
+
/**
* Class resolution order for Flink URL {@link ClassLoader}.
*/
@@ -140,7 +155,10 @@ public class FlinkUserCodeClassLoaders {
private FlinkUserCodeClassLoader ensureInner() {
if (inner == null) {
- throw new IllegalStateException("Trying to access closed classloader");
+ throw new IllegalStateException("Trying to access closed classloader. Please check if you store " +
+ "classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak " +
+ "occurs in a third party library and cannot be fixed immediately, you can disable this check " +
+ "with the configuration '" + CoreOptions.CHECK_LEAKED_CLASSLOADER.key() + "'.");
}
return inner;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
index b2bbacf..92963c2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
@@ -138,13 +138,15 @@ public class JobManagerSharedServices {
final String[] alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(config);
final boolean failOnJvmMetaspaceOomError = config.getBoolean(CoreOptions.FAIL_ON_USER_CLASS_LOADING_METASPACE_OOM);
+ final boolean checkClassLoaderLeak = config.getBoolean(CoreOptions.CHECK_LEAKED_CLASSLOADER);
final BlobLibraryCacheManager libraryCacheManager =
new BlobLibraryCacheManager(
blobServer,
BlobLibraryCacheManager.defaultClassLoaderFactory(
FlinkUserCodeClassLoaders.ResolveOrder.fromString(classLoaderResolveOrder),
alwaysParentFirstLoaderPatterns,
- failOnJvmMetaspaceOomError ? fatalErrorHandler : null));
+ failOnJvmMetaspaceOomError ? fatalErrorHandler : null,
+ checkClassLoaderLeak));
final Duration akkaTimeout;
try {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index ab5fb52..cf3e916 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -314,12 +314,15 @@ public class TaskManagerServices {
final boolean failOnJvmMetaspaceOomError =
taskManagerServicesConfiguration.getConfiguration().getBoolean(CoreOptions.FAIL_ON_USER_CLASS_LOADING_METASPACE_OOM);
+ final boolean checkClassLoaderLeak =
+ taskManagerServicesConfiguration.getConfiguration().getBoolean(CoreOptions.CHECK_LEAKED_CLASSLOADER);
final LibraryCacheManager libraryCacheManager = new BlobLibraryCacheManager(
permanentBlobService,
BlobLibraryCacheManager.defaultClassLoaderFactory(
taskManagerServicesConfiguration.getClassLoaderResolveOrder(),
taskManagerServicesConfiguration.getAlwaysParentFirstLoaderPatterns(),
- failOnJvmMetaspaceOomError ? fatalErrorHandler : null));
+ failOnJvmMetaspaceOomError ? fatalErrorHandler : null,
+ checkClassLoaderLeak));
return new TaskManagerServices(
unresolvedTaskManagerLocation,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
index 9b43a0a..38eabd0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
@@ -520,7 +520,8 @@ public class BlobLibraryCacheManagerTest extends TestLogger {
.defaultClassLoaderFactory(
FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST,
new String[0],
- null);
+ null,
+ true);
private TestingBlobLibraryCacheManagerBuilder() throws IOException {
final Configuration blobClientConfig = new Configuration();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
index eb8aec8..fa8a532 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheRecoveryITCase.java
@@ -80,7 +80,8 @@ public class BlobLibraryCacheRecoveryITCase extends TestLogger {
final BlobLibraryCacheManager.ClassLoaderFactory classLoaderFactory = BlobLibraryCacheManager.defaultClassLoaderFactory(
FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST,
new String[0],
- null);
+ null,
+ true);
for (int i = 0; i < server.length; i++) {
server[i] = new BlobServer(config, blobStoreService);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoadersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoadersTest.java
index 32a6f4b..0064b47 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoadersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/FlinkUserCodeClassLoadersTest.java
@@ -179,7 +179,8 @@ public class FlinkUserCodeClassLoadersTest extends TestLogger {
new URL[] { childCodePath },
parentClassLoader,
new String[] { parentFirstPattern },
- NOOP_EXCEPTION_HANDLER);
+ NOOP_EXCEPTION_HANDLER,
+ true);
final Class<?> clazz1 = Class.forName(className, false, parentClassLoader);
final Class<?> clazz2 = Class.forName(className, false, childClassLoader);
@@ -197,7 +198,8 @@ public class FlinkUserCodeClassLoadersTest extends TestLogger {
return FlinkUserCodeClassLoaders.parentFirst(
new URL[] { childCodePath },
parentClassLoader,
- NOOP_EXCEPTION_HANDLER);
+ NOOP_EXCEPTION_HANDLER,
+ true);
}
private static URLClassLoader createChildFirstClassLoader(URL childCodePath, ClassLoader parentClassLoader) {
@@ -205,7 +207,8 @@ public class FlinkUserCodeClassLoadersTest extends TestLogger {
new URL[] { childCodePath },
parentClassLoader,
new String[0],
- NOOP_EXCEPTION_HANDLER);
+ NOOP_EXCEPTION_HANDLER,
+ true);
}
@Test
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDbMultiClassLoaderTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDbMultiClassLoaderTest.java
index db16954..bc0f6fe 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDbMultiClassLoaderTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDbMultiClassLoaderTest.java
@@ -48,8 +48,18 @@ public class RocksDbMultiClassLoaderTest {
final URL codePath2 = RocksDB.class.getProtectionDomain().getCodeSource().getLocation();
final ClassLoader parent = getClass().getClassLoader();
- final ClassLoader loader1 = FlinkUserCodeClassLoaders.childFirst(new URL[] { codePath1, codePath2 }, parent, new String[0], NOOP_EXCEPTION_HANDLER);
- final ClassLoader loader2 = FlinkUserCodeClassLoaders.childFirst(new URL[] { codePath1, codePath2 }, parent, new String[0], NOOP_EXCEPTION_HANDLER);
+ final ClassLoader loader1 = FlinkUserCodeClassLoaders.childFirst(
+ new URL[]{codePath1, codePath2},
+ parent,
+ new String[0],
+ NOOP_EXCEPTION_HANDLER,
+ true);
+ final ClassLoader loader2 = FlinkUserCodeClassLoaders.childFirst(
+ new URL[]{codePath1, codePath2},
+ parent,
+ new String[0],
+ NOOP_EXCEPTION_HANDLER,
+ true);
final String className = RocksDBStateBackend.class.getName();
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
index d737746..3bc8884 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/ExecutionContextTest.java
@@ -383,14 +383,16 @@ public class ExecutionContextTest {
.parentFirst(
new URL[0],
TestClassLoaderCatalog.class.getClassLoader(),
- NOOP_EXCEPTION_HANDLER)
+ NOOP_EXCEPTION_HANDLER,
+ true)
.getClass();
private static final Class childFirstCL = FlinkUserCodeClassLoaders
.childFirst(
new URL[0],
TestClassLoaderCatalog.class.getClassLoader(),
new String[0],
- NOOP_EXCEPTION_HANDLER)
+ NOOP_EXCEPTION_HANDLER,
+ true)
.getClass();
TestClassLoaderCatalog(String name) {