[Kafka] Clean up code in the flink-cdc-pipeline-connector-kafka module that is compatible with older versions of Flink. (apache#4347)

Signed-off-by: Pei Yu <125331682@qq.com>

Author: sd4324530

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,7 +22,6 @@
 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;
@@ -70,7 +69,7 @@ public static SerializationSchema<Event> createSerializationSchema(
                 formatOptions.get(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
 
         final boolean ignoreNullFields =
-                JsonRowDataSerializationSchemaUtils.enableIgnoreNullFields(formatOptions);
+                formatOptions.get(JsonFormatOptions.ENCODE_IGNORE_NULL_FIELDS);
 
         switch (type) {
             case DEBEZIUM_JSON:

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,7 +27,6 @@
 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;
@@ -119,7 +118,7 @@ public byte[] serialize(Event event) {
             LogicalType rowType =
                     DataTypeUtils.toFlinkDataType(schema.toRowDataType()).getLogicalType();
             JsonRowDataSerializationSchema jsonSerializer =
-                    JsonRowDataSerializationSchemaUtils.createSerializationSchema(
+                    new JsonRowDataSerializationSchema(
                             createJsonRowType(fromLogicalToDataType(rowType)),
                             timestampFormat,
                             mapNullKeyMode,

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

@@ -18,7 +18,6 @@
 package org.apache.flink.cdc.connectors.kafka.json.debezium;
 
 import org.apache.flink.api.common.serialization.SerializationSchema;
-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.JsonParserRowDataDeserializationSchema;
@@ -90,7 +89,7 @@ public DebeziumJsonRowDataSerializationSchema(
         this.mapNullKeyLiteral = mapNullKeyLiteral;
         this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
         this.runtimeConverter =
-                JsonRowDataSerializationSchemaUtils.createRowDataToJsonConverters(
+                new RowDataToJsonConverters(
                                 timestampFormat,
                                 mapNullKeyMode,
                                 mapNullKeyLiteral,

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,7 +33,6 @@
 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;
@@ -136,7 +135,7 @@ private JsonRowDataSerializationSchema buildSerializationForPrimaryKey(Schema sc
         // the row should never be null
         DataType dataType = DataTypes.ROW(fields).notNull();
         LogicalType rowType = DataTypeUtils.toFlinkDataType(dataType).getLogicalType();
-        return JsonRowDataSerializationSchemaUtils.createSerializationSchema(
+        return new JsonRowDataSerializationSchema(
                 (RowType) rowType,
                 timestampFormat,
                 mapNullKeyMode,

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,7 +21,6 @@
 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;
@@ -56,8 +55,7 @@ public static SerializationSchema<Event> createSerializationSchema(
                     final boolean encodeDecimalAsPlainNumber =
                             formatOptions.get(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
                     final boolean ignoreNullFields =
-                            JsonRowDataSerializationSchemaUtils.enableIgnoreNullFields(
-                                    formatOptions);
+                            formatOptions.get(JsonFormatOptions.ENCODE_IGNORE_NULL_FIELDS);
                     return new JsonSerializationSchema(
                             timestampFormat,
                             mapNullKeyMode,