(flink) 20/32: [FLINK-30613][serializer] Migrate LinkedListSerializer to implement new method of resolving schema compatibility

[hangxiang]

This is an automated email from the ASF dual-hosted git repository.

hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 3cf1327124f8ad8d910d45a5c719d16bcb9b6093
Author: Hangxiang Yu <master...@gmail.com>
AuthorDate: Tue Jan 24 10:41:07 2023 +0800

    [FLINK-30613][serializer] Migrate LinkedListSerializer to implement new 
method of resolving schema compatibility
---
 .../flink/table/runtime/typeutils/LinkedListSerializer.java    | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java
index 9fe9ba42e12..59498a5fa6e 100644
--- 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/LinkedListSerializer.java
@@ -258,8 +258,14 @@ public final class LinkedListSerializer<T> extends 
TypeSerializer<LinkedList<T>>
 
         @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;