[FLINK-30613][serializer] Migrate PojoSerializer to implement new met…

…hod of resolving schema compatibility
pppsunil · Jan 15, 2024 · 21b22c0 · 21b22c0
1 parent 75c62a0
commit 21b22c0
Show file tree

Hide file tree

Showing 4 changed files with 161 additions and 178 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -664,7 +664,8 @@ public PojoSerializerSnapshot<T> snapshotConfiguration() {
                 registeredSerializers,
                 fields,
                 fieldSerializers,
-                subclassSerializerCache);
+                subclassSerializerCache,
+                executionConfig);
     }
 
     // --------------------------------------------------------------------------------------------
@@ -863,7 +864,8 @@ private static <T> PojoSerializerSnapshot<T> buildSnapshot(
             TypeSerializer<?>[] registeredSubclassSerializers,
             Field[] fields,
             TypeSerializer<?>[] fieldSerializers,
-            Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache) {
+            Map<Class<?>, TypeSerializer<?>> nonRegisteredSubclassSerializerCache,
+            ExecutionConfig executionConfig) {
 
         final LinkedHashMap<Class<?>, TypeSerializer<?>> subclassRegistry =
                 CollectionUtil.newLinkedHashMapWithExpectedSize(registeredSubclassesToTags.size());
@@ -877,6 +879,7 @@ private static <T> PojoSerializerSnapshot<T> buildSnapshot(
                 fields,
                 fieldSerializers,
                 subclassRegistry,
-                nonRegisteredSubclassSerializerCache);
+                nonRegisteredSubclassSerializerCache,
+                executionConfig);
     }
 }