[FLINK-36903][core][type] Add switch for built-in serialization suppo…

…rt for common collection types

docs/content/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md

@@ -170,7 +170,7 @@ Flink supports all Java and Scala primitive types such as `Integer`, `String`, a
 
 #### Common Collection Types
 
-Flink comes with dedicated serialization support for common Java collection types, which is more efficient than going
+Since Flink 2.0, Flink comes with dedicated serialization support for common Java collection types, which is more efficient than going
 through a general purpose serialization framework. Currently, only `Map`, `List`, `Set` and its super interface `Collection`
 are supported. To utilize it, you need to declare the collection type with:
 
@@ -182,6 +182,10 @@ are supported. To utilize it, you need to declare the collection type with:
 Other nonqualified collection types will be handled by Flink as general class types. If the implementation types are
 also required to be preserved, you also need to register it with a custom serializer.
 
+You may disable the built-in type support for collection types via
+[pipeline.built-in-collection-types]({{< ref "docs/deployment/config#pipeline-built-in-collection-types" >}})
+to achieve the same serialization behavior in Flink versions prior to 2.0.
+
 #### General Class Types
 
 Flink supports most Java and Scala classes (API and custom).

docs/layouts/shortcodes/generated/pipeline_configuration.html

@@ -20,6 +20,12 @@
             <td>Duration</td>
             <td>The interval of the automatic watermark emission. Watermarks are used throughout the streaming system to keep track of the progress of time. They are used, for example, for time based windowing.</td>
         </tr>
+        <tr>
+            <td><h5>pipeline.built-in-collection-types</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>If enabled, TypeExtractor will use built-in serializers for Maps, Lists, and Sets, which need to be treated by Kyro otherwise.</td>
+        </tr>
         <tr>
             <td><h5>pipeline.cached-files</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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

@@ -356,6 +356,9 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
         configuration
                 .getOptional(PipelineOptions.FORCE_KRYO_AVRO)
                 .ifPresent(this::setForceKryoAvro);
+        configuration
+                .getOptional(PipelineOptions.BUILT_IN_COLLECTION_TYPES)
+                .ifPresent(TypeExtractor::setBuiltInCollectionTypesEnabled);
 
         try {
             configuration

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

@@ -134,6 +134,18 @@ protected TypeExtractor() {
         // only create instances for special use cases
     }
 
+    /**
+     * Whether to use built-in types for common collection types, which need to be treated by Kyro
+     * otherwise. Since TypeExtractor methods are mostly static, we use a static variable to control
+     * their behavior as a temporary solution.
+     */
+    private static boolean builtInCollectionTypesEnabled = true;
+
+    @Internal
+    public static void setBuiltInCollectionTypesEnabled(boolean enabled) {
+        builtInCollectionTypesEnabled = enabled;
+    }
+
     // --------------------------------------------------------------------------------------------
     //  TypeInfoFactory registry
     // --------------------------------------------------------------------------------------------
@@ -1979,7 +1991,7 @@ private <OUT, IN1, IN2> TypeInformation<OUT> privateGetForClass(
         // - OK: List<String>, Collection<String>
         // - not OK: LinkedList<String> (implementation type), List (raw type), List<T> (generic
         // type argument), or List<?> (wildcard type argument)
-        if (parameterizedType != null) {
+        if (parameterizedType != null && builtInCollectionTypesEnabled) {
             Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
             boolean allTypeArgumentsConcrete =
                     Arrays.stream(actualTypeArguments).allMatch(arg -> arg instanceof Class<?>);

flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java

@@ -170,6 +170,13 @@ public class PipelineOptions {
                                                     + " would cause the program to fail.")
                                     .build());
 
+    public static final ConfigOption<Boolean> BUILT_IN_COLLECTION_TYPES =
+            key("pipeline.built-in-collection-types")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "If enabled, TypeExtractor will use built-in serializers for Maps, Lists, and Sets, which need to be treated by Kyro otherwise.");
+
     public static final ConfigOption<Map<String, String>> GLOBAL_JOB_PARAMETERS =
             key("pipeline.global-job-parameters")
                     .mapType()

flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java

@@ -2587,4 +2587,66 @@ private void testCollectionTypesInternal(TypeInformation<?> ti) {
                                 .getTypeInformation())
                 .isInstanceOf(GenericTypeInfo.class);
     }
+
+    @Test
+    public <T> void testCollectionTypesWithoutBuiltInTypes() {
+        TypeExtractor.setBuiltInCollectionTypesEnabled(false);
+
+        MapFunction<?, ?> function =
+                new MapFunction<PojoWithCollections<T>, PojoWithCollections<T>>() {
+                    @Override
+                    public PojoWithCollections map(PojoWithCollections<T> value) {
+                        return null;
+                    }
+                };
+        TypeInformation<?> ti =
+                TypeExtractor.getMapReturnTypes(
+                        function,
+                        (TypeInformation)
+                                TypeInformation.of(new TypeHint<PojoWithCollections<T>>() {}));
+        assertThat(ti).isInstanceOf(PojoTypeInfo.class);
+        testGenericCollectionTypes(ti);
+
+        // use getForClass()
+        TypeInformation<?> ti2 = TypeExtractor.getForClass(PojoWithCollections.class);
+        assertThat(ti2).isInstanceOf(PojoTypeInfo.class);
+        testGenericCollectionTypes(ti2);
+
+        // use getForObject()
+        PojoWithCollections<T> t = new PojoWithCollections<>();
+        TypeInformation<?> ti3 = TypeExtractor.getForObject(t);
+        assertThat(ti3).isInstanceOf(PojoTypeInfo.class);
+        testGenericCollectionTypes(ti3);
+
+        // restore state
+        TypeExtractor.setBuiltInCollectionTypesEnabled(true);
+    }
+
+    private void testGenericCollectionTypes(TypeInformation<?> ti) {
+        PojoTypeInfo<?> pojoTi = (PojoTypeInfo<?>) ti;
+        assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("mapVal")).getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("listVal")).getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(
+                        pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("collectionVal"))
+                                .getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("setVal")).getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(
+                        pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("linkedListVal"))
+                                .getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("rawListVal")).getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(
+                        pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("genericListVal"))
+                                .getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+        assertThat(
+                        pojoTi.getPojoFieldAt(pojoTi.getFieldIndex("wildcardListVal"))
+                                .getTypeInformation())
+                .isInstanceOf(GenericTypeInfo.class);
+    }
 }