[fix-34810][core] console dirty plugin 调整打印初始值，打印第一条脏数据信息

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/AbstractDirtyDataConsumer.java

@@ -83,8 +83,7 @@ public void run() {
             }
         } catch (Exception e) {
             LOG.error("consume dirtyData error", e);
-            errorCount.incrementAndGet();
-            if (errorCount.get() > errorLimit) {
+            if (errorCount.getAndIncrement() > errorLimit) {
                 throw new RuntimeException("The task failed due to the number of dirty data consume failed reached the limit " + errorLimit);
             }
         }

dirtyData/console/src/main/java/com/dtstack/flink/sql/dirty/console/ConsoleDirtyDataConsumer.java

@@ -43,8 +43,7 @@ public class ConsoleDirtyDataConsumer extends AbstractDirtyDataConsumer {
     @Override
     public void consume() throws InterruptedException {
         DirtyDataEntity dataEntity = queue.take();
-        count.incrementAndGet();
-        if (count.get() % printLimit == 0) {
+        if (count.getAndIncrement() % printLimit == 0) {
             LOG.warn("\nget dirtyData: " + dataEntity.getDirtyData() + "\n"
                     + "cause: " + dataEntity.getCause() + "\n"
                     + "processTime: " + dataEntity.getProcessDate() + "\n"

dirtyData/mysql/src/main/java/com/dtstack/flink/sql/dirty/mysql/MysqlDirtyDataConsumer.java

@@ -131,6 +131,7 @@ public void consume() throws Exception {
         statement.addBatch();
 
         if (count.get() % batchSize == 0) {
+            LOG.warn("Get dirty Data: " + entity.getDirtyData());
             statement.executeBatch();
         }
     }

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -19,9 +19,8 @@
 
 package com.dtstack.flink.sql.sink.hbase;
 
-import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.dirtyManager.manager.DirtyDataManager;
-import com.dtstack.flink.sql.enums.EUpdateMode;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
@@ -202,7 +201,7 @@ private void openKerberosConn() throws Exception {
     @Override
     public void writeRecord(Tuple2<Boolean, Row> record) {
         if (record.f0) {
-            if (this.batchSize != 0) {
+            if (this.batchSize > 1) {
                 writeBatchRecord(record.f1);
             } else {
                 dealInsert(record.f1);
@@ -231,10 +230,11 @@ protected synchronized void dealBatchOperation(List<Row> records) {
             // 判断数据是否插入成功
             for (int i = 0; i < results.length; i++) {
                 if (results[i] == null) {
-                    if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0 || LOG.isDebugEnabled()) {
-                        LOG.error("record insert failed ..{}", records.get(i).toString());
-                    }
                     // 脏数据记录
+                    dirtyDataManager.collectDirtyData(
+                            records.get(i).toString(),
+                            "Batch Hbase Sink Error"
+                    );
                     outDirtyRecords.inc();
                 } else {
                     // 输出结果条数记录