[flinksql][kafka sink没有将Retract流中的false数据过滤掉][21182]

Author: simenliuxing

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

@@ -113,9 +113,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 serializationSchema
         );
 
-        DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
-            return record.f1;
-        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
+        DataStream<Row> ds = dataStream
+                .filter((Tuple2<Boolean, Row> record) -> record.f0)
+                .map((Tuple2<Boolean, Row> record) -> {return record.f1;})
+                .returns(getOutputType().getTypeAt(1))
+                .setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }

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

@@ -117,9 +117,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 				serializationSchema
 		);
 
-		DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
-			return record.f1;
-		}).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
+		DataStream<Row> ds = dataStream
+				.filter((Tuple2<Boolean, Row> record) -> record.f0)
+				.map((Tuple2<Boolean, Row> record) -> {return record.f1;})
+				.returns(getOutputType().getTypeAt(1))
+				.setParallelism(parallelism);
 
 		kafkaTableSink.emitDataStream(ds);
 	}

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

@@ -120,9 +120,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 serializationSchema
         );
 
-        DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
-            return record.f1;
-        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
+        DataStream<Row> ds = dataStream
+                .filter((Tuple2<Boolean, Row> record) -> record.f0)
+                .map((Tuple2<Boolean, Row> record) -> {return record.f1;})
+                .returns(getOutputType().getTypeAt(1))
+                .setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }

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

@@ -119,9 +119,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 serializationSchema
         );
 
-        DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
-            return record.f1;
-        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
+        DataStream<Row> ds = dataStream
+                .filter((Tuple2<Boolean, Row> record) -> record.f0)
+                .map((Tuple2<Boolean, Row> record) -> {return record.f1;})
+                .returns(getOutputType().getTypeAt(1))
+                .setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }