Merge remote-tracking branch 'origin/hotfix_1.8_3.10.x_30426' into tmp_1.8_3.10.x_merge

# Conflicts:
#	flinkx-kudu/flinkx-kudu-writer/src/main/java/com/dtstack/flinkx/kudu/writer/KuduWriter.java

Author: yanghuaiGit

flinkx-emqx/flinkx-emqx-writer/src/main/java/com/dtstack/flinkx/emqx/writer/EmqxWriter.java

@@ -66,6 +66,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         builder.setPassword(password);
         builder.setCleanSession(isCleanSession);
         builder.setQos(qos);
+        builder.setDirtyPath(dirtyPath);
+        builder.setDirtyHadoopConfig(dirtyHadoopConfig);
+        builder.setSrcCols(srcCols);
         return createOutput(dataSet, builder.finish());
     }
 }

flinkx-kafka/flinkx-kafka-writer/src/main/java/com/dtstack/flinkx/kafka/writer/KafkaWriter.java

@@ -47,7 +47,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         format.setProducerSettings(producerSettings);
         format.setRestoreConfig(restoreConfig);
         format.setTableFields(tableFields);
-
+        format.setDirtyPath(dirtyPath);
+        format.setDirtyHadoopConfig(dirtyHadoopConfig);
+        format.setSrcFieldNames(srcCols);
         return createOutput(dataSet, format);
     }
 }

flinkx-kafka09/flinkx-kafka09-writer/src/main/java/com/dtstack/flinkx/kafka09/writer/Kafka09Writer.java

@@ -58,7 +58,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         format.setBrokerList(brokerList);
         format.setProducerSettings(producerSettings);
         format.setRestoreConfig(restoreConfig);
-
+        format.setDirtyPath(dirtyPath);
+        format.setDirtyHadoopConfig(dirtyHadoopConfig);
+        format.setSrcFieldNames(srcCols);
         return createOutput(dataSet, format);
     }
 }

flinkx-kafka10/flinkx-kafka10-writer/src/main/java/com/dtstack/flinkx/kafka10/writer/Kafka10Writer.java

@@ -46,7 +46,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         format.setTableFields(tableFields);
         format.setProducerSettings(producerSettings);
         format.setRestoreConfig(restoreConfig);
-
+        format.setDirtyPath(dirtyPath);
+        format.setDirtyHadoopConfig(dirtyHadoopConfig);
+        format.setSrcFieldNames(srcCols);
         return createOutput(dataSet, format);
     }
 }

flinkx-kafka11/flinkx-kafka11-writer/src/main/java/com/dtstack/flinkx/kafka11/writer/Kafka11Writer.java

@@ -46,7 +46,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         format.setProducerSettings(producerSettings);
         format.setRestoreConfig(restoreConfig);
         format.setTableFields(tableFields);
-
+        format.setDirtyPath(dirtyPath);
+        format.setDirtyHadoopConfig(dirtyHadoopConfig);
+        format.setSrcFieldNames(srcCols);
         return createOutput(dataSet, format);
     }
 }

flinkx-kb/flinkx-kb-writer/src/main/java/com/dtstack/flinkx/kafkabase/writer/KafkaBaseWriter.java

@@ -62,7 +62,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         format.setProducerSettings(producerSettings);
         format.setRestoreConfig(restoreConfig);
         format.setTableFields(tableFields);
-
+        format.setDirtyPath(dirtyPath);
+        format.setDirtyHadoopConfig(dirtyHadoopConfig);
+        format.setSrcFieldNames(srcCols);
         return createOutput(dataSet, format);
     }
 }

flinkx-kudu/flinkx-kudu-writer/src/main/java/com/dtstack/flinkx/kudu/writer/KuduWriter.java

@@ -95,6 +95,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         builder.setErrors(errors);
         builder.setErrorRatio(errorRatio);
         builder.setHadoopConfig(hadoopConfig);
+        builder.setDirtyPath(dirtyPath);
+        builder.setDirtyHadoopConfig(dirtyHadoopConfig);
+        builder.setSrcCols(srcCols);
         return createOutput(dataSet,builder.finish());
     }
 }

flinkx-restapi/flinkx-restapi-writer/src/main/java/com/dtstack/flinkx/restapi/writer/RestapiWriter.java

@@ -97,6 +97,9 @@ public DataStreamSink<?> writeData(DataStream<Row> dataSet) {
         builder.setColumn(column);
         builder.setParams(params);
         builder.setBatchInterval(batchInterval);
+        builder.setDirtyPath(dirtyPath);
+        builder.setDirtyHadoopConfig(dirtyHadoopConfig);
+        builder.setSrcCols(srcCols);
 
         return createOutput(dataSet, builder.finish());
     }