[refacto] Fix missing conversions of TypeSerializerSnapshot.resolveSchemaCompatibility()

Author: arnaud-daroussin

modules/flink-common-api/src/main/java/org/apache/flinkx/api/serializer/ScalaCaseClassSerializerSnapshot.java

@@ -32,7 +32,9 @@
 import static org.apache.flink.util.Preconditions.checkNotNull;
 import static org.apache.flink.util.Preconditions.checkState;
 
-/** {@link TypeSerializerSnapshot} for {@link CaseClassSerializer}. */
+/**
+ * {@link TypeSerializerSnapshot} for {@link CaseClassSerializer}.
+ */
 @Internal
 public final class ScalaCaseClassSerializerSnapshot<T extends scala.Product>
         extends CompositeTypeSerializerSnapshot<T, CaseClassSerializer<T>> {
@@ -42,13 +44,17 @@ public final class ScalaCaseClassSerializerSnapshot<T extends scala.Product>
     private Class<T> type;
     private boolean isCaseClassImmutable;
 
-    /** Used via reflection. */
+    /**
+     * Used via reflection.
+     */
     @SuppressWarnings("unused")
     public ScalaCaseClassSerializerSnapshot() {
         super(CaseClassSerializer.class);
     }
 
-    /** Used for the snapshot path. */
+    /**
+     * Used for the snapshot path.
+     */
     public ScalaCaseClassSerializerSnapshot(CaseClassSerializer<T> serializerInstance) {
         super(serializerInstance);
         this.type = checkNotNull(serializerInstance.getTupleClass(), "tuple class can not be NULL");
@@ -91,9 +97,13 @@ protected void readOuterSnapshot(
 
     @Override
     protected CompositeTypeSerializerSnapshot.OuterSchemaCompatibility
-    resolveOuterSchemaCompatibility(CaseClassSerializer<T> newSerializer) {
+    resolveOuterSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+        if (!(oldSerializerSnapshot instanceof ScalaCaseClassSerializerSnapshot)) {
+            return OuterSchemaCompatibility.INCOMPATIBLE;
+        }
+        var caseClassSerializerSnapshot = (ScalaCaseClassSerializerSnapshot<T>) oldSerializerSnapshot;
         var currentTypeName = Optional.ofNullable(type).map(Class::getName);
-        var newTypeName = Optional.ofNullable(newSerializer.getTupleClass()).map(Class::getName);
+        var newTypeName = Optional.ofNullable(caseClassSerializerSnapshot.type).map(Class::getName);
         if (currentTypeName.equals(newTypeName)) {
             return OuterSchemaCompatibility.COMPATIBLE_AS_IS;
         } else {

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/CoproductSerializer.scala

@@ -117,7 +117,9 @@ object CoproductSerializer {
       })
     }
 
-    override def resolveSchemaCompatibility(newSerializer: TypeSerializer[T]): TypeSerializerSchemaCompatibility[T] =
+    override def resolveSchemaCompatibility(
+        oldSerializer: TypeSerializerSnapshot[T]
+    ): TypeSerializerSchemaCompatibility[T] =
       TypeSerializerSchemaCompatibility.compatibleAsIs()
 
     override def restoreSerializer(): TypeSerializer[T] =

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/MapSerializer.scala

@@ -99,8 +99,9 @@ object MapSerializer {
     }
 
     override def resolveSchemaCompatibility(
-        newSerializer: TypeSerializer[Map[K, V]]
-    ): TypeSerializerSchemaCompatibility[Map[K, V]] = TypeSerializerSchemaCompatibility.compatibleAsIs()
+        oldSerializer: TypeSerializerSnapshot[Map[K, V]]
+    ): TypeSerializerSchemaCompatibility[Map[K, V]] =
+      TypeSerializerSchemaCompatibility.compatibleAsIs()
 
     override def restoreSerializer(): TypeSerializer[Map[K, V]] = new MapSerializer(keySerializer, valueSerializer)
   }

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/MappedSerializer.scala

@@ -92,7 +92,9 @@ object MappedSerializer {
       }
     }
 
-    override def resolveSchemaCompatibility(newSerializer: TypeSerializer[A]): TypeSerializerSchemaCompatibility[A] =
+    override def resolveSchemaCompatibility(
+        oldSerializer: TypeSerializerSnapshot[A]
+    ): TypeSerializerSchemaCompatibility[A] =
       TypeSerializerSchemaCompatibility.compatibleAsIs()
 
     override def writeSnapshot(out: DataOutputView): Unit = {

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/ScalaCaseObjectSerializer.scala

@@ -1,10 +1,9 @@
 package org.apache.flinkx.api.serializer
 
-import org.apache.flinkx.api.serializer.ScalaCaseObjectSerializer.ScalaCaseObjectSerializerSnapshot
 import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSchemaCompatibility, TypeSerializerSnapshot}
-import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton
 import org.apache.flink.core.memory.{DataInputView, DataOutputView}
 import org.apache.flink.util.InstantiationUtil
+import org.apache.flinkx.api.serializer.ScalaCaseObjectSerializer.ScalaCaseObjectSerializerSnapshot
 
 class ScalaCaseObjectSerializer[T](clazz: Class[T]) extends ImmutableSerializer[T] {
   override def copy(source: DataInputView, target: DataOutputView): Unit = {}
@@ -32,8 +31,10 @@ object ScalaCaseObjectSerializer {
       out.writeUTF(clazz.getName)
     }
 
-    override def getCurrentVersion: Int                                                                             = 1
-    override def resolveSchemaCompatibility(newSerializer: TypeSerializer[T]): TypeSerializerSchemaCompatibility[T] =
+    override def getCurrentVersion: Int = 1
+    override def resolveSchemaCompatibility(
+        oldSerializer: TypeSerializerSnapshot[T]
+    ): TypeSerializerSchemaCompatibility[T] =
       TypeSerializerSchemaCompatibility.compatibleAsIs()
 
     override def restoreSerializer(): TypeSerializer[T] =