[FLINK-31619] Upgrade Stateful Functions to Flink 1.16.2. This closes #331

* [FLINK-31619] Upgrade Stateful Functions to Flink 1.16.1

Upgrade Flink version to 1.16.2 (this has been released since the issue was created) and update dependency-confict resolutions.
Handle org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders moving to org.apache.flink.util.FlinkUserCodeClassLoaders.
Fix ReductionsTest.java.
Modify construction of Netty request headers

Author: galenwarren

pom.xml

@@ -79,7 +79,7 @@ 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.15.2</flink.version>
+        <flink.version>1.16.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>
@@ -110,26 +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)
-             -->
+                flink-core:1.16.2 depends on kryo:2.24.0
+                flink-java:1.16.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)
+                flink-connector-kinesis:1.16.2 depends on jackson-databind:2.13.4.2
+                flink-connector-kinesis:1.16.2 depends on jackson-databind:2.13.4
+                (via com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.13.4)
              -->
             <dependency>
                 <groupId>com.fasterxml.jackson.core</groupId>
                 <artifactId>jackson-databind</artifactId>
-                <version>2.13.2.2</version>
+                <version>2.13.4.2</version>
             </dependency>
         </dependencies>
     </dependencyManagement>

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/main/java/org/apache/flink/statefun/flink/core/nettyclient/NettyRequestReplyHandler.java

@@ -132,7 +132,7 @@ private DefaultHttpHeaders headers(NettyRequest req, ByteBuf bodyBuf) {
     if (cachedHeaders != null) {
       headers = cachedHeaders;
     } else {
-      headers = new DefaultHttpHeaders(false);
+      headers = new DefaultHttpHeaders();
       headers.add(req.headers());
       this.cachedHeaders = headers;
     }

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

@@ -21,54 +21,24 @@
 import static org.junit.Assert.assertThat;
 
 import java.io.Serializable;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.accumulators.Accumulator;
-import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.*;
 import org.apache.flink.api.common.accumulators.Histogram;
-import org.apache.flink.api.common.accumulators.IntCounter;
-import org.apache.flink.api.common.accumulators.LongCounter;
 import org.apache.flink.api.common.cache.DistributedCache;
 import org.apache.flink.api.common.externalresource.ExternalResourceInfo;
 import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
 import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.common.state.AggregatingState;
-import org.apache.flink.api.common.state.AggregatingStateDescriptor;
-import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.state.MapState;
-import org.apache.flink.api.common.state.MapStateDescriptor;
-import org.apache.flink.api.common.state.ReducingState;
-import org.apache.flink.api.common.state.ReducingStateDescriptor;
-import org.apache.flink.api.common.state.State;
-import org.apache.flink.api.common.state.StateDescriptor;
-import org.apache.flink.api.common.state.ValueState;
-import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.state.*;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.metrics.CharacterFilter;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Meter;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.metrics.*;
 import org.apache.flink.metrics.groups.OperatorMetricGroup;
-import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
-import org.apache.flink.runtime.state.Keyed;
-import org.apache.flink.runtime.state.KeyedStateBackend;
-import org.apache.flink.runtime.state.KeyedStateFunction;
-import org.apache.flink.runtime.state.PriorityComparable;
-import org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory;
-import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.*;
 import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
 import org.apache.flink.runtime.state.internal.InternalListState;
 import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.MoreExecutors;
@@ -352,10 +322,13 @@ public boolean deregisterKeySelectionListener(KeySelectionListener<Object> liste
 
     @Nonnull
     @Override
-    public <N, SV, SEV, S extends State, IS extends S> IS createInternalState(
-        @Nonnull TypeSerializer<N> namespaceSerializer,
-        @Nonnull StateDescriptor<S, SV> stateDesc,
-        @Nonnull StateSnapshotTransformFactory<SEV> snapshotTransformFactory) {
+    public <N, SV, SEV, S extends State, IS extends S> IS createOrUpdateInternalState(
+        @Nonnull TypeSerializer<N> typeSerializer,
+        @Nonnull StateDescriptor<S, SV> stateDescriptor,
+        @Nonnull
+            StateSnapshotTransformer.StateSnapshotTransformFactory<SEV>
+                stateSnapshotTransformFactory)
+        throws Exception {
       throw new UnsupportedOperationException();
     }
 

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 apache/flink:1.16.2-scala_2.12-java8
 
 ENV ROLE worker
 ENV MASTER_HOST localhost