[FLINK-36366][core] Remove deprecated serializer related config option and method

Author: reswqa

flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+
+import com.esotericsoftware.kryo.Serializer;
+
+import java.io.Serializable;
+
+/**
+ * The wrapper to make serializer serializable.
+ *
+ * <p>This can be removed after {@link KryoSerializer} only allow serializer class.
+ */
+@Internal
+public class SerializableSerializer<T extends Serializer<?> & Serializable>
+        implements Serializable {
+    private static final long serialVersionUID = 4687893502781067189L;
+
+    private T serializer;
+
+    public SerializableSerializer(T serializer) {
+        this.serializer = serializer;
+    }
+
+    public T getSerializer() {
+        return serializer;
+    }
+}

flink-core/src/main/java/org/apache/flink/api/common/SerializableSerializer.java

@@ -20,7 +20,6 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.configuration.ReadableConfig;
@@ -128,30 +127,10 @@ <T extends Serializer<?> & Serializable> void registerTypeWithKryoSerializer(
     @Internal
     void registerKryoType(Class<?> type);
 
-    /**
-     * Returns the registered types with Kryo Serializers.
-     *
-     * @deprecated The method is deprecated because instance-type Kryo serializer definition based
-     *     on {@link ExecutionConfig.SerializableSerializer} is deprecated. Use class-type Kryo
-     *     serializers instead.
-     */
-    @Deprecated
-    LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            getRegisteredTypesWithKryoSerializers();
-
     /** Returns the registered types with their Kryo Serializer classes. */
     LinkedHashMap<Class<?>, Class<? extends Serializer<?>>>
             getRegisteredTypesWithKryoSerializerClasses();
 
-    /**
-     * Returns the registered default Kryo Serializers.
-     *
-     * @deprecated The method is deprecated because {@link ExecutionConfig.SerializableSerializer}
-     *     is deprecated.
-     */
-    @Deprecated
-    LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>> getDefaultKryoSerializers();
-
     /** Returns the registered default Kryo Serializer classes. */
     LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> getDefaultKryoSerializerClasses();
 

flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java

@@ -19,7 +19,7 @@
 package org.apache.flink.api.common.serialization;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.SerializableSerializer;
 import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -55,14 +55,14 @@ public final class SerializerConfigImpl implements SerializerConfig {
     // we store them in linked maps/sets to ensure they are registered in order in all kryo
     // instances.
 
-    private LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            registeredTypesWithKryoSerializers = new LinkedHashMap<>();
+    private LinkedHashMap<Class<?>, SerializableSerializer<?>> registeredTypesWithKryoSerializers =
+            new LinkedHashMap<>();
 
     private LinkedHashMap<Class<?>, Class<? extends Serializer<?>>>
             registeredTypesWithKryoSerializerClasses = new LinkedHashMap<>();
 
-    private LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            defaultKryoSerializers = new LinkedHashMap<>();
+    private LinkedHashMap<Class<?>, SerializableSerializer<?>> defaultKryoSerializers =
+            new LinkedHashMap<>();
 
     private LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses =
             new LinkedHashMap<>();
@@ -104,7 +104,7 @@ public <T extends Serializer<?> & Serializable> void addDefaultKryoSerializer(
             throw new NullPointerException("Cannot register null class or serializer.");
         }
 
-        defaultKryoSerializers.put(type, new ExecutionConfig.SerializableSerializer<>(serializer));
+        defaultKryoSerializers.put(type, new SerializableSerializer<>(serializer));
     }
 
     /**
@@ -137,8 +137,7 @@ public <T extends Serializer<?> & Serializable> void registerTypeWithKryoSeriali
             throw new NullPointerException("Cannot register null class or serializer.");
         }
 
-        registeredTypesWithKryoSerializers.put(
-                type, new ExecutionConfig.SerializableSerializer<>(serializer));
+        registeredTypesWithKryoSerializers.put(type, new SerializableSerializer<>(serializer));
     }
 
     /**
@@ -192,7 +191,7 @@ public void registerKryoType(Class<?> type) {
     }
 
     /** Returns the registered types with Kryo Serializers. */
-    public LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
+    public LinkedHashMap<Class<?>, SerializableSerializer<?>>
             getRegisteredTypesWithKryoSerializers() {
         return registeredTypesWithKryoSerializers;
     }
@@ -204,8 +203,7 @@ public void registerKryoType(Class<?> type) {
     }
 
     /** Returns the registered default Kryo Serializers. */
-    public LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            getDefaultKryoSerializers() {
+    public LinkedHashMap<Class<?>, SerializableSerializer<?>> getDefaultKryoSerializers() {
         return defaultKryoSerializers;
     }
 
@@ -359,21 +357,6 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
                 .getOptional(PipelineOptions.FORCE_KRYO_AVRO)
                 .ifPresent(this::setForceKryoAvro);
 
-        configuration
-                .getOptional(PipelineOptions.KRYO_DEFAULT_SERIALIZERS)
-                .map(s -> parseKryoSerializersWithExceptionHandling(classLoader, s))
-                .ifPresent(s -> this.defaultKryoSerializerClasses = s);
-
-        configuration
-                .getOptional(PipelineOptions.POJO_REGISTERED_CLASSES)
-                .map(c -> loadClasses(c, classLoader, "Could not load pojo type to be registered."))
-                .ifPresent(c -> this.registeredPojoTypes = c);
-
-        configuration
-                .getOptional(PipelineOptions.KRYO_REGISTERED_CLASSES)
-                .map(c -> loadClasses(c, classLoader, "Could not load kryo type to be registered."))
-                .ifPresent(c -> this.registeredKryoTypes = c);
-
         try {
             configuration
                     .getOptional(PipelineOptions.SERIALIZATION_CONFIG)

flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java

@@ -19,7 +19,7 @@
 package org.apache.flink.api.java.typeutils.runtime;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.SerializableSerializer;
 import org.apache.flink.util.Preconditions;
 
 import com.esotericsoftware.kryo.Kryo;
@@ -60,8 +60,7 @@ public enum SerializerDefinitionType {
      * serializer definition type is {@link SerializerDefinitionType#INSTANCE}.
      */
     @Nullable
-    private final ExecutionConfig.SerializableSerializer<? extends Serializer<?>>
-            serializableSerializerInstance;
+    private final SerializableSerializer<? extends Serializer<?>> serializableSerializerInstance;
 
     private final SerializerDefinitionType serializerDefinitionType;
 
@@ -86,8 +85,7 @@ public KryoRegistration(
 
     public KryoRegistration(
             Class<?> registeredClass,
-            ExecutionConfig.SerializableSerializer<? extends Serializer<?>>
-                    serializableSerializerInstance) {
+            SerializableSerializer<? extends Serializer<?>> serializableSerializerInstance) {
         this.registeredClass = Preconditions.checkNotNull(registeredClass);
 
         this.serializerClass = null;
@@ -111,8 +109,7 @@ public Class<? extends Serializer<?>> getSerializerClass() {
     }
 
     @Nullable
-    public ExecutionConfig.SerializableSerializer<? extends Serializer<?>>
-            getSerializableSerializerInstance() {
+    public SerializableSerializer<? extends Serializer<?>> getSerializableSerializerInstance() {
         return serializableSerializerInstance;
     }
 

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoRegistration.java

@@ -19,9 +19,9 @@
 package org.apache.flink.api.java.typeutils.runtime.kryo;
 
 import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
+import org.apache.flink.api.common.SerializableSerializer;
 import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.java.typeutils.AvroUtils;
@@ -138,8 +138,7 @@ private static ChillSerializerRegistrar loadFlinkChillPackageRegistrar() {
 
     // ------------------------------------------------------------------------
 
-    private final LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            defaultSerializers;
+    private final LinkedHashMap<Class<?>, SerializableSerializer<?>> defaultSerializers;
     private final LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> defaultSerializerClasses;
 
     /**
@@ -167,8 +166,7 @@ private static ChillSerializerRegistrar loadFlinkChillPackageRegistrar() {
     // ------------------------------------------------------------------------
     // legacy fields; these fields cannot yet be removed to retain backwards compatibility
 
-    private LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-            registeredTypesWithSerializers;
+    private LinkedHashMap<Class<?>, SerializableSerializer<?>> registeredTypesWithSerializers;
     private LinkedHashMap<Class<?>, Class<? extends Serializer<?>>>
             registeredTypesWithSerializerClasses;
     private LinkedHashSet<Class<?>> registeredTypes;
@@ -181,15 +179,17 @@ private static ChillSerializerRegistrar loadFlinkChillPackageRegistrar() {
     public KryoSerializer(Class<T> type, SerializerConfig serializerConfig) {
         this.type = checkNotNull(type);
 
-        this.defaultSerializers = serializerConfig.getDefaultKryoSerializers();
+        this.defaultSerializers =
+                ((SerializerConfigImpl) serializerConfig).getDefaultKryoSerializers();
         this.defaultSerializerClasses = serializerConfig.getDefaultKryoSerializerClasses();
 
         this.kryoRegistrations =
                 buildKryoRegistrations(
                         this.type,
                         serializerConfig.getRegisteredKryoTypes(),
                         serializerConfig.getRegisteredTypesWithKryoSerializerClasses(),
-                        serializerConfig.getRegisteredTypesWithKryoSerializers(),
+                        ((SerializerConfigImpl) serializerConfig)
+                                .getRegisteredTypesWithKryoSerializers(),
                         serializerConfig.isForceKryoAvroEnabled());
     }
 
@@ -206,7 +206,7 @@ protected KryoSerializer(KryoSerializer<T> toCopy) {
                 CollectionUtil.newLinkedHashMapWithExpectedSize(toCopy.kryoRegistrations.size());
 
         // deep copy the serializer instances in defaultSerializers
-        for (Map.Entry<Class<?>, ExecutionConfig.SerializableSerializer<?>> entry :
+        for (Map.Entry<Class<?>, SerializableSerializer<?>> entry :
                 toCopy.defaultSerializers.entrySet()) {
 
             this.defaultSerializers.put(entry.getKey(), deepCopySerializer(entry.getValue()));
@@ -220,7 +220,7 @@ protected KryoSerializer(KryoSerializer<T> toCopy) {
             if (kryoRegistration.getSerializerDefinitionType()
                     == KryoRegistration.SerializerDefinitionType.INSTANCE) {
 
-                ExecutionConfig.SerializableSerializer<? extends Serializer<?>> serializerInstance =
+                SerializableSerializer<? extends Serializer<?>> serializerInstance =
                         kryoRegistration.getSerializableSerializerInstance();
 
                 if (serializerInstance != null) {
@@ -561,7 +561,7 @@ private void checkKryoInitialized() {
 
             // Add default serializers first, so that the type registrations without a serializer
             // are registered with a default serializer
-            for (Map.Entry<Class<?>, ExecutionConfig.SerializableSerializer<?>> entry :
+            for (Map.Entry<Class<?>, SerializableSerializer<?>> entry :
                     defaultSerializers.entrySet()) {
                 kryo.addDefaultSerializer(entry.getKey(), entry.getValue().getSerializer());
             }
@@ -606,8 +606,7 @@ private static LinkedHashMap<String, KryoRegistration> buildKryoRegistrations(
             LinkedHashSet<Class<?>> registeredTypes,
             LinkedHashMap<Class<?>, Class<? extends Serializer<?>>>
                     registeredTypesWithSerializerClasses,
-            LinkedHashMap<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-                    registeredTypesWithSerializers,
+            LinkedHashMap<Class<?>, SerializableSerializer<?>> registeredTypesWithSerializers,
             TernaryBoolean isForceAvroKryoEnabledOpt) {
 
         final LinkedHashMap<String, KryoRegistration> kryoRegistrations = new LinkedHashMap<>();
@@ -629,9 +628,8 @@ private static LinkedHashMap<String, KryoRegistration> buildKryoRegistrations(
                             registeredTypeWithSerializerClassEntry.getValue()));
         }
 
-        for (Map.Entry<Class<?>, ExecutionConfig.SerializableSerializer<?>>
-                registeredTypeWithSerializerEntry :
-                        checkNotNull(registeredTypesWithSerializers).entrySet()) {
+        for (Map.Entry<Class<?>, SerializableSerializer<?>> registeredTypeWithSerializerEntry :
+                checkNotNull(registeredTypesWithSerializers).entrySet()) {
 
             kryoRegistrations.put(
                     registeredTypeWithSerializerEntry.getKey().getName(),
@@ -684,8 +682,8 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
         }
     }
 
-    private ExecutionConfig.SerializableSerializer<? extends Serializer<?>> deepCopySerializer(
-            ExecutionConfig.SerializableSerializer<? extends Serializer<?>> original) {
+    private SerializableSerializer<? extends Serializer<?>> deepCopySerializer(
+            SerializableSerializer<? extends Serializer<?>> original) {
         try {
             return InstantiationUtil.clone(
                     original, Thread.currentThread().getContextClassLoader());

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java

@@ -18,7 +18,7 @@
 
 package org.apache.flink.api.java.typeutils.runtime.kryo;
 
-import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
+import org.apache.flink.api.common.SerializableSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java

@@ -18,7 +18,7 @@
 
 package org.apache.flink.api.java.typeutils.runtime.kryo;
 
-import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
+import org.apache.flink.api.common.SerializableSerializer;
 import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
 import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
 import org.apache.flink.api.java.typeutils.runtime.KryoRegistration;

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotData.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.typeutils.AvroUtils;
@@ -89,7 +90,7 @@ public static void recursivelyRegisterType(
         if (type.isArray()) {
             recursivelyRegisterType(type.getComponentType(), config, alreadySeen);
         } else {
-            config.registerKryoType(type);
+            ((SerializerConfigImpl) config).registerKryoType(type);
             // add serializers for Avro type if necessary
             AvroUtils.getAvroUtils().addAvroSerializersIfRequired(config, type);