fix redis async cache error

Author: todd5167

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

@@ -45,9 +45,7 @@
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.HashBasedTable;
@@ -61,11 +59,12 @@
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
-import java.io.Serializable;
 import java.sql.Timestamp;
-import java.util.*;
-import java.util.stream.Collectors;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
 
 import static org.apache.calcite.sql.SqlKind.*;
 

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -135,17 +135,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         if(openCache()){
             CacheObj val = getFromCache(key);
             if(val != null){
-
                 if(ECacheContentType.MissVal == val.getType()){
                     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, val.getContent());
-                        rowList.add(row);
-                    }
-                    resultFuture.complete(rowList);
+                    Row row = fillData(input, val.getContent());
+                    resultFuture.complete(Collections.singleton(row));
                 }else{
                     RuntimeException exception = new RuntimeException("not support cache obj type " + val.getType());
                     resultFuture.completeExceptionally(exception);
@@ -171,10 +166,10 @@ public void accept(List<KeyValue<String, String>> keyValues) {
                             keyValue.put(splitKeys[3], keyValues.get(i).getValue());
                         }
                         Row row = fillData(input, keyValue);
-                        resultFuture.complete(Collections.singleton(row));
                         if (openCache()) {
                             putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, keyValue));
                         }
+                        resultFuture.complete(Collections.singleton(row));
                     } else {
                         dealMissKey(input, resultFuture);
                         if (openCache()) {