[FLINK-3154][API] Upgrade from Kryo 2.x to Kryo 5.x. Removed twitter chill companion library which is abandoned.

Author: kurtostfeld

flink-architecture-tests/flink-architecture-tests-production/archunit-violations/7602816f-5c01-4b7a-9e3e-235dfedec245

@@ -106,8 +106,6 @@ org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer does not satisfy
 org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
 org.apache.flink.api.java.typeutils.runtime.kryo.Serializers$DummyAvroKryoSerializerClass does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
 org.apache.flink.api.java.typeutils.runtime.kryo.Serializers$DummyAvroRegisteredClass does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
-org.apache.flink.api.java.typeutils.runtime.kryo.Serializers$SpecificInstanceCollectionSerializer does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
-org.apache.flink.api.java.typeutils.runtime.kryo.Serializers$SpecificInstanceCollectionSerializerForArrayList does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
 org.apache.flink.runtime.io.network.api.CancelCheckpointMarker does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
 org.apache.flink.runtime.io.network.api.CheckpointBarrier does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated
 org.apache.flink.runtime.io.network.api.EndOfData does not satisfy: annotated with @Internal or annotated with @Experimental or annotated with @PublicEvolving or annotated with @Public or annotated with @Deprecated

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableComparator.java

@@ -27,6 +27,7 @@
 import org.apache.flink.util.InstantiationUtil;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.apache.hadoop.io.Writable;
 import org.objenesis.strategy.StdInstantiatorStrategy;
 
@@ -174,12 +175,10 @@ private void checkKryoInitialized() {
         if (this.kryo == null) {
             this.kryo = new Kryo();
 
-            Kryo.DefaultInstantiatorStrategy instantiatorStrategy =
-                    new Kryo.DefaultInstantiatorStrategy();
+            DefaultInstantiatorStrategy instantiatorStrategy = new DefaultInstantiatorStrategy();
             instantiatorStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
             kryo.setInstantiatorStrategy(instantiatorStrategy);
 
-            this.kryo.setAsmEnabled(true);
             this.kryo.register(type);
         }
     }

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java

@@ -27,6 +27,7 @@
 import org.apache.flink.util.InstantiationUtil;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Writable;
 import org.objenesis.strategy.StdInstantiatorStrategy;
@@ -126,12 +127,10 @@ private void checkKryoInitialized() {
         if (this.kryo == null) {
             this.kryo = new Kryo();
 
-            Kryo.DefaultInstantiatorStrategy instantiatorStrategy =
-                    new Kryo.DefaultInstantiatorStrategy();
+            DefaultInstantiatorStrategy instantiatorStrategy = new DefaultInstantiatorStrategy();
             instantiatorStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
             kryo.setInstantiatorStrategy(instantiatorStrategy);
 
-            this.kryo.setAsmEnabled(true);
             this.kryo.register(typeClass);
         }
     }

flink-connectors/flink-sql-connector-hive-2.3.10/src/main/resources/META-INF/NOTICE

@@ -49,7 +49,7 @@ the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
 - org.apache.parquet:parquet-hadoop-bundle:1.8.1
 - org.apache.thrift:libthrift:0.14.1
 - org.apache.thrift:libfb303:0.9.3
-- org.objenesis:objenesis:2.1
+- org.objenesis:objenesis:3.4
 
 The bundled Apache Hive org.apache.hive:hive-exec dependency bundles the following dependencies under the BSD license.
 See bundled license files for details.

flink-connectors/flink-sql-connector-hive-3.1.3/src/main/resources/META-INF/NOTICE

@@ -46,7 +46,7 @@ the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
 - org.apache.parquet:parquet-hadoop-bundle:1.10.0
 - org.codehaus.jackson:jackson-core-asl:1.9.13
 - org.codehaus.jackson:jackson-mapper-asl:1.9.13
-- org.objenesis:objenesis:2.1
+- org.objenesis:objenesis:3.4
 
 The bundled Apache Hive org.apache.hive:hive-exec dependency bundles the following dependencies under the BSD license.
 See bundled license files for details.

flink-core/pom.xml

@@ -112,7 +112,7 @@ under the License.
 
 		<!-- for the fallback generic serializer -->
 		<dependency>
-			<groupId>com.esotericsoftware.kryo</groupId>
+			<groupId>com.esotericsoftware</groupId>
 			<artifactId>kryo</artifactId>
 			<!-- managed version -->
 		</dependency>

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoRegistration.java

@@ -24,7 +24,7 @@
 
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
-import com.esotericsoftware.kryo.factories.ReflectionSerializerFactory;
+import com.esotericsoftware.kryo.SerializerFactory.ReflectionSerializerFactory;
 
 import javax.annotation.Nullable;
 
