mongo rdb syncside bug fix

Author: todd5167

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

@@ -130,6 +130,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
                 resultFuture.complete(null);
+                return;
             }
             basicDBObject.put(sideInfo.getEqualFieldList().get(i), equalObj);
         }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -75,8 +75,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
                 resultFuture.complete(null);
+                return;
             }
-
             inputParams.add(equalObj);
         }
 
@@ -89,12 +89,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-
+                    List<Row> rowList = Lists.newArrayList();
                     for (Object jsonArray : (List) val.getContent()) {
                         Row row = fillData(input, jsonArray);
-                        resultFuture.complete(Collections.singleton(row));
+                        rowList.add(row);
                     }
-
+                    resultFuture.complete(rowList);
                 } else {
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }