(flink) 08/32: [FLINK-30613][serializer] Migrate NullableSerializer 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 4fb1ca6fcca0c67019f5874b9f96884a1367c69e
Author: Hangxiang Yu <master...@gmail.com>
AuthorDate: Tue Jan 24 01:04:27 2023 +0800

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

diff --git 
a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
 
b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
index 32f2467bb14..635b2f1323d 100644
--- 
a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
+++ 
b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java
@@ -328,8 +328,13 @@ public class NullableSerializer<T> extends 
TypeSerializer<T> {
 
         @Override
         protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(
-                NullableSerializer<T> newSerializer) {
-            return (nullPaddingLength == newSerializer.nullPaddingLength())
+                TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+            if (!(oldSerializerSnapshot instanceof 
NullableSerializerSnapshot)) {
+                return OuterSchemaCompatibility.INCOMPATIBLE;
+            }
+            NullableSerializerSnapshot<T> oldNullableSerializerSnapshot =
+                    (NullableSerializerSnapshot<T>) oldSerializerSnapshot;
+            return (nullPaddingLength == 
oldNullableSerializerSnapshot.nullPaddingLength)
                     ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
                     : OuterSchemaCompatibility.INCOMPATIBLE;
         }