fix rdb asynside bug

Author: todd5167

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

@@ -26,19 +26,15 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.ext.sql.SQLClient;
 import io.vertx.ext.sql.SQLConnection;
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.math.BigInteger;
 import java.sql.Timestamp;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -122,18 +118,21 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
 
                 int resultSize = rs.result().getResults().size();
                 if (resultSize > 0) {
-                    for (JsonArray line : rs.result().getResults()) {
+                    List<Row> rowList = Lists.newArrayList();
 
+                    for (JsonArray line : rs.result().getResults()) {
                         Row row = fillData(input, line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
-                        resultFuture.complete(Collections.singleton(row));
+                        rowList.add(row);
                     }
 
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
+
+                    resultFuture.complete(rowList);
                 } else {
                     dealMissKey(input, resultFuture);
                     if (openCache()) {