Merge branch 'v1.8.0_dev_kudu' into 'v1.8.0_dev'

解决kudu 结果表没有Metric问题

解决kudu 结果表没有Metric问题

See merge request !144

Author: simenliuxing

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java

@@ -9,6 +9,7 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.table.sinks.RetractStreamTableSink;
@@ -63,7 +64,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 .setFieldTypes(this.fieldTypes);
         KuduOutputFormat kuduOutputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(kuduOutputFormat);
-        dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        dataStreamSink.name(tableName);
+        if (parallelism > 0) {
+            dataStreamSink.setParallelism(parallelism);
+        }
     }
 
     @Override