use tuple replace crow

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -21,10 +21,10 @@
 import com.dtstack.flink.sql.metric.MetricConstant;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
-import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.types.Row;
 
 
@@ -35,11 +35,11 @@
  * author: toutian
  * create: 2019/12/24
  */
-public class SerializationMetricWrapper implements SerializationSchema<CRow> {
+public class SerializationMetricWrapper implements SerializationSchema<Tuple2<Boolean,Row>> {
 
     private static final long serialVersionUID = 1L;
 
-    private SerializationSchema<CRow> serializationSchema;
+    private SerializationSchema<Tuple2<Boolean,Row>> serializationSchema;
 
     private transient RuntimeContext runtimeContext;
 
@@ -48,7 +48,7 @@ public class SerializationMetricWrapper implements SerializationSchema<CRow> {
     protected transient Meter dtNumRecordsOutRate;
 
 
-    public SerializationMetricWrapper(SerializationSchema<CRow> serializationSchema) {
+    public SerializationMetricWrapper(SerializationSchema<Tuple2<Boolean,Row>> serializationSchema) {
         this.serializationSchema = serializationSchema;
     }
 
@@ -58,7 +58,7 @@ public void initMetric() {
     }
 
     @Override
-    public byte[] serialize(CRow element) {
+    public byte[] serialize(Tuple2<Boolean,Row> element) {
         beforeSerialize();
         byte[] row = serializationSchema.serialize(element);
         afterSerialize();
@@ -80,7 +80,7 @@ public void setRuntimeContext(RuntimeContext runtimeContext) {
         this.runtimeContext = runtimeContext;
     }
 
-    public SerializationSchema<CRow> getSerializationSchema() {
+    public SerializationSchema<Tuple2<Boolean,Row>> getSerializationSchema() {
         return serializationSchema;
     }
 

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

@@ -19,16 +19,17 @@
 
 import com.dtstack.flink.sql.format.FormatType;
 import com.dtstack.flink.sql.format.SerializationMetricWrapper;
-import com.dtstack.flink.sql.sink.kafka.serialization.AvroCRowSerializationSchema;
-import com.dtstack.flink.sql.sink.kafka.serialization.CsvCRowSerializationSchema;
-import com.dtstack.flink.sql.sink.kafka.serialization.JsonCRowSerializationSchema;
+import com.dtstack.flink.sql.sink.kafka.serialization.AvroTuple2SerializationSchema;
+import com.dtstack.flink.sql.sink.kafka.serialization.CsvTupleSerializationSchema;
+import com.dtstack.flink.sql.sink.kafka.serialization.JsonTupleSerializationSchema;
 import com.dtstack.flink.sql.sink.kafka.table.KafkaSinkTableInfo;
 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.api.java.tuple.Tuple2;
 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;
+import org.apache.flink.types.Row;
 
 import java.util.Optional;
 import java.util.Properties;
@@ -51,28 +52,29 @@ public abstract class AbstractKafkaProducerFactory {
      * @param partitioner
      * @return
      */
-    public abstract RichSinkFunction<CRow> createKafkaProducer(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<CRow> typeInformation, Properties properties, Optional<FlinkKafkaPartitioner<CRow>> partitioner, String[] partitionKeys);
+    public abstract RichSinkFunction<Tuple2<Boolean,Row>> createKafkaProducer(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Tuple2<Boolean,Row>> typeInformation,
+                                                    Properties properties, Optional<FlinkKafkaPartitioner<Tuple2<Boolean,Row>>> partitioner, String[] partitionKeys);
 
-    protected SerializationMetricWrapper createSerializationMetricWrapper(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<CRow> typeInformation) {
-        SerializationSchema<CRow> serializationSchema = createSerializationSchema(kafkaSinkTableInfo, typeInformation);
+    protected SerializationMetricWrapper createSerializationMetricWrapper(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Tuple2<Boolean,Row>> typeInformation) {
+        SerializationSchema<Tuple2<Boolean,Row>> serializationSchema = createSerializationSchema(kafkaSinkTableInfo, typeInformation);
         return new SerializationMetricWrapper(serializationSchema);
     }
 
-    private SerializationSchema<CRow> createSerializationSchema(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<CRow> typeInformation) {
-        SerializationSchema<CRow> serializationSchema = null;
+    private SerializationSchema<Tuple2<Boolean,Row>> createSerializationSchema(KafkaSinkTableInfo kafkaSinkTableInfo, TypeInformation<Tuple2<Boolean,Row>> typeInformation) {
+        SerializationSchema<Tuple2<Boolean,Row>> serializationSchema = null;
         if (FormatType.JSON.name().equalsIgnoreCase(kafkaSinkTableInfo.getSinkDataType())) {
             if (StringUtils.isNotBlank(kafkaSinkTableInfo.getSchemaString())) {
-                serializationSchema = new JsonCRowSerializationSchema(kafkaSinkTableInfo.getSchemaString(), kafkaSinkTableInfo.getUpdateMode());
+                serializationSchema = new JsonTupleSerializationSchema(kafkaSinkTableInfo.getSchemaString(), kafkaSinkTableInfo.getUpdateMode());
             } else if (typeInformation != null && typeInformation.getArity() != 0) {
-                serializationSchema = new JsonCRowSerializationSchema(typeInformation, kafkaSinkTableInfo.getUpdateMode());
+                serializationSchema = new JsonTupleSerializationSchema(typeInformation, kafkaSinkTableInfo.getUpdateMode());
             } else {
                 throw new IllegalArgumentException("sinkDataType:" + FormatType.JSON.name() + " must set schemaString（JSON Schema）or TypeInformation<Row>");
             }
         } else if (FormatType.CSV.name().equalsIgnoreCase(kafkaSinkTableInfo.getSinkDataType())) {
             if (StringUtils.isBlank(kafkaSinkTableInfo.getFieldDelimiter())) {
                 throw new IllegalArgumentException("sinkDataType:" + FormatType.CSV.name() + " must set fieldDelimiter");
             }
-            final CsvCRowSerializationSchema.Builder serSchemaBuilder = new CsvCRowSerializationSchema.Builder(typeInformation);
+            final CsvTupleSerializationSchema.Builder serSchemaBuilder = new CsvTupleSerializationSchema.Builder(typeInformation);
             serSchemaBuilder.setFieldDelimiter(kafkaSinkTableInfo.getFieldDelimiter().toCharArray()[0]);
             serSchemaBuilder.setUpdateMode(kafkaSinkTableInfo.getUpdateMode());
 
@@ -81,7 +83,7 @@ private SerializationSchema<CRow> createSerializationSchema(KafkaSinkTableInfo k
             if (StringUtils.isBlank(kafkaSinkTableInfo.getSchemaString())) {
                 throw new IllegalArgumentException("sinkDataType:" + FormatType.AVRO.name() + " must set schemaString");
             }
-            serializationSchema = new AvroCRowSerializationSchema(kafkaSinkTableInfo.getSchemaString(), kafkaSinkTableInfo.getUpdateMode());
+            serializationSchema = new AvroTuple2SerializationSchema(kafkaSinkTableInfo.getSchemaString(), kafkaSinkTableInfo.getUpdateMode());
         }
 
         if (null == serializationSchema) {