fix bug 29911

Author: simenliuxing

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.kafka;
 
+import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.kafka.table.KafkaSinkTableInfo;
 import org.apache.commons.lang3.StringUtils;
@@ -60,6 +61,7 @@ public abstract class AbstractKafkaSink implements RetractStreamTableSink<Row>,
     protected int parallelism;
     protected String topic;
     protected String tableName;
+    protected String updateMode;
 
     protected TableSchema schema;
     protected SinkFunction<Tuple2<Boolean,Row>> kafkaProducer011;
@@ -107,6 +109,9 @@ protected String[] getPartitionKeys(KafkaSinkTableInfo kafkaSinkTableInfo) {
 
     @Override
     public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
+        if (updateMode.equalsIgnoreCase(EUpdateMode.APPEND.name())) {
+            dataStream = dataStream.filter((Tuple2<Boolean, Row> record) -> record.f0);
+        }
         DataStreamSink<Tuple2<Boolean, Row>> dataStreamSink = dataStream.addSink(kafkaProducer011).name(sinkOperatorName);
         if (parallelism > 0) {
             dataStreamSink.setParallelism(parallelism);

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

@@ -36,6 +36,7 @@ public KafkaSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
 
         Properties kafkaProperties = getKafkaProperties(kafkaSinkTableInfo);
         this.tableName = kafkaSinkTableInfo.getName();
+        this.updateMode = kafkaSinkTableInfo.getUpdateMode();
         this.topic = kafkaSinkTableInfo.getTopic();
         this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafkaSinkTableInfo);

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -38,6 +38,7 @@ public KafkaSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
 
         Properties kafkaProperties = getKafkaProperties(kafka09SinkTableInfo);
         this.tableName = kafka09SinkTableInfo.getName();
+        this.updateMode = kafka09SinkTableInfo.getUpdateMode();
         this.topic = kafka09SinkTableInfo.getTopic();
         this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafka09SinkTableInfo);

kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -42,6 +42,7 @@ public KafkaSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
 
         Properties kafkaProperties = getKafkaProperties(kafka10SinkTableInfo);
         this.tableName = kafka10SinkTableInfo.getName();
+        this.updateMode = kafka10SinkTableInfo.getUpdateMode();
         this.topic = kafka10SinkTableInfo.getTopic();
         this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafka10SinkTableInfo);

kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -41,6 +41,7 @@ public KafkaSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
 
         Properties kafkaProperties = getKafkaProperties(kafka11SinkTableInfo);
         this.tableName = kafka11SinkTableInfo.getName();
+        this.updateMode = kafka11SinkTableInfo.getUpdateMode();
         this.topic = kafka11SinkTableInfo.getTopic();
         this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafka11SinkTableInfo);