修改之前合并代码冲突导致的代码丢失

Author: xuchao

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaProducerFactory.java

@@ -26,9 +26,6 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.formats.avro.AvroRowSerializationSchema;
-import org.apache.flink.formats.csv.CsvRowSerializationSchema;
-import org.apache.flink.formats.json.DTJsonRowSerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;

…s/json/DTJsonRowSerializationSchema.java …zation/DTJsonRowSerializationSchema.javakafka-base/kafka-base-sink/src/main/java/org/apache/flink/formats/json/DTJsonRowSerializationSchema.java renamed to kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/DTJsonRowSerializationSchema.java kafka-base/kafka-base-sink/src/main/java/org/apache/flink/formats/json/DTJsonRowSerializationSchema.java renamed to kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/DTJsonRowSerializationSchema.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.formats.json;
+package com.dtstack.flink.sql.sink.kafka.serialization;
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -25,6 +25,7 @@
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSchemaConverter;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/JsonCRowSerializationSchema.java

@@ -159,34 +159,34 @@ private ObjectNode convertRow(ObjectNode reuse, RowTypeInfo info, Row row) {
     }
 
     private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) {
-        if (info == Types.VOID || object == null) {
+        if (info.equals(Types.VOID) || object == null) {
             return container.nullNode();
-        } else if (info == Types.BOOLEAN) {
+        } else if (info.equals(Types.BOOLEAN)) {
             return container.booleanNode((Boolean) object);
-        } else if (info == Types.STRING) {
+        } else if (info.equals(Types.STRING)) {
             return container.textNode((String) object);
-        } else if (info == Types.BIG_DEC) {
+        } else if (info.equals(Types.BIG_DEC)) {
             // convert decimal if necessary
             if (object instanceof BigDecimal) {
                 return container.numberNode((BigDecimal) object);
             }
             return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue()));
-        } else if (info == Types.BIG_INT) {
+        } else if (info.equals(Types.BIG_INT)) {
             // convert integer if necessary
             if (object instanceof BigInteger) {
                 return container.numberNode((BigInteger) object);
             }
             return container.numberNode(BigInteger.valueOf(((Number) object).longValue()));
-        } else if (info == Types.SQL_DATE) {
+        } else if (info.equals(Types.SQL_DATE)) {
             return container.textNode(object.toString());
-        } else if (info == Types.SQL_TIME) {
+        } else if (info.equals(Types.SQL_TIME)) {
             final Time time = (Time) object;
             // strip milliseconds if possible
             if (time.getTime() % 1000 > 0) {
                 return container.textNode(timeFormatWithMillis.format(time));
             }
             return container.textNode(timeFormat.format(time));
-        } else if (info == Types.SQL_TIMESTAMP) {
+        } else if (info.equals(Types.SQL_TIMESTAMP)) {
             return container.textNode(timestampFormat.format((Timestamp) object));
         } else if (info instanceof RowTypeInfo) {
             if (reuse != null && reuse instanceof ObjectNode) {