[FLINK-8645][configuration] Split classloader.parent-first-patterns i…

…nto "base" and "append

This closes apache#5544.

docs/ops/config.md

@@ -75,12 +75,16 @@ without explicit scheme definition, such as `/user/USERNAME/in.txt`, is going to
 - `classloader.resolve-order`: Whether Flink should use a child-first `ClassLoader` when loading
 user-code classes or a parent-first `ClassLoader`. Can be one of `parent-first` or `child-first`. (default: `child-first`)
 
-- `classloader.parent-first-patterns`: A (semicolon-separated) list of patterns that specifies which
+- `classloader.parent-first-patterns.default`: A (semicolon-separated) list of patterns that specifies which
 classes should always be resolved through the parent `ClassLoader` first. A pattern is a simple
 prefix that is checked against the fully qualified class name. By default, this is set to
-`java.;org.apache.flink.;javax.annotation;org.slf4j;org.apache.log4j;org.apache.logging.log4j;ch.qos.logback`.
-If you want to change this setting you have to make sure to also include the default patterns in
-your list of patterns if you want to keep that default behaviour.
+`"java.;scala.;org.apache.flink.;com.esotericsoftware.kryo;org.apache.hadoop.;javax.annotation.;org.slf4j;org.apache.log4j;org.apache.logging.log4j;ch.qos.logback"`.
+To extend this list beyond the default it is recommended to configure `classloader.parent-first-patterns.additional` instead of modifying this setting directly.
+
+- `classloader.parent-first-patterns.additional`: A (semicolon-separated) list of patterns that specifies which
+classes should always be resolved through the parent `ClassLoader` first. A pattern is a simple
+prefix that is checked against the fully qualified class name.
+This list is appended to `classloader.parent-first-patterns.default`.
 
 ## Advanced Options
 

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -45,7 +45,7 @@ public class CoreOptions {
  * which means that user code jars can include and load different dependencies than
  * Flink uses (transitively).
  *
- * <p>Exceptions to the rules are defined via {@link #ALWAYS_PARENT_FIRST_LOADER}.
+ * <p>Exceptions to the rules are defined via {@link #ALWAYS_PARENT_FIRST_LOADER_PATTERNS}.
  */
  public static final ConfigOption<String> CLASSLOADER_RESOLVE_ORDER = ConfigOptions
  .key("classloader.resolve-order")
@@ -85,12 +85,41 @@ public class CoreOptions {
  * log bindings.</li>
  * </ul>
  */
- public static final ConfigOption<String> ALWAYS_PARENT_FIRST_LOADER = ConfigOptions
- .key("classloader.parent-first-patterns")
+ public static final ConfigOption<String> ALWAYS_PARENT_FIRST_LOADER_PATTERNS = ConfigOptions
+ .key("classloader.parent-first-patterns.default")
  .defaultValue("java.;scala.;org.apache.flink.;com.esotericsoftware.kryo;org.apache.hadoop.;javax.annotation.;org.slf4j;org.apache.log4j;org.apache.logging.log4j;ch.qos.logback")
+ .withDeprecatedKeys("classloader.parent-first-patterns")
  .withDescription("A (semicolon-separated) list of patterns that specifies which classes should always be" +
  " resolved through the parent ClassLoader first. A pattern is a simple prefix that is checked against" +
- " the fully qualified class name.");
+ " the fully qualified class name. This setting should generally not be modified. To add another pattern we" +
+ " recommend to use \"classloader.parent-first-patterns.additional\" instead.");
+
+ public static final ConfigOption<String> ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL = ConfigOptions
+ .key("classloader.parent-first-patterns.additional")
+ .defaultValue("")
+ .withDescription("A (semicolon-separated) list of patterns that specifies which classes should always be" +
+ " resolved through the parent ClassLoader first. A pattern is a simple prefix that is checked against" +
+ " the fully qualified class name. These patterns are appended to \"" + ALWAYS_PARENT_FIRST_LOADER_PATTERNS.key() + "\".");
+
+ public static String[] getParentFirstLoaderPatterns(Configuration config) {
+ String base = config.getString(ALWAYS_PARENT_FIRST_LOADER_PATTERNS);
+ String append = config.getString(ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL);
+
+ String[] basePatterns = base.isEmpty()
+ ? new String[0]
+ : base.split(";");
+
+ if (append.isEmpty()) {
+ return basePatterns;
+ } else {
+ String[] appendPatterns = append.split(";");
+
+ String[] joinedPatterns = new String[basePatterns.length + appendPatterns.length];
+ System.arraycopy(basePatterns, 0, joinedPatterns, 0, basePatterns.length);
+ System.arraycopy(appendPatterns, 0, joinedPatterns, basePatterns.length, appendPatterns.length);
+ return joinedPatterns;
+ }
+ }
 
  // ------------------------------------------------------------------------
  // process parameters

flink-core/src/test/java/org/apache/flink/configuration/CoreOptionsTest.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.configuration;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests for {@link CoreOptions}.
+ */
+public class CoreOptionsTest {
+ @Test
+ public void testGetParentFirstLoaderPatterns() {
+ Configuration config = new Configuration();
+
+ Assert.assertArrayEquals(
+ CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS.defaultValue().split(";"),
+ CoreOptions.getParentFirstLoaderPatterns(config));
+
+ config.setString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS, "hello;world");
+
+ Assert.assertArrayEquals(
+ "hello;world".split(";"),
+ CoreOptions.getParentFirstLoaderPatterns(config));
+
+ config.setString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL, "how;are;you");
+
+ Assert.assertArrayEquals(
+ "hello;world;how;are;you".split(";"),
+ CoreOptions.getParentFirstLoaderPatterns(config));
+
+ config.setString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS, "");
+
+ Assert.assertArrayEquals(
+ "how;are;you".split(";"),
+ CoreOptions.getParentFirstLoaderPatterns(config));
+ }
+}

