[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-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,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-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());

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

@@ -40,7 +40,9 @@
 import com.esotericsoftware.kryo.KryoException;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.KryoBufferUnderflowException;
 import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.apache.commons.lang3.exception.CloneFailedException;
 import org.objenesis.strategy.StdInstantiatorStrategy;
 import org.slf4j.Logger;
@@ -344,9 +346,9 @@ public void serialize(T record, DataOutputView target) throws IOException {
                 kryo.writeClassAndObject(output, record);
                 output.flush();
             } catch (KryoException ke) {
-                // make sure that the Kryo output buffer is cleared in case that we can recover from
+                // make sure that the Kryo output buffer is reset in case that we can recover from
                 // the exception (e.g. EOFException which denotes buffer full)
-                output.clear();
+                output.reset();
 
                 Throwable cause = ke.getCause();
                 if (cause instanceof EOFException) {
@@ -380,6 +382,9 @@ public T deserialize(DataInputView source) throws IOException {
 
             try {
                 return (T) kryo.readClassAndObject(input);
+            } catch (KryoBufferUnderflowException ke) {
+                // 2023-04-26: Existing Flink code expects a java.io.EOFException in this scenario
+                throw new EOFException(ke.getMessage());
             } catch (KryoException ke) {
                 Throwable cause = ke.getCause();
 
@@ -488,7 +493,7 @@ private Kryo getKryoInstance() {
                 LOG.info("Kryo serializer scala extensions are not available.");
             }
 
-            Kryo.DefaultInstantiatorStrategy initStrategy = new Kryo.DefaultInstantiatorStrategy();
+            DefaultInstantiatorStrategy initStrategy = new DefaultInstantiatorStrategy();
             initStrategy.setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
 
             Kryo kryo = new Kryo();

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

@@ -32,16 +32,13 @@
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
-import com.esotericsoftware.kryo.serializers.CollectionSerializer;
 
-import java.io.Serializable;
 import java.lang.reflect.Field;
 import java.lang.reflect.GenericArrayType;
 import java.lang.reflect.Modifier;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 
@@ -171,50 +168,8 @@ public void write(Kryo kryo, Output output, Object o) {
         }
 
         @Override
-        public T read(Kryo kryo, Input input, Class<T> aClass) {
+        public T read(Kryo kryo, Input input, Class<? extends T> aClass) {
             throw new UnsupportedOperationException("Could not find required Avro dependency.");
         }
     }
-
-    // --------------------------------------------------------------------------------------------
-    // Custom Serializers
-    // --------------------------------------------------------------------------------------------
-
-    /** Special serializer for Java's {@link ArrayList} used for Avro's GenericData.Array. */
-    @SuppressWarnings("rawtypes")
-    public static class SpecificInstanceCollectionSerializerForArrayList
-            extends SpecificInstanceCollectionSerializer<ArrayList> {
-        private static final long serialVersionUID = 1L;
-
-        public SpecificInstanceCollectionSerializerForArrayList() {
-            super(ArrayList.class);
-        }
-    }
-
-    /**
-     * Special serializer for Java collections enforcing certain instance types. Avro is serializing
-     * collections with an "GenericData.Array" type. Kryo is not able to handle this type, so we use
-     * ArrayLists.
-     */
-    @SuppressWarnings("rawtypes")
-    public static class SpecificInstanceCollectionSerializer<T extends Collection>
-            extends CollectionSerializer implements Serializable {
-        private static final long serialVersionUID = 1L;
-
-        private Class<T> type;
-
-        public SpecificInstanceCollectionSerializer(Class<T> type) {
-            this.type = type;
-        }
-
-        @Override
-        protected Collection create(Kryo kryo, Input input, Class<Collection> type) {
-            return kryo.newInstance(this.type);
-        }
-
-        @Override
-        protected Collection createCopy(Kryo kryo, Collection original) {
-            return kryo.newInstance(this.type);
-        }
-    }
 }