Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java

@@ -161,6 +161,12 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
         //  @wenbaoup fix bug
         inputParams.forEach((key, value) -> {
             Object transformValue = transformValue(value);
+            if (transformValue == null) {
+                scannerBuilder.addPredicate(
+                       KuduPredicate.newIsNullPredicate(schema.getColumn(key))
+                );
+                return;
+            }
             scannerBuilder.addPredicate(
                 KuduPredicate.newInListPredicate(
                     schema.getColumn(key),
@@ -228,7 +234,7 @@ private Object transformValue(Object value) {
     public String buildCacheKey(Map<String, Object> inputParams) {
         StringBuilder sb = new StringBuilder();
         for (Object ele : inputParams.values()) {
-            sb.append(ele.toString())
+            sb.append(ele)
                     .append("_");
         }
 
@@ -295,6 +301,11 @@ public GetListRowCB() {
 
         @Override
         public Deferred<List<Row>> call(RowResultIterator results) throws Exception {
+            if (results == null) {
+                resultFuture.complete(Collections.emptyList());
+                return null;
+            }
+
             for (RowResult result : results) {
                 Map<String, Object> oneRow = Maps.newHashMap();
                 for (String sideFieldName1 : StringUtils.split(sideInfo.getSideSelectFields(), ",")) {
@@ -327,6 +338,7 @@ public Deferred<List<Row>> call(RowResultIterator results) throws Exception {
                 }
             }
 
+            resultFuture.complete(Collections.emptyList());
             return null;
         }
     }