[FLINK-30613][serializer] Remove useless correspondingSerializerClass…

… of CompositeTypeSerializerSnapshot

flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java

@@ -28,7 +28,6 @@
 import org.apache.flink.core.memory.DataOutputView;
 
 import java.io.IOException;
-import java.util.Arrays;
 
 import static org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil.IntermediateCompatibilityResult;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -123,30 +122,29 @@ protected enum OuterSchemaCompatibility {
 
     private NestedSerializersSnapshotDelegate nestedSerializersSnapshotDelegate;
 
-    private final Class<S> correspondingSerializerClass;
-
     /**
      * Constructor to be used for read instantiation.
      *
+     * @deprecated correspondingSerializerClass is not used to resolve and cast after FLIP-263,
+     *     please use {@link CompositeTypeSerializerSnapshot()} instead.
      * @param correspondingSerializerClass the expected class of the new serializer.
      */
-    @SuppressWarnings("unchecked")
+    @Deprecated
     public CompositeTypeSerializerSnapshot(
-            Class<? extends TypeSerializer> correspondingSerializerClass) {
-        this.correspondingSerializerClass = (Class<S>) checkNotNull(correspondingSerializerClass);
-    }
+            Class<? extends TypeSerializer> correspondingSerializerClass) {}
+
+    /** Constructor to be used for read instantiation. */
+    public CompositeTypeSerializerSnapshot() {}
 
     /**
      * Constructor to be used for writing the snapshot.
      *
      * @param serializerInstance an instance of the originating serializer of this snapshot.
      */
-    @SuppressWarnings("unchecked")
     public CompositeTypeSerializerSnapshot(S serializerInstance) {
         checkNotNull(serializerInstance);
         this.nestedSerializersSnapshotDelegate =
                 new NestedSerializersSnapshotDelegate(getNestedSerializers(serializerInstance));
-        this.correspondingSerializerClass = (Class<S>) serializerInstance.getClass();
     }
 
     @Override

flink-core/src/main/java/org/apache/flink/api/common/typeutils/SingleThreadAccessCheckingTypeSerializer.java

