sink name

Author: jiemotongxue

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

@@ -105,7 +105,7 @@ public TypeInformation<Row> getRecordType() {
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         DataStream<Row> mapDataStream = dataStream.map((Tuple2<Boolean, Row> record) -> record.f1).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
-        mapDataStream.addSink(flinkKafkaProducer).name(TableConnectorUtils.generateRuntimeName(FlinkKafkaProducer.class, getFieldNames()));
+        mapDataStream.addSink(flinkKafkaProducer).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
     }
 
     @Override

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

@@ -109,7 +109,7 @@ public TypeInformation<Row> getRecordType() {
 	@Override
 	public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 		DataStream<Row> mapDataStream = dataStream.map((Tuple2<Boolean, Row> record) -> record.f1).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
-		mapDataStream.addSink(kafkaProducer09).name(TableConnectorUtils.generateRuntimeName(FlinkKafkaProducer09.class, getFieldNames()));
+		mapDataStream.addSink(kafkaProducer09).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
 	}
 
 	@Override

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

@@ -112,7 +112,7 @@ public TypeInformation<Row> getRecordType() {
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         DataStream<Row> mapDataStream = dataStream.map((Tuple2<Boolean, Row> record) -> record.f1).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
-        mapDataStream.addSink(kafkaProducer010).name(TableConnectorUtils.generateRuntimeName(FlinkKafkaProducer010.class, getFieldNames()));
+        mapDataStream.addSink(kafkaProducer010).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
     }
 
     @Override

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

@@ -111,7 +111,7 @@ public TypeInformation<Row> getRecordType() {
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         DataStream<Row> mapDataStream = dataStream.map((Tuple2<Boolean, Row> record) -> record.f1).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
-        mapDataStream.addSink(kafkaProducer011).name(TableConnectorUtils.generateRuntimeName(FlinkKafkaProducer011.class, getFieldNames()));
+        mapDataStream.addSink(kafkaProducer011).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
     }
 
     @Override