[FLINK-29814][statefun] Change supported Flink version to 1.15.2

galenwarren · tzulitai · commit b4a85ef031e5 · 2022-11-01T09:40:36.000-07:00
This closes #319.
diff --git a/README.md b/README.md
@@ -155,7 +155,7 @@ This section contains information for building this project.
 
 2. Build Stateful Functions Docker image: This step requires that you've already compiled artifacts from the source code.
   ```
-  $ ./tools/docker/build-distribution.sh
+  $ ./tools/docker/build-stateful-functions.sh
   ```
   This builds a local Docker image tagged as `flink-statefun:<version_of_current_source_version>`.
 
diff --git a/pom.xml b/pom.xml
@@ -79,11 +79,12 @@ under the License.
         <protobuf.version>3.7.1</protobuf.version>
         <unixsocket.version>2.3.2</unixsocket.version>
         <protoc-jar-maven-plugin.version>3.11.1</protoc-jar-maven-plugin.version>
-        <flink.version>1.14.3</flink.version>
+        <flink.version>1.15.2</flink.version>
         <scala.binary.version>2.12</scala.binary.version>
         <scala.version>2.12.7</scala.version>
         <lz4-java.version>1.8.0</lz4-java.version>
-        <flink-shaded-jackson.version>2.12.4-14.0</flink-shaded-jackson.version>
+        <flink-shaded-jackson.version>2.12.4-15.0</flink-shaded-jackson.version>
+        <slf4j-log4j12.version>1.7.32</slf4j-log4j12.version>
         <test.unit.pattern>**/*Test.*</test.unit.pattern>
     </properties>
 
@@ -109,6 +110,27 @@ under the License.
                 <version>1.3</version>
                 <scope>test</scope>
             </dependency>
+            <!--
+                Resolve dependency convergence issue:
+                flink-core:1.15.2 depends on kryo:2.24.0
+                flink-java:1.15.2 depends on kryo:2.21 (via com.twitter:chill-java:0.7.6)
+             -->
+            <dependency>
+                <groupId>com.esotericsoftware.kryo</groupId>
+                <artifactId>kryo</artifactId>
+                <version>2.24.0</version>
+            </dependency>
+            <!--
+                Resolve dependency convergence issue:
+                flink-connector-kinesis:1.15.2 depends on jackson-databind:2.13.2.2
+                flink-connector-kinesis:1.15.2 depends on jackson-databind:2.13.2
+                (via com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.13.2)
+             -->
+            <dependency>
+                <groupId>com.fasterxml.jackson.core</groupId>
+                <artifactId>jackson-databind</artifactId>
+                <version>2.13.2.2</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-common/pom.xml b/statefun-e2e-tests/statefun-smoke-e2e-common/pom.xml
@@ -70,7 +70,7 @@ under the License.
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>
-            <version>1.7.15</version>
+            <version>${slf4j-log4j12.version}</version>
         </dependency>
         <dependency>
             <groupId>log4j</groupId>
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-driver/pom.xml b/statefun-e2e-tests/statefun-smoke-e2e-driver/pom.xml
@@ -28,7 +28,7 @@ under the License.
     <artifactId>statefun-smoke-e2e-driver</artifactId>
 
     <properties>
-        <commons-math3.version>3.5</commons-math3.version>
+        <commons-math3.version>3.6.1</commons-math3.version>
         <additional-sources.dir>target/additional-sources</additional-sources.dir>
     </properties>
 
@@ -72,7 +72,7 @@ under the License.
         <!-- streaming runtime -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
             <exclusions>
                 <!--
diff --git a/statefun-flink/pom.xml b/statefun-flink/pom.xml
@@ -54,7 +54,7 @@ under the License.
             <!-- Flink -->
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+                <artifactId>flink-streaming-java</artifactId>
                 <version>${flink.version}</version>
                 <exclusions>
                     <!--
@@ -69,7 +69,7 @@ under the License.
             </dependency>
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+                <artifactId>flink-connector-kafka</artifactId>
                 <version>${flink.version}</version>
                 <exclusions>
                     <!--
diff --git a/statefun-flink/statefun-flink-core/pom.xml b/statefun-flink/statefun-flink-core/pom.xml
@@ -32,7 +32,7 @@ under the License.
     <properties>
         <okhttp.version>3.14.6</okhttp.version>
         <additional-sources.dir>target/additional-sources</additional-sources.dir>
-        <flink-shaded-netty.version>4.1.65.Final-14.0</flink-shaded-netty.version>
+        <flink-shaded-netty.version>4.1.70.Final-15.0</flink-shaded-netty.version>
     </properties>
 
     <dependencies>
@@ -67,7 +67,7 @@ under the License.
         <!-- flink runtime -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <scope>provided</scope>
         </dependency>
         <dependency>
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
@@ -57,10 +57,10 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu
   }
 
   private static Set<String> parentFirstClassloaderPatterns(Configuration configuration) {
-    final String[] split =
-        configuration.get(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL).split(";");
-    final Set<String> parentFirstClassloaderPatterns = new HashSet<>(split.length);
-    for (String s : split) {
+    final List<String> patterns =
+        configuration.get(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL);
+    final Set<String> parentFirstClassloaderPatterns = new HashSet<>(patterns.size());
+    for (String s : patterns) {
       parentFirstClassloaderPatterns.add(s.trim().toLowerCase(Locale.ENGLISH));
     }
     return parentFirstClassloaderPatterns;
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
@@ -17,6 +17,7 @@
  */
 package org.apache.flink.statefun.flink.core;
 