@@ -151,12 +151,8 @@ public static class SingleThreadAccessCheckingTypeSerializerSnapshot<T>
             extends CompositeTypeSerializerSnapshot<
                     T, SingleThreadAccessCheckingTypeSerializer<T>> {
 
-        @SuppressWarnings({"unchecked", "unused"})
-        public SingleThreadAccessCheckingTypeSerializerSnapshot() {
-            super(
-                    (Class<SingleThreadAccessCheckingTypeSerializer<T>>)
-                            (Class<?>) SingleThreadAccessCheckingTypeSerializer.class);
-        }
+        @SuppressWarnings("unused")
+        public SingleThreadAccessCheckingTypeSerializerSnapshot() {}
 
         SingleThreadAccessCheckingTypeSerializerSnapshot(
                 SingleThreadAccessCheckingTypeSerializer<T> serializerInstance) {

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerSnapshot.java

@@ -43,9 +43,7 @@ public final class GenericArraySerializerSnapshot<C>
     private Class<C> componentClass;
 
     /** Constructor to be used for read instantiation. */
-    public GenericArraySerializerSnapshot() {
-        super(GenericArraySerializer.class);
-    }
+    public GenericArraySerializerSnapshot() {}
 
     /** Constructor to be used for writing the snapshot. */
     public GenericArraySerializerSnapshot(GenericArraySerializer<C> genericArraySerializer) {
@@ -59,7 +57,6 @@ public GenericArraySerializerSnapshot(GenericArraySerializer<C> genericArraySeri
      */
     @SuppressWarnings("deprecation")
     GenericArraySerializerSnapshot(Class<C> componentClass) {
-        super(GenericArraySerializer.class);
         this.componentClass = componentClass;
     }
 

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshot.java

@@ -30,9 +30,7 @@ public class ListSerializerSnapshot<T>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    public ListSerializerSnapshot() {
-        super(ListSerializer.class);
-    }
+    public ListSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public ListSerializerSnapshot(ListSerializer<T> listSerializer) {

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/MapSerializerSnapshot.java

@@ -31,9 +31,7 @@ public class MapSerializerSnapshot<K, V>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    public MapSerializerSnapshot() {
-        super(MapSerializer.class);
-    }
+    public MapSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public MapSerializerSnapshot(MapSerializer<K, V> mapSerializer) {

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

@@ -34,10 +34,7 @@ public class JavaEitherSerializerSnapshot<L, R>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    @SuppressWarnings("unused")
-    public JavaEitherSerializerSnapshot() {
-        super(EitherSerializer.class);
-    }
+    public JavaEitherSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public JavaEitherSerializerSnapshot(EitherSerializer<L, R> eitherSerializer) {

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

@@ -271,17 +271,14 @@ public static class NullableSerializerSnapshot<T>
         private int nullPaddingLength;
 
         @SuppressWarnings("unused")
-        public NullableSerializerSnapshot() {
-            super(NullableSerializer.class);
-        }
+        public NullableSerializerSnapshot() {}
 
         public NullableSerializerSnapshot(NullableSerializer<T> serializerInstance) {
             super(serializerInstance);
             this.nullPaddingLength = serializerInstance.nullPaddingLength();
         }
 
         private NullableSerializerSnapshot(int nullPaddingLength) {
-            super(NullableSerializer.class);
             checkArgument(
                     nullPaddingLength >= 0,
                     "Computed NULL padding can not be negative. %s",

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

@@ -485,9 +485,7 @@ public static final class RowSerializerSnapshot
 
         private boolean supportsRowKind = true;
 
-        public RowSerializerSnapshot() {
-            super(RowSerializer.class);
-        }
+        public RowSerializerSnapshot() {}
 
         RowSerializerSnapshot(RowSerializer serializerInstance) {
             super(serializerInstance);

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

@@ -41,9 +41,7 @@ public final class TupleSerializerSnapshot<T extends Tuple>
     private Class<T> tupleClass;
 
     @SuppressWarnings("unused")
-    public TupleSerializerSnapshot() {
-        super(TupleSerializer.class);
-    }
+    public TupleSerializerSnapshot() {}
 
     TupleSerializerSnapshot(TupleSerializer<T> serializerInstance) {
         super(serializerInstance);
@@ -56,7 +54,6 @@ public TupleSerializerSnapshot() {
      * TupleSerializer#resolveSchemaCompatibilityViaRedirectingToNewSnapshotClass}.
      */
     TupleSerializerSnapshot(Class<T> tupleClass) {
-        super(TupleSerializer.class);
         this.tupleClass = checkNotNull(tupleClass, "tuple class can not be NULL");
     }
 

flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeSerializerTest.java

@@ -214,7 +214,6 @@ public static class TestListCompositeSerializerSnapshot
 
         /** Constructor for read instantiation. */
         public TestListCompositeSerializerSnapshot() {
-            super(TestListCompositeSerializer.class);
             this.isImmutableTargetType = false;
         }
 

flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java

@@ -367,9 +367,7 @@ public static class TestCompositeTypeSerializerSnapshot
 
         private OuterSchemaCompatibility mockOuterSchemaCompatibility;
 
-        public TestCompositeTypeSerializerSnapshot() {
-            super(TestCompositeTypeSerializer.class);
-        }
+        public TestCompositeTypeSerializerSnapshot() {}
 
         TestCompositeTypeSerializerSnapshot(TestCompositeTypeSerializer serializer) {
             super(serializer);

flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/ValueWithTs.java

@@ -122,9 +122,7 @@ public static final class ValueWithTsSerializerSnapshot
         private static final int VERSION = 2;
 
         @SuppressWarnings("unused")
-        public ValueWithTsSerializerSnapshot() {
-            super(Serializer.class);
-        }
+        public ValueWithTsSerializerSnapshot() {}
 
         ValueWithTsSerializerSnapshot(Serializer serializerInstance) {
             super(serializerInstance);

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java

@@ -984,9 +984,7 @@ public static final class MigratedNFASerializerSnapshot<T>
 
         private static final int VERSION = 2;
 
-        public MigratedNFASerializerSnapshot() {
-            super(NFASerializer.class);
-        }
+        public MigratedNFASerializerSnapshot() {}
 
         MigratedNFASerializerSnapshot(NFASerializer<T> legacyNfaSerializer) {
             super(legacyNfaSerializer);

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAStateSerializerSnapshot.java

@@ -39,9 +39,7 @@ public class NFAStateSerializerSnapshot
     private boolean supportsPreviousTimestamp = true;
 
     /** Constructor for read instantiation. */
-    public NFAStateSerializerSnapshot() {
-        super(NFAStateSerializer.class);
-    }
+    public NFAStateSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public NFAStateSerializerSnapshot(NFAStateSerializer serializerInstance) {

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java

@@ -159,9 +159,7 @@ public static final class SharedBufferSerializerSnapshot<K, V>
 
         private static final int VERSION = 2;
 
-        public SharedBufferSerializerSnapshot() {
-            super(SharedBufferSerializer.class);
-        }
+        public SharedBufferSerializerSnapshot() {}
 
         public SharedBufferSerializerSnapshot(SharedBufferSerializer<K, V> sharedBufferSerializer) {
             super(sharedBufferSerializer);

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/LockableTypeSerializerSnapshot.java

@@ -31,9 +31,7 @@ public class LockableTypeSerializerSnapshot<E>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    public LockableTypeSerializerSnapshot() {
-        super(Lockable.LockableTypeSerializer.class);
-    }
+    public LockableTypeSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public LockableTypeSerializerSnapshot(

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/NodeId.java

@@ -167,9 +167,7 @@ public static final class NodeIdSerializerSnapshot
 
             private static final int VERSION = 1;
 
-            public NodeIdSerializerSnapshot() {
-                super(NodeIdSerializer.class);
-            }
+            public NodeIdSerializerSnapshot() {}
 
             public NodeIdSerializerSnapshot(NodeIdSerializer nodeIdSerializer) {
                 super(nodeIdSerializer);

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferEdge.java

@@ -170,9 +170,7 @@ public static final class SharedBufferEdgeSerializerSnapshot
 
             private static final int VERSION = 1;
 
-            public SharedBufferEdgeSerializerSnapshot() {
-                super(SharedBufferEdgeSerializer.class);
-            }
+            public SharedBufferEdgeSerializerSnapshot() {}
 
             public SharedBufferEdgeSerializerSnapshot(
                     SharedBufferEdgeSerializer sharedBufferEdgeSerializer) {

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNode.java

@@ -171,9 +171,7 @@ public static final class SharedBufferNodeSerializerSnapshot
 
             private static final int VERSION = 1;
 
-            public SharedBufferNodeSerializerSnapshot() {
-                super(SharedBufferNodeSerializer.class);
-            }
+            public SharedBufferNodeSerializerSnapshot() {}
 
             public SharedBufferNodeSerializerSnapshot(
                     SharedBufferNodeSerializer sharedBufferNodeSerializer) {

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferNodeSerializerSnapshotV2.java

@@ -27,9 +27,7 @@ public final class SharedBufferNodeSerializerSnapshotV2
 
     private static final int VERSION = 1;
 
-    public SharedBufferNodeSerializerSnapshotV2() {
-        super(SharedBufferNodeSerializer.class);
-    }
+    public SharedBufferNodeSerializerSnapshotV2() {}
 
     public SharedBufferNodeSerializerSnapshotV2(
             SharedBufferNodeSerializer sharedBufferNodeSerializer) {

flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializerSnapshot.java

@@ -30,9 +30,7 @@ public class ArrayListSerializerSnapshot<T>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    public ArrayListSerializerSnapshot() {
-        super(ArrayListSerializer.class);
-    }
+    public ArrayListSerializerSnapshot() {}
 
     /** Constructor for creating the snapshot for writing. */
     public ArrayListSerializerSnapshot(ArrayListSerializer<T> arrayListSerializer) {

flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlStateFactory.java

@@ -349,9 +349,7 @@ public static final class TtlSerializerSnapshot<T>
         private static final int VERSION = 2;
 
         @SuppressWarnings({"WeakerAccess", "unused"})
-        public TtlSerializerSnapshot() {
-            super(TtlSerializer.class);
-        }
+        public TtlSerializerSnapshot() {}
 
         TtlSerializerSnapshot(TtlSerializer<T> serializerInstance) {
             super(serializerInstance);

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ScalaCaseClassSerializerSnapshot.java

@@ -43,9 +43,7 @@ public final class ScalaCaseClassSerializerSnapshot<T extends scala.Product>
 
     /** Used via reflection. */
     @SuppressWarnings("unused")
-    public ScalaCaseClassSerializerSnapshot() {
-        super(ScalaCaseClassSerializer.class);
-    }
+    public ScalaCaseClassSerializerSnapshot() {}
 
     /** Used for the snapshot path. */
     public ScalaCaseClassSerializerSnapshot(ScalaCaseClassSerializer<T> serializerInstance) {

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ScalaEitherSerializerSnapshot.java

@@ -33,9 +33,7 @@ public class ScalaEitherSerializerSnapshot<L, R>
     private static final int CURRENT_VERSION = 1;
 
     /** Constructor for read instantiation. */
-    public ScalaEitherSerializerSnapshot() {
-        super(EitherSerializer.class);
-    }
+    public ScalaEitherSerializerSnapshot() {}
 
     /** Constructor to create the snapshot for writing. */
     public ScalaEitherSerializerSnapshot(EitherSerializer<L, R> eitherSerializer) {

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ScalaOptionSerializerSnapshot.java

@@ -31,9 +31,7 @@ public final class ScalaOptionSerializerSnapshot<E>
     private static final int VERSION = 2;
 
     @SuppressWarnings("WeakerAccess")
-    public ScalaOptionSerializerSnapshot() {
-        super(OptionSerializer.class);
-    }
+    public ScalaOptionSerializerSnapshot() {}
 
     public ScalaOptionSerializerSnapshot(OptionSerializer<E> serializerInstance) {
         super(serializerInstance);

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ScalaTrySerializerSnapshot.java

@@ -38,9 +38,7 @@ public class ScalaTrySerializerSnapshot<E>
 
     /** This empty nullary constructor is required for deserializing the configuration. */
     @SuppressWarnings("unused")
-    public ScalaTrySerializerSnapshot() {
-        super(TrySerializer.class);
-    }
+    public ScalaTrySerializerSnapshot() {}
 
     public ScalaTrySerializerSnapshot(TrySerializer<E> trySerializer) {
         super(trySerializer);

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/TraversableSerializerSnapshot.java

@@ -47,17 +47,14 @@ public class TraversableSerializerSnapshot<T extends TraversableOnce<E>, E>
     private String cbfCode;
 
     @SuppressWarnings("unused")
-    public TraversableSerializerSnapshot() {
-        super(TraversableSerializer.class);
-    }
+    public TraversableSerializerSnapshot() {}
 
     public TraversableSerializerSnapshot(TraversableSerializer<T, E> serializerInstance) {
         super(serializerInstance);
         this.cbfCode = serializerInstance.cbfCode();
     }
 
     TraversableSerializerSnapshot(String cbfCode) {
-        super(TraversableSerializer.class);
         checkArgument(cbfCode != null, "cbfCode cannot be null");
 
         this.cbfCode = cbfCode;

flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/Tuple2CaseClassSerializerSnapshot.java

@@ -46,9 +46,7 @@ public final class Tuple2CaseClassSerializerSnapshot<T1, T2>
     private Class<Tuple2<T1, T2>> type;
 
     @SuppressWarnings("unused")
-    public Tuple2CaseClassSerializerSnapshot() {
-        super(correspondingSerializerClass());
-    }
+    public Tuple2CaseClassSerializerSnapshot() {}
 
     public Tuple2CaseClassSerializerSnapshot(
             ScalaCaseClassSerializer<Tuple2<T1, T2>> serializerInstance) {
@@ -102,9 +100,4 @@ protected void readOuterSnapshot(
                 ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
                 : OuterSchemaCompatibility.INCOMPATIBLE;
     }
-
-    private static <T1, T2>
-            Class<ScalaCaseClassSerializer<scala.Tuple2<T1, T2>>> correspondingSerializerClass() {
-        return package$.MODULE$.tuple2ClassForJava();
-    }
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java

@@ -752,9 +752,7 @@ public static class UnionSerializerSnapshot<T1, T2>
         private static final int VERSION = 2;
 
         @SuppressWarnings("WeakerAccess")
-        public UnionSerializerSnapshot() {
-            super(UnionSerializer.class);
-        }
+        public UnionSerializerSnapshot() {}
 
         UnionSerializerSnapshot(UnionSerializer<T1, T2> serializerInstance) {
             super(serializerInstance);