[FLINK-18783] Load Akka with separate classloader

lovepoem · Jul 13, 2021 · e341e47 · e341e47
1 parent 1aa7d16
commit e341e47
Show file tree

Hide file tree

Showing 99 changed files with 1,513 additions and 392 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,15 +149,15 @@ 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>
 		</dependency>
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-metrics-jmx_${scala.binary.version}</artifactId>
+			<artifactId>flink-metrics-jmx</artifactId>
 			<version>${project.version}</version>
 			<scope>test</scope>
 		</dependency>

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://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
@@ -1319,13 +1319,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>
 
@@ -254,21 +254,21 @@ under the License.
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-metrics-influxdb_${scala.binary.version}</artifactId>
+			<artifactId>flink-metrics-influxdb</artifactId>
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</dependency>
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-metrics-prometheus_${scala.binary.version}</artifactId>
+			<artifactId>flink-metrics-prometheus</artifactId>
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</dependency>
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-metrics-jmx_${scala.binary.version}</artifactId>
+			<artifactId>flink-metrics-jmx</artifactId>
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</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>