flink-core/src/test/java/org/apache/flink/configuration/ParentFirstPatternsTest.java

@@ -34,7 +34,7 @@
 public class ParentFirstPatternsTest extends TestLogger {
 
  private static final HashSet<String> PARENT_FIRST_PACKAGES = new HashSet<>(
- Arrays.asList(CoreOptions.ALWAYS_PARENT_FIRST_LOADER.defaultValue().split(";")));
+ Arrays.asList(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS.defaultValue().split(";")));
 
  /**
  * All java and Flink classes must be loaded parent first.

flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroKryoClassloadingTest.java

@@ -73,7 +73,7 @@ public void testKryoInChildClasspath() throws Exception {
  final ClassLoader userAppClassLoader = FlinkUserCodeClassLoaders.childFirst(
  new URL[] { avroLocation, kryoLocation },
  parentClassLoader,
- CoreOptions.ALWAYS_PARENT_FIRST_LOADER.defaultValue().split(";"));
+ CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS.defaultValue().split(";"));
 
  final Class<?> userLoadedAvroClass = Class.forName(avroClass.getName(), false, userAppClassLoader);
  assertNotEquals(avroClass, userLoadedAvroClass);

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java

@@ -131,9 +131,7 @@ public static JobManagerSharedServices fromConfiguration(
  final String classLoaderResolveOrder =
  config.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER);
 
- final String alwaysParentFirstLoaderString =
- config.getString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER);
- final String[] alwaysParentFirstLoaderPatterns = alwaysParentFirstLoaderString.split(";");
+ final String[] alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(config);
 
  final BlobLibraryCacheManager libraryCacheManager =
  new BlobLibraryCacheManager(

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerConfiguration.java

@@ -241,9 +241,7 @@ public static TaskManagerConfiguration fromConfiguration(Configuration configura
  final String classLoaderResolveOrder =
  configuration.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER);
 
- final String alwaysParentFirstLoaderString =
- configuration.getString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER);
- final String[] alwaysParentFirstLoaderPatterns = alwaysParentFirstLoaderString.split(";");
+ final String[] alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(configuration);
 
  final String taskManagerLogPath = configuration.getString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, System.getProperty("log.file"));
  final String taskManagerStdoutPath;

flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala

@@ -2430,9 +2430,8 @@ object JobManager {
  val timeout: FiniteDuration = AkkaUtils.getTimeout(configuration)
 
  val classLoaderResolveOrder = configuration.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER)
- val alwaysParentFirstLoaderString =
- configuration.getString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER)
- val alwaysParentFirstLoaderPatterns = alwaysParentFirstLoaderString.split(';')
+
+ val alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(configuration)
 
  val restartStrategy = RestartStrategyFactory.createRestartStrategyFactory(configuration)