Merge branch 'feat_1.10_replaceCompleteNull' into 'v1.10.0_dev'

[feat] replace complete(null) 为 complete(Collections.EMPTY_LIST);

See merge request dt-insight-engine/flinkStreamSQL!78

Author: FlechazoW

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -211,7 +211,7 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                     if (openCache()) {
                         putCache(key, CacheMissVal.getMissKeyObj());
                     }
-                    resultFuture.complete(null);
+                    resultFuture.complete(Collections.EMPTY_LIST);
                 }
             }
 

core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java

@@ -145,7 +145,7 @@ protected void dealMissKey(Row input, ResultFuture<BaseRow> resultFuture) {
                 dealFillDataError(input, resultFuture, e);
             }
         } else {
-            resultFuture.complete(null);
+            resultFuture.complete(Collections.EMPTY_LIST);
         }
     }
 
@@ -163,14 +163,14 @@ public void timeout(Row input, ResultFuture<BaseRow> resultFuture) throws Except
         }
         timeOutNum++;
         if (sideInfo.getJoinType() == JoinType.LEFT) {
-            resultFuture.complete(null);
+            resultFuture.complete(Collections.EMPTY_LIST);
             return;
         }
         if (timeOutNum > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)) {
             resultFuture.completeExceptionally(new Exception("Async function call timedoutNum beyond limit."));
             return;
         }
-        resultFuture.complete(null);
+        resultFuture.complete(Collections.EMPTY_LIST);
     }
 
     protected void preInvoke(Row input, ResultFuture<BaseRow> resultFuture)

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/AbstractRowKeyModeDealer.java

@@ -85,7 +85,7 @@ protected void dealMissKey(Row input, ResultFuture<BaseRow> resultFuture) {
                 resultFuture.completeExceptionally(e);
             }
         } else {
-            resultFuture.complete(null);
+            resultFuture.complete(Collections.EMPTY_LIST);
         }
     }
 

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/PreRowKeyModeDealerDealer.java

@@ -130,7 +130,7 @@ private String dealOneRow(ArrayList<ArrayList<KeyValue>> args, String rowKeyStr,
                     resultFuture.completeExceptionally(e);
                 }
             } catch (Exception e) {
-                resultFuture.complete(null);
+                resultFuture.complete(Collections.EMPTY_LIST);
                 LOG.error("record:" + input);
                 LOG.error("get side record exception:", e);
             }
@@ -150,7 +150,7 @@ private String dealOneRow(ArrayList<ArrayList<KeyValue>> args, String rowKeyStr,
     private String dealFail(Object arg2, Row input, ResultFuture<BaseRow> resultFuture){
         LOG.error("record:" + input);
         LOG.error("get side record exception:" + arg2);
-        resultFuture.complete(null);
+        resultFuture.complete(Collections.EMPTY_LIST);
         return "";
     }
 }

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java

@@ -118,7 +118,7 @@ public void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFu
         }, arg2 -> {
             LOG.error("record:" + input);
             LOG.error("get side record exception:" + arg2);
-            resultFuture.complete(null);
+            resultFuture.complete(Collections.EMPTY_LIST);
             return "";
         });
     }

mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java

@@ -133,7 +133,7 @@ public void apply(final Document document) {
             public void onResult(final Void result, final Throwable t) {
                 if (atomicInteger.get() <= 0) {
                     LOG.warn("Cannot retrieve the data from the database");
-                    resultFuture.complete(null);
+                    resultFuture.complete(Collections.EMPTY_LIST);
                 } else {
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));