[FLINK-30613][serializer] Migrate LinkedListSerializer to implement n…

…ew method of resolving schema compatibility
pppsunil · Jan 15, 2024 · 3cf1327 · 3cf1327
1 parent 8430068
commit 3cf1327
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/...-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java b/...-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java
@@ -258,8 +258,14 @@ protected void writeOuterSnapshot(DataOutputView out) throws IOException {
 
         @Override
         protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(
-                LinkedListSerializer<T> newSerializer) {
-            if (hasNullMask != newSerializer.hasNullMask) {
+                TypeSerializerSnapshot<LinkedList<T>> oldSerializerSnapshot) {
+            if (!(oldSerializerSnapshot instanceof LinkedListSerializerSnapshot)) {
+                return OuterSchemaCompatibility.INCOMPATIBLE;
+            }
+
+            LinkedListSerializerSnapshot<T> oldLinkedListSerializerSnapshot =
+                    (LinkedListSerializerSnapshot<T>) oldSerializerSnapshot;
+            if (hasNullMask != oldLinkedListSerializerSnapshot.hasNullMask) {
                 return OuterSchemaCompatibility.COMPATIBLE_AFTER_MIGRATION;
             }
             return OuterSchemaCompatibility.COMPATIBLE_AS_IS;