bugfix sink initMetric

Author: jiemotongxue

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

@@ -39,7 +39,6 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafkaSinkTableInfo.setName(tableName);
         kafkaSinkTableInfo.setType(MathUtil.getString(props.get(KafkaSinkTableInfo.TYPE_KEY.toLowerCase())));
         parseFieldsInfo(fieldsInfo, kafkaSinkTableInfo);
-        kafkaSinkTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase())));
 
         if (props.get(KafkaSinkTableInfo.SINK_DATA_TYPE) != null) {
             kafkaSinkTableInfo.setSinkDataType(props.get(KafkaSinkTableInfo.SINK_DATA_TYPE).toString());

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

@@ -48,6 +48,7 @@ public KafkaProducer(String topicId, SerializationSchema<Row> serializationSchem
     public void open(Configuration configuration) throws Exception {
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
+        serializationMetricWrapper.initMetric();
         super.open(configuration);
     }
 

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

@@ -48,6 +48,7 @@ public KafkaProducer09(String topicId, SerializationSchema<Row> serializationSch
     public void open(Configuration configuration) {
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
+        serializationMetricWrapper.initMetric();
         super.open(configuration);
     }
 

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

@@ -48,6 +48,7 @@ public KafkaProducer010(String topicId, SerializationSchema<Row> serializationSc
     public void open(Configuration configuration) {
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
+        serializationMetricWrapper.initMetric();
         super.open(configuration);
     }
 

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

@@ -49,6 +49,7 @@ public KafkaProducer011(String topicId, SerializationSchema<Row> serializationSc
     public void open(Configuration configuration) throws Exception {
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
+        serializationMetricWrapper.initMetric();
         super.open(configuration);
     }