[opt-34810][hbase][dirty] 调整hbase batch 提交脏数据处理逻辑；优化脏数据日志打印内容

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyDataManager.java

@@ -129,15 +129,17 @@ public void close() {
     public void collectDirtyData(String dataInfo, String cause) {
         DirtyDataEntity dirtyDataEntity = new DirtyDataEntity(dataInfo, System.currentTimeMillis(), cause);
         try {
-            consumer.collectDirtyData(dirtyDataEntity, blockingInterval);
             count.incrementAndGet();
-        } catch (Exception ignored) {
+            consumer.collectDirtyData(dirtyDataEntity, blockingInterval);
+        } catch (Exception e) {
             LOG.warn("dirty Data insert error ... Failed number: " + errorCount.incrementAndGet());
-            LOG.warn("error dirty data:" + dirtyDataEntity.toString());
+            LOG.warn("error cause: " + e.getMessage());
+            LOG.warn("error dirty data:" + dirtyDataEntity.getDirtyData());
             if (errorCount.get() > Math.ceil(count.longValue() * errorLimitRate)) {
                 // close consumer and manager
                 close();
-                throw new RuntimeException(String.format("The number of failed number 【%s】 reaches the limit, manager fails", errorCount.get()));
+                throw new RuntimeException(
+                        String.format("The number of failed number 【%s】 reaches the limit, manager fails", errorCount.get()));
             }
         }
     }

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

@@ -227,27 +227,26 @@ protected synchronized void dealBatchOperation(List<Row> records) {
             }
             table.batch(puts, results);
 
+            // 打印结果
+            if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
+                // 只打印最后一条数据
+                LOG.info(records.get(records.size() - 1).toString());
+            }
+        } catch (IOException | InterruptedException e) {
             // 判断数据是否插入成功
             for (int i = 0; i < results.length; i++) {
-                if (results[i] == null) {
+                if (results[i] != null) {
                     // 脏数据记录
                     dirtyDataManager.collectDirtyData(
                             records.get(i).toString(),
-                            "Batch Hbase Sink Error"
+                            ((Exception) results[i]).getMessage()
                     );
                     outDirtyRecords.inc();
                 } else {
                     // 输出结果条数记录
                     outRecords.inc();
                 }
             }
-            // 打印结果
-            if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
-                // 只打印最后一条数据
-                LOG.info(records.get(records.size() - 1).toString());
-            }
-        } catch (IOException | InterruptedException e) {
-            LOG.error("", e);
         } finally {
             // 添加完数据之后数据清空records
             records.clear();