Support Flink 1.17.1

pom.xml

@@ -76,15 +76,28 @@ under the License.
         <java.version>1.8</java.version>
         <spotless-maven-plugin.version>1.20.0</spotless-maven-plugin.version>
         <auto-service.version>1.0-rc6</auto-service.version>
-        <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.15.2</flink.version>
+        <protobuf.version>3.23.2</protobuf.version>
+        <unixsocket.version>2.6.2</unixsocket.version>
+        <protoc-jar-maven-plugin.version>3.11.4</protoc-jar-maven-plugin.version>
+        <flink.version>1.17.1</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-15.0</flink-shaded-jackson.version>
+        <flink-shaded-jackson.version>2.14.2-17.0</flink-shaded-jackson.version>
         <slf4j-log4j12.version>1.7.32</slf4j-log4j12.version>
+        <flink-connector-kinesis.version>4.1.0-1.17</flink-connector-kinesis.version>
+        <flink-connector-aws-kinesis-streams.version>4.1.0-1.17</flink-connector-aws-kinesis-streams.version>
+        <okhttp.version>3.14.6</okhttp.version>
+        <flink-shaded-netty.version>4.1.82.Final-16.1</flink-shaded-netty.version>
+        <junit.version>4.12</junit.version>
+        <hamcrest-all.version>1.3</hamcrest-all.version>
+        <kryo.version>2.24.0</kryo.version>
+        <jackson-databind.version>2.13.2.2</jackson-databind.version>
+        <flink-shaded-netty.version>4.1.82.Final-16.1</flink-shaded-netty.version>
+        <flink-shaded-force-shading.version>16.1</flink-shaded-force-shading.version>
+        <commons-codec.version>1.15</commons-codec.version>
+        <commons-logging.version>1.2</commons-logging.version>
+        <slf4j-api.version>1.7.36</slf4j-api.version>
         <test.unit.pattern>**/*Test.*</test.unit.pattern>
     </properties>
 
@@ -101,15 +114,20 @@ under the License.
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>
-                <version>4.12</version>
+                <version>${junit.version}</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
                 <groupId>org.hamcrest</groupId>
                 <artifactId>hamcrest-all</artifactId>
-                <version>1.3</version>
+                <version>${hamcrest-all.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+              <groupId>com.google.protobuf</groupId>
+              <artifactId>protobuf-java</artifactId>
+              <version>${protobuf.version}</version>
+            </dependency>
             <!--
                 Resolve dependency convergence issue:
                 flink-core:1.15.2 depends on kryo:2.24.0
@@ -118,7 +136,7 @@ under the License.
             <dependency>
                 <groupId>com.esotericsoftware.kryo</groupId>
                 <artifactId>kryo</artifactId>
-                <version>2.24.0</version>
+                <version>${kryo.version}</version>
             </dependency>
             <!--
                 Resolve dependency convergence issue:
@@ -129,7 +147,66 @@ under the License.
             <dependency>
                 <groupId>com.fasterxml.jackson.core</groupId>
                 <artifactId>jackson-databind</artifactId>
-                <version>2.13.2.2</version>
+                <version>${jackson-databind.version}</version>
+            </dependency>
+            <!--
+                Resolve dependency convergence issue:
+                org.apache.flink:flink-streaming-java:1.17.1 depends on org.apache.flink:flink-shaded-netty:4.1.82.Final-16.1
+                org.apache.flink:statefun-flink-core:3.4-SNAPSHOT depends on org.apache.flink:flink-shaded-netty:4.1.70.Final-15.0
+                (via com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.13.2)
+             -->
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-shaded-netty</artifactId>
+                <version>${flink-shaded-netty.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-core</artifactId>
+                <version>${flink.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-table-common</artifactId>
+                <version>${flink.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-connector-base</artifactId>
+                <version>${flink.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-shaded-force-shading</artifactId>
+                <version>${flink-shaded-force-shading.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
+                <artifactId>flink-shaded-jackson</artifactId>
+                <version>${flink-shaded-jackson.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>commons-codec</groupId>
+                <artifactId>commons-codec</artifactId>
+                <version>${commons-codec.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>commons-logging</groupId>
+                <artifactId>commons-logging</artifactId>
+                <version>${commons-logging.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>slf4j-api</artifactId>
+                <version>${slf4j-api.version}</version>
             </dependency>
         </dependencies>
     </dependencyManagement>
@@ -234,10 +311,12 @@ under the License.
                                     <outputTarget>
                                         <type>descriptor</type>
                                         <outputDirectory>${basedir}/target/test-classes</outputDirectory>
+                                        <addSources>main</addSources>
                                     </outputTarget>
                                     <outputTarget>
                                         <type>java</type>
                                         <outputDirectory>${basedir}/target/generated-test-sources/protoc-jar</outputDirectory>
+                                        <addSources>test</addSources>
                                     </outputTarget>
                                 </outputTargets>
                             </configuration>
@@ -412,4 +491,3 @@ under the License.
         </plugins>
     </build>
 </project>
-

statefun-flink/statefun-flink-core/pom.xml

@@ -30,9 +30,7 @@ under the License.
     <artifactId>statefun-flink-core</artifactId>
 
     <properties>
-        <okhttp.version>3.14.6</okhttp.version>
         <additional-sources.dir>target/additional-sources</additional-sources.dir>
-        <flink-shaded-netty.version>4.1.70.Final-15.0</flink-shaded-netty.version>
     </properties>
 
     <dependencies>
@@ -90,6 +88,7 @@ under the License.
             <groupId>com.kohlschutter.junixsocket</groupId>
             <artifactId>junixsocket-core</artifactId>
             <version>${unixsocket.version}</version>
+            <type>pom</type>
         </dependency>
 
         <dependency>

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java

@@ -265,7 +265,7 @@ public void setEmbedded(boolean embedded) {
    */
   public StatefulFunctionsUniverseProvider getProvider(ClassLoader cl) {
     try {
-      return InstantiationUtil.deserializeObject(universeInitializerClassBytes, cl, false);
+      return InstantiationUtil.deserializeObject(universeInitializerClassBytes, cl);
     } catch (IOException | ClassNotFoundException e) {
       throw new IllegalStateException("Unable to initialize.", e);
     }

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java

@@ -23,12 +23,12 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
 import org.apache.flink.statefun.flink.core.feedback.FeedbackKey;
 import org.apache.flink.statefun.flink.core.message.Message;
 import org.apache.flink.statefun.flink.core.translation.FlinkUniverse;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.util.FlinkUserCodeClassLoader;
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
 
 public class StatefulFunctionsJob {
 

statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/functions/ReductionsTest.java

@@ -352,10 +352,11 @@ public boolean deregisterKeySelectionListener(KeySelectionListener<Object> liste
 
     @Nonnull
     @Override
-    public <N, SV, SEV, S extends State, IS extends S> IS createInternalState(
+    public <N, SV, SEV, S extends State, IS extends S> IS createOrUpdateInternalState(
         @Nonnull TypeSerializer<N> namespaceSerializer,
         @Nonnull StateDescriptor<S, SV> stateDesc,
-        @Nonnull StateSnapshotTransformFactory<SEV> snapshotTransformFactory) {
+        @Nonnull StateSnapshotTransformFactory<SEV> snapshotTransformFactory)
+        throws Exception {
       throw new UnsupportedOperationException();
     }
 

statefun-flink/statefun-flink-datastream/pom.xml

@@ -51,13 +51,13 @@ under the License.
             <artifactId>statefun-flink-io</artifactId>
             <version>${project.version}</version>
         </dependency>
-  
-        <!-- The following dependencies are here with scope provided, because: 
+
+        <!-- The following dependencies are here with scope provided, because:
              a) they are transitively required by the statefun-flink-* dependencies
-             b) they are provided at runtime, by the embedding application. 
-             
-             Also note that org.slf4j:slf4j-api is excluded from all the artifacts, since maven 
-             convergence plugging fails. 
+             b) they are provided at runtime, by the embedding application.
+
+             Also note that org.slf4j:slf4j-api is excluded from all the artifacts, since maven
+             convergence plugging fails.
            -->
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -87,7 +87,7 @@ under the License.
             <groupId>org.hamcrest</groupId>
             <artifactId>hamcrest-all</artifactId>
         </dependency>
-        
+
     </dependencies>
 
     <build>
@@ -152,4 +152,4 @@ under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

statefun-flink/statefun-flink-distribution/pom.xml

@@ -74,7 +74,7 @@ under the License.
             <artifactId>statefun-flink-launcher</artifactId>
             <version>${project.version}</version>
         </dependency>
-        
+
         <!-- flink runtime metrics -->
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -199,4 +199,4 @@ under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

statefun-flink/statefun-flink-io-bundle/pom.xml

@@ -90,7 +90,12 @@ under the License.
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kinesis</artifactId>
-            <version>${flink.version}</version>
+            <version>${flink-connector-kinesis.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-aws-kinesis-streams</artifactId>
+            <version>${flink-connector-aws-kinesis-streams.version}</version>
         </dependency>
 
         <!-- 3rd party -->
@@ -110,4 +115,4 @@ under the License.
         </dependency>
     </dependencies>
 
-</project>
+</project>

statefun-flink/statefun-flink-io/pom.xml

@@ -124,4 +124,4 @@ under the License.
     </build>
 
 
-</project>
+</project>

statefun-shaded/statefun-protobuf-shaded/pom.xml

@@ -32,7 +32,6 @@ under the License.
 
     <properties>
         <protobuf-sources.dir>${generated-sources.basedir}/shaded-protobuf-java/</protobuf-sources.dir>
-        <protobuf.version>3.7.1</protobuf.version>
     </properties>
 
     <dependencies>
@@ -107,4 +106,4 @@ under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

statefun-shaded/statefun-protocol-shaded/pom.xml

@@ -33,7 +33,7 @@ under the License.
     <properties>
         <protocol-messages.dir>${generated-sources.basedir}/protocol-messages/</protocol-messages.dir>
         <proto-sources.dir>target/proto-sources</proto-sources.dir>
-        <protoc-jar-maven-plugin.version>3.11.1</protoc-jar-maven-plugin.version>
+        <protoc-jar-maven-plugin.version>3.11.4</protoc-jar-maven-plugin.version>
     </properties>
 
     <dependencies>
@@ -47,6 +47,11 @@ under the License.
             <artifactId>statefun-protobuf-shaded</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+            <version>${protobuf.version}</version>
+        </dependency>
     </dependencies>
 
     <build>
@@ -130,4 +135,4 @@ under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

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.15.2-scala_2.12-java8
+FROM flink:1.17.1-scala_2.12-java11
 
 ENV ROLE worker
 ENV MASTER_HOST localhost
@@ -45,7 +45,7 @@ RUN mkdir -p $FLINK_HOME/plugins/azure-fs-hadoop && \
 # add tcnative
 RUN mv $FLINK_HOME/opt/flink-shaded-netty-tcnative-dynamic-*.jar  $FLINK_HOME/lib/
 
-# entry point 
+# entry point
 ADD docker-entry-point.sh /docker-entry-point.sh
 
 ENTRYPOINT ["/docker-entry-point.sh"]