[FLINK-36858][pipeline-connector][kafka] Fix compatibility with Flink 1.20 JsonRowDataSerializationSchema

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/pom.xml

@@ -115,6 +115,19 @@ limitations under the License.
                     </execution>
                 </executions>
             </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>test-jar</id>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/ChangeLogJsonFormatFactory.java

@@ -22,6 +22,7 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonSerializationSchema;
 import org.apache.flink.cdc.connectors.kafka.json.debezium.DebeziumJsonSerializationSchema;
+import org.apache.flink.cdc.connectors.kafka.utils.JsonRowDataSerializationSchemaUtils;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions;
@@ -57,6 +58,9 @@ public static SerializationSchema<Event> createSerializationSchema(
         final boolean encodeDecimalAsPlainNumber =
                 formatOptions.get(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
 
+        final boolean ignoreNullFields =
+                JsonRowDataSerializationSchemaUtils.enableIgnoreNullFields(formatOptions);
+
         switch (type) {
             case DEBEZIUM_JSON:
                 {
@@ -65,7 +69,8 @@ public static SerializationSchema<Event> createSerializationSchema(
                             mapNullKeyMode,
                             mapNullKeyLiteral,
                             zoneId,
-                            encodeDecimalAsPlainNumber);
+                            encodeDecimalAsPlainNumber,
+                            ignoreNullFields);
                 }
             case CANAL_JSON:
                 {
@@ -74,7 +79,8 @@ public static SerializationSchema<Event> createSerializationSchema(
                             mapNullKeyMode,
                             mapNullKeyLiteral,
                             zoneId,
-                            encodeDecimalAsPlainNumber);
+                            encodeDecimalAsPlainNumber,
+                            ignoreNullFields);
                 }
             default:
                 {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java

@@ -27,6 +27,7 @@
 import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
 import org.apache.flink.cdc.common.utils.SchemaUtils;
 import org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo;
+import org.apache.flink.cdc.connectors.kafka.utils.JsonRowDataSerializationSchemaUtils;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions;
 import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
@@ -73,6 +74,8 @@ public class CanalJsonSerializationSchema implements SerializationSchema<Event>
 
     private final boolean encodeDecimalAsPlainNumber;
 
+    private final boolean ignoreNullFields;
+
     private final ZoneId zoneId;
 
     private InitializationContext context;
@@ -82,13 +85,15 @@ public CanalJsonSerializationSchema(
             JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
             String mapNullKeyLiteral,
             ZoneId zoneId,
-            boolean encodeDecimalAsPlainNumber) {
+            boolean encodeDecimalAsPlainNumber,
+            boolean ignoreNullFields) {
         this.timestampFormat = timestampFormat;
         this.mapNullKeyMode = mapNullKeyMode;
         this.mapNullKeyLiteral = mapNullKeyLiteral;
         this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
         this.zoneId = zoneId;
         jsonSerializers = new HashMap<>();
+        this.ignoreNullFields = ignoreNullFields;
     }
 
     @Override
@@ -114,12 +119,13 @@ public byte[] serialize(Event event) {
             LogicalType rowType =
                     DataTypeUtils.toFlinkDataType(schema.toRowDataType()).getLogicalType();
             JsonRowDataSerializationSchema jsonSerializer =
-                    new JsonRowDataSerializationSchema(
+                    JsonRowDataSerializationSchemaUtils.createSerializationSchema(
                             createJsonRowType(fromLogicalToDataType(rowType)),
                             timestampFormat,
                             mapNullKeyMode,
                             mapNullKeyLiteral,
-                            encodeDecimalAsPlainNumber);
+                            encodeDecimalAsPlainNumber,
+                            ignoreNullFields);
             try {
                 jsonSerializer.open(context);
             } catch (Exception e) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java

@@ -27,6 +27,7 @@
 import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
 import org.apache.flink.cdc.common.utils.SchemaUtils;
 import org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo;
+import org.apache.flink.cdc.connectors.kafka.utils.JsonRowDataSerializationSchemaUtils;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions;
 import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
@@ -72,6 +73,8 @@ public class DebeziumJsonSerializationSchema implements SerializationSchema<Even
 
     private final boolean encodeDecimalAsPlainNumber;
 
+    private final boolean ignoreNullFields;
+
     private final ZoneId zoneId;
 
     private InitializationContext context;
@@ -81,13 +84,15 @@ public DebeziumJsonSerializationSchema(
             JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
             String mapNullKeyLiteral,
             ZoneId zoneId,
-            boolean encodeDecimalAsPlainNumber) {
+            boolean encodeDecimalAsPlainNumber,
+            boolean ignoreNullFields) {
         this.timestampFormat = timestampFormat;
         this.mapNullKeyMode = mapNullKeyMode;
         this.mapNullKeyLiteral = mapNullKeyLiteral;
         this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
         this.zoneId = zoneId;
         jsonSerializers = new HashMap<>();
+        this.ignoreNullFields = ignoreNullFields;
     }
 
     @Override
@@ -113,12 +118,13 @@ public byte[] serialize(Event event) {
             LogicalType rowType =
                     DataTypeUtils.toFlinkDataType(schema.toRowDataType()).getLogicalType();
             JsonRowDataSerializationSchema jsonSerializer =
-                    new JsonRowDataSerializationSchema(
+                    JsonRowDataSerializationSchemaUtils.createSerializationSchema(
                             createJsonRowType(fromLogicalToDataType(rowType)),
                             timestampFormat,
                             mapNullKeyMode,
                             mapNullKeyLiteral,
-                            encodeDecimalAsPlainNumber);
+                            encodeDecimalAsPlainNumber,
+                            ignoreNullFields);
             try {
                 jsonSerializer.open(context);
             } catch (Exception e) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/serialization/JsonSerializationSchema.java

@@ -33,6 +33,7 @@
 import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
 import org.apache.flink.cdc.common.utils.SchemaUtils;
 import org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo;
+import org.apache.flink.cdc.connectors.kafka.utils.JsonRowDataSerializationSchemaUtils;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions;
 import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
@@ -61,6 +62,8 @@ public class JsonSerializationSchema implements SerializationSchema<Event> {
 
     private final boolean encodeDecimalAsPlainNumber;
 
+    private final boolean ignoreNullFields;
+
     private final ZoneId zoneId;
 
     private InitializationContext context;
@@ -70,13 +73,15 @@ public JsonSerializationSchema(
             JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
             String mapNullKeyLiteral,
             ZoneId zoneId,
-            boolean encodeDecimalAsPlainNumber) {
+            boolean encodeDecimalAsPlainNumber,
+            boolean ignoreNullFields) {
         this.timestampFormat = timestampFormat;
         this.mapNullKeyMode = mapNullKeyMode;
         this.mapNullKeyLiteral = mapNullKeyLiteral;
         this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
         this.zoneId = zoneId;
         jsonSerializers = new HashMap<>();
+        this.ignoreNullFields = ignoreNullFields;
     }
 
     @Override
@@ -131,11 +136,12 @@ private JsonRowDataSerializationSchema buildSerializationForPrimaryKey(Schema sc
         // the row should never be null
         DataType dataType = DataTypes.ROW(fields).notNull();
         LogicalType rowType = DataTypeUtils.toFlinkDataType(dataType).getLogicalType();
-        return new JsonRowDataSerializationSchema(
+        return JsonRowDataSerializationSchemaUtils.createSerializationSchema(
                 (RowType) rowType,
                 timestampFormat,
                 mapNullKeyMode,
                 mapNullKeyLiteral,
-                encodeDecimalAsPlainNumber);
+                encodeDecimalAsPlainNumber,
+                ignoreNullFields);
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KeySerializationFactory.java

@@ -21,6 +21,7 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.connectors.kafka.serialization.CsvSerializationSchema;
 import org.apache.flink.cdc.connectors.kafka.serialization.JsonSerializationSchema;
+import org.apache.flink.cdc.connectors.kafka.utils.JsonRowDataSerializationSchemaUtils;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions;
@@ -54,12 +55,16 @@ public static SerializationSchema<Event> createSerializationSchema(
 
                     final boolean encodeDecimalAsPlainNumber =
                             formatOptions.get(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
+                    final boolean ignoreNullFields =
+                            JsonRowDataSerializationSchemaUtils.enableIgnoreNullFields(
+                                    formatOptions);
                     return new JsonSerializationSchema(
                             timestampFormat,
                             mapNullKeyMode,
                             mapNullKeyLiteral,
                             zoneId,
-                            encodeDecimalAsPlainNumber);
+                            encodeDecimalAsPlainNumber,
+                            ignoreNullFields);
                 }
             case CSV:
                 {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/utils/JsonRowDataSerializationSchemaUtils.java

@@ -0,0 +1,99 @@
+/*
+ * 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.cdc.connectors.kafka.utils;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+
+/**
+ * Utils for creating JsonRowDataSerializationSchema.TODO: Remove this class after bump to Flink
+ * 1.20 or higher.
+ */
+public class JsonRowDataSerializationSchemaUtils {
+
+    /**
+     * In flink>=1.20, the constructor of JsonRowDataSerializationSchema has 6 parameters, and in
+     * flink<1.20, the constructor of JsonRowDataSerializationSchema has 5 parameters.
+     */
+    public static JsonRowDataSerializationSchema createSerializationSchema(
+            RowType rowType,
+            TimestampFormat timestampFormat,
+            JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
+            String mapNullKeyLiteral,
+            boolean encodeDecimalAsPlainNumber,
+            boolean ignoreNullFields) {
+        try {
+            Class<?>[] fullParams =
+                    new Class[] {
+                        RowType.class,
+                        TimestampFormat.class,
+                        JsonFormatOptions.MapNullKeyMode.class,
+                        String.class,
+                        boolean.class,
+                        boolean.class
+                    };
+
+            Object[] fullParamValues =
+                    new Object[] {
+                        rowType,
+                        timestampFormat,
+                        mapNullKeyMode,
+                        mapNullKeyLiteral,
+                        encodeDecimalAsPlainNumber,
+                        ignoreNullFields
+                    };
+
+            for (int i = fullParams.length; i >= 5; i--) {
+                try {
+                    Constructor<?> constructor =
+                            JsonRowDataSerializationSchema.class.getConstructor(
+                                    Arrays.copyOfRange(fullParams, 0, i));
+
+                    return (JsonRowDataSerializationSchema)
+                            constructor.newInstance(Arrays.copyOfRange(fullParamValues, 0, i));
+                } catch (NoSuchMethodException ignored) {
+                }
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    "Failed to create JsonRowDataSerializationSchema,please check your Flink version is 1.19 or 1.20.",
+                    e);
+        }
+        throw new RuntimeException(
+                "Failed to find appropriate constructor for JsonRowDataSerializationSchema,please check your Flink version is 1.19 or 1.20.");
+    }
+
+    /** flink>=1.20 only has the ENCODE_IGNORE_NULL_FIELDS parameter. */
+    public static boolean enableIgnoreNullFields(ReadableConfig formatOptions) {
+        try {
+            Field field = JsonFormatOptions.class.getField("ENCODE_IGNORE_NULL_FIELDS");
+            ConfigOption<Boolean> encodeOption = (ConfigOption<Boolean>) field.get(null);
+            return formatOptions.get(encodeOption);
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            return false;
+        }
+    }
+}

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml

@@ -98,6 +98,13 @@ limitations under the License.
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-cdc-pipeline-connector-kafka</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-test-util</artifactId>
@@ -126,6 +133,13 @@ limitations under the License.
             <version>${scala.version}</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -245,6 +259,16 @@ limitations under the License.
                             </outputDirectory>
                         </artifactItem>
 
+                        <artifactItem>
+                            <groupId>org.apache.flink</groupId>
+                            <artifactId>flink-cdc-pipeline-connector-kafka</artifactId>
+                            <version>${project.version}</version>
+                            <destFileName>kafka-cdc-pipeline-connector.jar</destFileName>
+                            <type>jar</type>
+                            <outputDirectory>${project.build.directory}/dependencies
+                            </outputDirectory>
+                        </artifactItem>
+
                         <artifactItem>
                             <groupId>org.apache.flink</groupId>
                             <artifactId>flink-cdc-pipeline-udf-examples</artifactId>