fix rdb asynside bug
@@ -26,19 +26,15 @@ import com.dtstack.flink.sql.side.rdb.util.SwitchUtil;
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 class RdbAsyncReqRow extends AsyncReqRow {
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()) {