+import java.util.Arrays;
 import java.util.Optional;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
@@ -48,7 +49,7 @@ public void testSetConfigurations() {
     configuration.set(StatefulFunctionsConfig.ASYNC_MAX_OPERATIONS_PER_TASK, 100);
     configuration.set(
         CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
-        "org.apache.flink.statefun;org.apache.kafka;com.google.protobuf");
+        Arrays.asList("org.apache.flink.statefun", "org.apache.kafka", "com.google.protobuf"));
     configuration.set(ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, 1);
     configuration.setString("statefun.module.global-config.key1", "value1");
     configuration.setString("statefun.module.global-config.key2", "value2");
@@ -81,7 +82,7 @@ private static Configuration baseConfiguration() {
     configuration.set(StatefulFunctionsConfig.ASYNC_MAX_OPERATIONS_PER_TASK, 100);
     configuration.set(
         CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
-        "org.apache.flink.statefun;org.apache.kafka;com.google.protobuf");
+        Arrays.asList("org.apache.flink.statefun", "org.apache.kafka", "com.google.protobuf"));
     configuration.set(ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, 1);
     return configuration;
   }
diff --git a/statefun-flink/statefun-flink-datastream/pom.xml b/statefun-flink/statefun-flink-datastream/pom.xml
@@ -73,7 +73,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
        </dependency>
diff --git a/statefun-flink/statefun-flink-distribution/pom.xml b/statefun-flink/statefun-flink-distribution/pom.xml
@@ -34,7 +34,7 @@ under the License.
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>
-            <version>1.7.15</version>
+            <version>${slf4j-log4j12.version}</version>
         </dependency>
         <dependency>
             <groupId>log4j</groupId>
@@ -97,7 +97,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
@@ -115,7 +115,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <artifactId>flink-clients</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
diff --git a/statefun-flink/statefun-flink-harness/pom.xml b/statefun-flink/statefun-flink-harness/pom.xml
@@ -37,7 +37,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
         </dependency>
     </dependencies>
@@ -58,13 +58,13 @@ under the License.
             <dependencies>
                 <dependency>
                     <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+                    <artifactId>flink-streaming-java</artifactId>
                     <version>${flink.version}</version>
                     <scope>compile</scope>
                 </dependency>
                 <dependency>
                     <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
+                    <artifactId>flink-runtime-web</artifactId>
                     <version>${flink.version}</version>
                     <scope>compile</scope>
                 </dependency>