@@ -118,7 +118,7 @@ public Serializer<?> getSerializer(Kryo kryo) {
             case UNSPECIFIED:
                 return null;
             case CLASS:
-                return ReflectionSerializerFactory.makeSerializer(
+                return ReflectionSerializerFactory.newSerializer(
                         kryo, serializerClass, registeredClass);
             case INSTANCE:
                 return serializableSerializerInstance.getSerializer();

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NoFetchingInput.java

@@ -22,6 +22,7 @@
 
 import com.esotericsoftware.kryo.KryoException;
 import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.KryoBufferUnderflowException;
 
 import java.io.EOFException;
 import java.io.IOException;
@@ -33,11 +34,6 @@ public NoFetchingInput(InputStream inputStream) {
         super(inputStream, 8);
     }
 
-    @Override
-    public boolean eof() {
-        throw new UnsupportedOperationException("NoFetchingInput does not support EOF.");
-    }
-
     @Override
     public int read() throws KryoException {
         require(1);
@@ -60,33 +56,56 @@ public boolean canReadLong() throws KryoException {
      * bytes. Thus, it will only load the data which is required and never prefetch data.
      *
      * @param required the number of bytes being available in the buffer
-     * @return the number of bytes remaining, which is equal to required
+     * @return The number of bytes remaining in the buffer, which will be at least <code>required
+     *     </code> bytes.
      * @throws KryoException
      */
     @Override
     protected int require(int required) throws KryoException {
+        // The main change between this and Kryo 5 Input.require is this will never read more bytes
+        // than required.
+        // There are also formatting changes to be compliant with the Flink project styling rules.
+        int remaining = limit - position;
+        if (remaining >= required) {
+            return remaining;
+        }
         if (required > capacity) {
             throw new KryoException(
-                    "Buffer too small: capacity: " + capacity + ", " + "required: " + required);
+                    "Buffer too small: capacity: " + capacity + ", required: " + required);
         }
 
-        position = 0;
-        int bytesRead = 0;
         int count;
-        while (true) {
-            count = fill(buffer, bytesRead, required - bytesRead);
-
+        // Try to fill the buffer.
+        if (remaining > 0) {
+            // Logical change 1 (from Kryo Input.require): "capacity - limit" -> "required - limit"
+            count = fill(buffer, limit, required - limit);
             if (count == -1) {
-                throw new KryoException(new EOFException("No more bytes left."));
+                throw new KryoBufferUnderflowException("Buffer underflow.");
             }
-
-            bytesRead += count;
-            if (bytesRead == required) {
-                break;
+            remaining += count;
+            if (remaining >= required) {
+                limit += count;
+                return remaining;
             }
         }
-        limit = required;
-        return required;
+
+        // Was not enough, compact and try again.
+        System.arraycopy(buffer, position, buffer, 0, remaining);
+        total += position;
+        position = 0;
+
+        do {
+            // Logical change 2 (from Kryo Input.require): "capacity - remaining" -> "required -
+            // remaining"
+            count = fill(buffer, remaining, required - remaining);
+            if (count == -1) {
+                throw new KryoBufferUnderflowException("Buffer underflow.");
+            }
+            remaining += count;
+        } while (remaining < required);
+
+        limit = remaining;
+        return remaining;
     }
 
     @Override

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueComparator.java

@@ -28,6 +28,7 @@
 import org.apache.flink.util.InstantiationUtil;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.objenesis.strategy.StdInstantiatorStrategy;
 
 import java.io.IOException;
@@ -142,12 +143,11 @@ private void checkKryoInitialized() {
         if (this.kryo == null) {
             this.kryo = new Kryo();
 
-            Kryo.DefaultInstantiatorStrategy instantiatorStrategy =
-                    new Kryo.DefaultInstantiatorStrategy();
+            DefaultInstantiatorStrategy instantiatorStrategy = new DefaultInstantiatorStrategy();
             instantiatorStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
             kryo.setInstantiatorStrategy(instantiatorStrategy);
 
-            this.kryo.setAsmEnabled(true);
+            // this.kryo.setAsmEnabled(true);
             this.kryo.register(type);
         }
     }

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java

@@ -29,6 +29,7 @@
 import org.apache.flink.util.InstantiationUtil;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.objenesis.strategy.StdInstantiatorStrategy;
 
 import java.io.IOException;
@@ -138,12 +139,11 @@ private void checkKryoInitialized() {
         if (this.kryo == null) {
             this.kryo = new Kryo();
 
-            Kryo.DefaultInstantiatorStrategy instantiatorStrategy =
-                    new Kryo.DefaultInstantiatorStrategy();
-            instantiatorStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
-            kryo.setInstantiatorStrategy(instantiatorStrategy);
+            DefaultInstantiatorStrategy initStrategy = new DefaultInstantiatorStrategy();
+            initStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
+            kryo.setInstantiatorStrategy(initStrategy);
 
-            this.kryo.setAsmEnabled(true);
+            // this.kryo.setAsmEnabled(true);
 
             KryoUtils.applyRegistrations(
                     this.kryo, kryoRegistrations.values(), this.kryo.getNextRegistrationId());