@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);
        }
        return dataStreamSink;
    }
Date: 2020/3/30
Company: www.dtstack.com
@author maqi
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaSink.java