diff --git a/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java b/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java
@@ -189,7 +189,7 @@ public void accept(T t) {
   private static void configureStrictlyRequiredFlinkConfigs(Configuration flinkConfig) {
     flinkConfig.set(
         CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
-        String.join(";", StatefulFunctionsConfigValidator.PARENT_FIRST_CLASSLOADER_PATTERNS));
+        StatefulFunctionsConfigValidator.PARENT_FIRST_CLASSLOADER_PATTERNS);
     flinkConfig.set(
         ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS,
         StatefulFunctionsConfigValidator.MAX_CONCURRENT_CHECKPOINTS);
diff --git a/statefun-flink/statefun-flink-io-bundle/pom.xml b/statefun-flink/statefun-flink-io-bundle/pom.xml
@@ -75,21 +75,21 @@ under the License.
         <!-- flink  -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
         <!-- flink kafka connector -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+            <artifactId>flink-connector-kafka</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
         <!-- flink kinesis connector -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis_${scala.binary.version}</artifactId>
+            <artifactId>flink-connector-kinesis</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
diff --git a/statefun-flink/statefun-flink-io/pom.xml b/statefun-flink/statefun-flink-io/pom.xml
@@ -51,7 +51,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
         </dependency>
         <dependency>
diff --git a/statefun-flink/statefun-flink-launcher/pom.xml b/statefun-flink/statefun-flink-launcher/pom.xml
@@ -38,7 +38,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <artifactId>flink-clients</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
diff --git a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java
@@ -19,6 +19,7 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.util.List;
 import java.util.Optional;
 import javax.annotation.Nonnull;
 import org.apache.flink.annotation.VisibleForTesting;
@@ -129,17 +130,12 @@ private static boolean isNoExecutionModeConfigured(Configuration configuration)
   }
 
   private static void addStatefulFunctionsConfiguration(Configuration configuration) {
-    String parentFirst =
-        configuration.getString(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL, "");
-    if (parentFirst.isEmpty()) {
-      parentFirst = Constants.STATEFUL_FUNCTIONS_PACKAGE;
-    } else if (parentFirst.endsWith(";")) {
-      parentFirst = parentFirst + Constants.STATEFUL_FUNCTIONS_PACKAGE;
-    } else {
-      parentFirst = parentFirst + ";" + Constants.STATEFUL_FUNCTIONS_PACKAGE;
+    List<String> patterns =
+        configuration.get(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL);
+    if (!patterns.contains(Constants.STATEFUL_FUNCTIONS_PACKAGE)) {
+      patterns.add(Constants.STATEFUL_FUNCTIONS_PACKAGE);
     }
-    configuration.setString(
-        CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL, parentFirst);
+    configuration.set(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL, patterns);
   }
 
   @Override
diff --git a/statefun-flink/statefun-flink-state-processor/pom.xml b/statefun-flink/statefun-flink-state-processor/pom.xml
@@ -43,7 +43,7 @@ under the License.
         <!-- Apache Flink dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-state-processor-api_${scala.binary.version}</artifactId>
+            <artifactId>flink-state-processor-api</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
@@ -55,7 +55,7 @@ under the License.
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
+            <artifactId>flink-statebackend-rocksdb</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
@@ -74,7 +74,7 @@ under the License.
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <artifactId>flink-clients</artifactId>
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>
diff --git a/tools/docker/Dockerfile b/tools/docker/Dockerfile
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM apache/flink:1.14.3-scala_2.12-java8
+FROM apache/flink:1.15.2-scala_2.12-java8
 
 ENV ROLE worker
 ENV MASTER_HOST localhost

Original file line number	Diff line number	Diff line change
`@@ -57,10 +57,10 @@ private static void validateParentFirstClassloaderPatterns(Configuration configu`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`private static Set<String> parentFirstClassloaderPatterns(Configuration configuration) {`
`60`		`- final String[] split =`
`61`		`- configuration.get(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL).split(";");`
`62`		`- final Set<String> parentFirstClassloaderPatterns = new HashSet<>(split.length);`
`63`		`- for (String s : split) {`
	`60`	`+ final List<String> patterns =`
	`61`	`+ configuration.get(CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL);`
	`62`	`+ final Set<String> parentFirstClassloaderPatterns = new HashSet<>(patterns.size());`
	`63`	`+ for (String s : patterns) {`
`64`	`64`	`parentFirstClassloaderPatterns.add(s.trim().toLowerCase(Locale.ENGLISH));`
`65`	`65`	`}`
`66`	`66`	`return parentFirstClassloaderPatterns;`