[FLINK-18783] Load Akka with separate classloader

tongcheng-elong · Jul 21, 2021 · b34ef1e · b34ef1e
1 parent f9286a5
commit b34ef1e
Show file tree

Hide file tree

Showing 94 changed files with 1,491 additions and 370 deletions.
diff --git a/docs/content.zh/docs/deployment/resource-providers/standalone/kubernetes.md b/docs/content.zh/docs/deployment/resource-providers/standalone/kubernetes.md
@@ -310,7 +310,7 @@ data:
  appender.rolling.strategy.max = 10
 
  # Suppress the irrelevant (wrong) warnings from the Netty channel handler
- logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+ logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
  logger.netty.level = OFF
 ```
 
@@ -380,7 +380,7 @@ data:
  appender.rolling.strategy.max = 10
 
  # Suppress the irrelevant (wrong) warnings from the Netty channel handler
- logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+ logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
  logger.netty.level = OFF
 ```
 

diff --git a/docs/content/docs/deployment/resource-providers/standalone/kubernetes.md b/docs/content/docs/deployment/resource-providers/standalone/kubernetes.md
@@ -310,7 +310,7 @@ data:
  appender.rolling.strategy.max = 10
 
  # Suppress the irrelevant (wrong) warnings from the Netty channel handler
- logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+ logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
  logger.netty.level = OFF
 ```
 
@@ -380,7 +380,7 @@ data:
  appender.rolling.strategy.max = 10
 
  # Suppress the irrelevant (wrong) warnings from the Netty channel handler
- logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+ logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
  logger.netty.level = OFF
 ```
 

diff --git a/flink-clients/pom.xml b/flink-clients/pom.xml
@@ -47,13 +47,13 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  </dependency>
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-optimizer_${scala.binary.version}</artifactId>
+ <artifactId>flink-optimizer</artifactId>
  <version>${project.version}</version>
  </dependency>
 
@@ -77,7 +77,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  <type>test-jar</type>

diff --git a/flink-connectors/flink-connector-base/pom.xml b/flink-connectors/flink-connector-base/pom.xml
@@ -68,7 +68,7 @@
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-cassandra/pom.xml b/flink-connectors/flink-connector-cassandra/pom.xml
@@ -183,7 +183,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  <type>test-jar</type>

diff --git a/flink-connectors/flink-connector-elasticsearch-base/pom.xml b/flink-connectors/flink-connector-elasticsearch-base/pom.xml
@@ -94,7 +94,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>

diff --git a/flink-connectors/flink-connector-elasticsearch5/pom.xml b/flink-connectors/flink-connector-elasticsearch5/pom.xml
@@ -135,7 +135,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-elasticsearch6/pom.xml b/flink-connectors/flink-connector-elasticsearch6/pom.xml
@@ -112,7 +112,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-elasticsearch7/pom.xml b/flink-connectors/flink-connector-elasticsearch7/pom.xml
@@ -110,7 +110,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-files/pom.xml b/flink-connectors/flink-connector-files/pom.xml
@@ -107,7 +107,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-gcp-pubsub/pom.xml b/flink-connectors/flink-connector-gcp-pubsub/pom.xml
@@ -88,7 +88,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>

diff --git a/flink-connectors/flink-connector-hbase-1.4/pom.xml b/flink-connectors/flink-connector-hbase-1.4/pom.xml
@@ -348,7 +348,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-hive/pom.xml b/flink-connectors/flink-connector-hive/pom.xml
@@ -867,7 +867,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-jdbc/pom.xml b/flink-connectors/flink-connector-jdbc/pom.xml
@@ -72,7 +72,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-kafka/pom.xml b/flink-connectors/flink-connector-kafka/pom.xml
@@ -99,7 +99,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>
@@ -149,7 +149,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>

diff --git a/flink-connectors/flink-connector-kinesis/pom.xml b/flink-connectors/flink-connector-kinesis/pom.xml
@@ -180,14 +180,14 @@ under the License.
  </dependency>
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>
  </dependency>
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>

