[hotfix-36143][hbase]修复hbase 写入数据全为null时，table put 异常

Author: FlechazoW

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

@@ -140,9 +140,12 @@ public static String buildDefaultDirty() {
      * 脏数据收集任务停止，任务停止之前，需要将队列中所有的数据清空
      */
     public void close() {
-        if (checkConsumer()) {
+        if (consumer != null && checkConsumer()) {
             LOG.info("dirty consumer is closing ...");
             consumer.close();
+        }
+
+        if (dirtyDataConsumer != null) {
             dirtyDataConsumer.shutdownNow();
         }
     }

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

@@ -49,6 +49,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -223,12 +224,23 @@ public void writeBatchRecord(Row row) {
 
     protected synchronized void dealBatchOperation(List<Row> records) {
         // A null in the result array means that the call for that action failed, even after retries.
-        Object[] results = new Object[records.size()];
+        Object[] results = null;
         try {
             List<Put> puts = new ArrayList<>();
             for (Row record : records) {
-                puts.add(getPutByRow(record));
+                Put put = getPutByRow(record);
+                if (put == null || put.isEmpty()) {
+                    dirtyDataManager.execute();
+                    dirtyDataManager.collectDirtyData(
+                        record.toString(),
+                        "HBase put is empty, check record please!"
+                    );
+                    outDirtyRecords.inc();
+                } else {
+                    puts.add(put);
+                }
             }
+            results = new Object[puts.size()];
             table.batch(puts, results);
 
             // 打印结果
@@ -240,7 +252,7 @@ protected synchronized void dealBatchOperation(List<Row> records) {
             // ignore exception
         } finally {
             // 判断数据是否插入成功
-            for (int i = 0; i < results.length; i++) {
+            for (int i = 0; i < Objects.requireNonNull(results).length; i++) {
                 if (results[i] instanceof Exception) {
                     dirtyDataManager.execute();
                     // 脏数据记录