[FLINK-34239][core,table] Add copy() in SerializerConfig

flink-ci · Mar 12, 2024 · 195c8ab · 195c8ab
1 parent 94b55d1
commit 195c8ab
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 38 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java
@@ -209,4 +209,6 @@ <T extends Serializer<?> & Serializable> void registerTypeWithKryoSerializer(
      * @param classLoader a class loader to use when loading classes
      */
     void configure(ReadableConfig configuration, ClassLoader classLoader);
+
+    SerializerConfigImpl copy();
 }
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java
@@ -557,4 +557,29 @@ private void registerTypeWithTypeInfoFactory(
     public ExecutionConfig getExecutionConfig() {
         return executionConfig;
     }
+
+    @Override
+    public SerializerConfigImpl copy() {
+        final SerializerConfigImpl newSerializerConfig = new SerializerConfigImpl();
+        newSerializerConfig.configure(configuration, this.getClass().getClassLoader());
+
+        getRegisteredTypesWithKryoSerializers()
+                .forEach(
+                        (c, s) ->
+                                newSerializerConfig.registerTypeWithKryoSerializer(
+                                        c, s.getSerializer()));
+        getRegisteredTypesWithKryoSerializerClasses()
+                .forEach(newSerializerConfig::registerTypeWithKryoSerializer);
+        getDefaultKryoSerializers()
+                .forEach(
+                        (c, s) ->
+                                newSerializerConfig.addDefaultKryoSerializer(c, s.getSerializer()));
+        getDefaultKryoSerializerClasses().forEach(newSerializerConfig::addDefaultKryoSerializer);
+        getRegisteredKryoTypes().forEach(newSerializerConfig::registerKryoType);
+        getRegisteredPojoTypes().forEach(newSerializerConfig::registerPojoType);
+        getRegisteredTypeInfoFactories()
+                .forEach(newSerializerConfig::registerTypeWithTypeInfoFactory);
+
+        return newSerializerConfig;
+    }
 }
diff --git a/...link-table-api-java/src/main/java/org/apache/flink/table/catalog/DataTypeFactoryImpl.java b/...link-table-api-java/src/main/java/org/apache/flink/table/catalog/DataTypeFactoryImpl.java
@@ -131,46 +131,12 @@ public LogicalType createLogicalType(UnresolvedIdentifier identifier) {
     private static Supplier<SerializerConfig> createSerializerConfig(
             ClassLoader classLoader, ReadableConfig config, SerializerConfig serializerConfig) {
         return () -> {
-            final SerializerConfig newSerializerConfig = new SerializerConfigImpl();
-
+            SerializerConfig newSerializerConfig = new SerializerConfigImpl();
             if (serializerConfig != null) {
-                if (serializerConfig.isForceKryoEnabled()) {
-                    newSerializerConfig.setForceKryo(true);
-                }
-
-                if (serializerConfig.isForceAvroEnabled()) {
-                    newSerializerConfig.setForceAvro(true);
-                }
-
-                serializerConfig
-                        .getDefaultKryoSerializers()
-                        .forEach(
-                                (c, s) ->
-                                        newSerializerConfig.addDefaultKryoSerializer(
-                                                c, s.getSerializer()));
-
-                serializerConfig
-                        .getDefaultKryoSerializerClasses()
-                        .forEach(newSerializerConfig::addDefaultKryoSerializer);
-
-                serializerConfig
-                        .getRegisteredKryoTypes()
-                        .forEach(newSerializerConfig::registerKryoType);
-
-                serializerConfig
-                        .getRegisteredTypesWithKryoSerializerClasses()
-                        .forEach(newSerializerConfig::registerTypeWithKryoSerializer);
-
-                serializerConfig
-                        .getRegisteredTypesWithKryoSerializers()
-                        .forEach(
-                                (c, s) ->
-                                        newSerializerConfig.registerTypeWithKryoSerializer(
-                                                c, s.getSerializer()));
+                newSerializerConfig = serializerConfig.copy();
+            } else {
+                newSerializerConfig.configure(config, classLoader);
             }
-
-            newSerializerConfig.configure(config, classLoader);
-
             return newSerializerConfig;
         };
     }