diff --git a/flink-connectors/flink-connector-rabbitmq/pom.xml b/flink-connectors/flink-connector-rabbitmq/pom.xml
@@ -65,14 +65,14 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-statebackend-changelog_${scala.binary.version}</artifactId>
+ <artifactId>flink-statebackend-changelog</artifactId>
  <version>${project.version}</version>
  <scope>test</scope>
  </dependency>
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <type>test-jar</type>
  <scope>test</scope>

diff --git a/flink-container/pom.xml b/flink-container/pom.xml
@@ -39,7 +39,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  <scope>provided</scope>
  </dependency>

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
@@ -183,7 +183,8 @@ public static String[] getPluginParentFirstLoaderPatterns(Configuration config)
  return parseParentFirstLoaderPatterns(base, append);
  }
 
- private static String[] parseParentFirstLoaderPatterns(String base, String append) {
+ @Internal
+ public static String[] parseParentFirstLoaderPatterns(String base, String append) {
  Splitter splitter = Splitter.on(';').omitEmptyStrings();
  return Iterables.toArray(
  Iterables.concat(splitter.split(base), splitter.split(append)), String.class);

diff --git a/flink-core/src/main/java/org/apache/flink/core/classloading/SubmoduleClassLoader.java b/flink-core/src/main/java/org/apache/flink/core/classloading/SubmoduleClassLoader.java
@@ -0,0 +1,45 @@
+/*
+ * 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
+ *
+ * http: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.core.classloading;
+
+import org.apache.flink.configuration.CoreOptions;
+
+import java.net.URL;
+
+/**
+ * Loads all classes from the submodule jar, except for explicitly white-listed packages.
+ *
+ * <p>To ensure that classes from the submodule are always loaded through the submodule classloader
+ * (and thus from the submodule jar), even if the classes are also on the classpath (e.g., during
+ * tests), all classes from the "org.apache.flink" package are loaded child-first.
+ *
+ * <p>Classes related to logging (e.g., log4j) are loaded parent-first.
+ *
+ * <p>All other classes can only be loaded if they are either available in the submodule jar or the
+ * bootstrap/app classloader (i.e., provided by the JDK).
+ */
+public class SubmoduleClassLoader extends ComponentClassLoader {
+ public SubmoduleClassLoader(URL[] classpath, ClassLoader parentClassLoader) {
+ super(
+ classpath,
+ parentClassLoader,
+ CoreOptions.parseParentFirstLoaderPatterns(
+ CoreOptions.PARENT_FIRST_LOGGING_PATTERNS, ""),
+ new String[] {"org.apache.flink"});
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java b/flink-core/src/main/java/org/apache/flink/util/concurrent/FutureUtils.java
@@ -1369,13 +1369,25 @@ public static void throwIfCompletedExceptionally(CompletableFuture<?> future) th
  }
 
  private static <T> BiConsumer<T, Throwable> forwardTo(CompletableFuture<T> target) {
- return (value, throwable) -> {
- if (throwable != null) {
- target.completeExceptionally(throwable);
- } else {
- target.complete(value);
- }
- };
+ return (value, throwable) -> doForward(value, throwable, target);
+ }
+
+ /**
+ * Completes the given future with either the given value or throwable, depending on which
+ * parameter is not null.
+ *
+ * @param value value with which the future should be completed
+ * @param throwable throwable with which the future should be completed exceptionally
+ * @param target future to complete
+ * @param <T> completed future
+ */
+ public static <T> void doForward(
+ @Nullable T value, @Nullable Throwable throwable, CompletableFuture<T> target) {
+ if (throwable != null) {
+ target.completeExceptionally(throwable);
+ } else {
+ target.complete(value);
+ }
  }
 
  /**

diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
@@ -61,7 +61,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-runtime</artifactId>
  <version>${project.version}</version>
  </dependency>
 
@@ -73,7 +73,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-optimizer_${scala.binary.version}</artifactId>
+ <artifactId>flink-optimizer</artifactId>
  <version>${project.version}</version>
  </dependency>
 
@@ -376,7 +376,7 @@ under the License.
 
  <dependency>
  <groupId>org.apache.flink</groupId>
- <artifactId>flink-queryable-state-runtime_${scala.binary.version}</artifactId>
+ <artifactId>flink-queryable-state-runtime</artifactId>
  <version>${project.version}</version>
  <scope>provided</scope>
  </dependency>
@@ -688,11 +688,6 @@ under the License.
  <exclude>org.apache.flink:flink-shaded-zookeeper-3</exclude>
  </excludes>
  </artifactSet>
- <transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
- <resource>reference.conf</resource>
- </transformer>
- </transformers>
  </configuration>
  </execution>
  <execution>

diff --git a/flink-dist/src/main/assemblies/opt.xml b/flink-dist/src/main/assemblies/opt.xml
@@ -105,9 +105,9 @@
 
  <!-- Queryable State -->
  <file>
- <source>../flink-queryable-state/flink-queryable-state-runtime/target/flink-queryable-state-runtime_${scala.binary.version}-${project.version}.jar</source>
+ <source>../flink-queryable-state/flink-queryable-state-runtime/target/flink-queryable-state-runtime-${project.version}.jar</source>
  <outputDirectory>opt/</outputDirectory>
- <destName>flink-queryable-state-runtime_${scala.binary.version}-${project.version}.jar</destName>
+ <destName>flink-queryable-state-runtime-${project.version}.jar</destName>
  <fileMode>0644</fileMode>
  </file>
 

diff --git a/flink-dist/src/main/flink-bin/conf/log4j-cli.properties b/flink-dist/src/main/flink-bin/conf/log4j-cli.properties
@@ -63,5 +63,5 @@ logger.hadoopnative.name = org.apache.hadoop.util.NativeCodeLoader
 logger.hadoopnative.level = OFF
 
 # Suppress the irrelevant (wrong) warnings from the Netty channel handler
-logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
 logger.netty.level = OFF
diff --git a/flink-dist/src/main/flink-bin/conf/log4j-console.properties b/flink-dist/src/main/flink-bin/conf/log4j-console.properties
@@ -64,5 +64,5 @@ appender.rolling.strategy.type = DefaultRolloverStrategy
 appender.rolling.strategy.max = ${env:MAX_LOG_FILE_NUMBER:-10}
 
 # Suppress the irrelevant (wrong) warnings from the Netty channel handler
-logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
 logger.netty.level = OFF
diff --git a/flink-dist/src/main/flink-bin/conf/log4j-session.properties b/flink-dist/src/main/flink-bin/conf/log4j-session.properties
@@ -28,7 +28,7 @@ appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
 
 # Suppress the irrelevant (wrong) warnings from the Netty channel handler
-logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
 logger.netty.level = OFF
 logger.zookeeper.name = org.apache.zookeeper
 logger.zookeeper.level = WARN

diff --git a/flink-dist/src/main/flink-bin/conf/log4j.properties b/flink-dist/src/main/flink-bin/conf/log4j.properties
@@ -57,5 +57,5 @@ appender.main.strategy.type = DefaultRolloverStrategy
 appender.main.strategy.max = ${env:MAX_LOG_FILE_NUMBER:-10}
 
 # Suppress the irrelevant (wrong) warnings from the Netty channel handler
-logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
 logger.netty.level = OFF
diff --git a/flink-dist/src/main/flink-bin/conf/logback-console.xml b/flink-dist/src/main/flink-bin/conf/logback-console.xml
@@ -60,5 +60,5 @@
  <logger name="org.apache.zookeeper" level="INFO"/>
 
  <!-- Suppress the irrelevant (wrong) warnings from the Netty channel handler -->
- <logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR"/>
+ <logger name="org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR"/>
 </configuration>