Merge branch '1.10_release_4.1.x' into hotfix_1.10_4.1.x_36409

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java

Author: xiuzhu9527

localTest/pom.xml

@@ -272,13 +272,7 @@
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
-            <version>1.2.29</version>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.alibaba</groupId>
-            <artifactId>fastjson</artifactId>
-            <version>1.2.70</version>
+            <version>1.2.76</version>
             <scope>compile</scope>
         </dependency>
 

pom.xml

@@ -99,7 +99,7 @@
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
-            <version>1.2.70</version>
+            <version>1.2.76</version>
         </dependency>
     </dependencies>
 

redis5/redis5-side/redis-async-side/pom.xml

@@ -35,14 +35,20 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>
-                <version>1.4</version>
+                <version>3.0.0</version>
                 <executions>
                     <execution>
                         <phase>package</phase>
                         <goals>
                             <goal>shade</goal>
                         </goals>
                         <configuration>
+                            <relocations>
+                                <relocation>
+                                    <pattern>io.netty</pattern>
+                                    <shadedPattern>com.dtstack.flink.sql.side.redis.io.netty</shadedPattern>
+                                </relocation>
+                            </relocations>
                             <artifactSet>
                                 <excludes>
 

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

@@ -37,17 +37,18 @@
 import io.lettuce.core.api.async.RedisKeyAsyncCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import io.lettuce.core.internal.HostAndPort;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.dataformat.BaseRow;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.Consumer;
 import java.util.regex.Matcher;
 
 /**
@@ -133,16 +134,34 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
                 async = connection.async();
                 break;
             case CLUSTER:
-                RedisURI clusterURI = RedisURI.create("redis://" + url);
-                clusterURI.setPassword(password);
-                clusterClient = RedisClusterClient.create(clusterURI);
+                List<RedisURI> clusterURIs = buildClusterURIs(url);
+                clusterClient = RedisClusterClient.create(clusterURIs);
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
             default:
                 break;
         }
     }
 
+    private List<RedisURI> buildClusterURIs(String url) {
+        String password = redisSideTableInfo.getPassword();
+        String database = redisSideTableInfo.getDatabase();
+        String[] addresses = StringUtils.split(url, ",");
+        List<RedisURI> redisURIs = new ArrayList<>(addresses.length);
+        for (String addr : addresses){
+            HostAndPort hostAndPort = HostAndPort.parse(addr);
+            RedisURI redisURI = RedisURI.create(hostAndPort.hostText, hostAndPort.port);
+            if (StringUtils.isNotEmpty(password)) {
+                redisURI.setPassword(password);
+            }
+            if (StringUtils.isNotEmpty(database)) {
+                redisURI.setDatabase(Integer.valueOf(database));
+            }
+            redisURIs.add(redisURI);
+        }
+        return redisURIs;
+    }
+
     private RedisURI.Builder buildSentinelUri(
             String host,
             String port,
@@ -171,7 +190,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
             if (MapUtils.isNotEmpty(values)) {
                 try {
                     BaseRow row = fillData(input, values);
-                    dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
+                    dealCacheData(key, CacheObj.buildCacheObj(ECacheContentType.SingleLine, values));
                     RowDataComplete.completeBaseRow(resultFuture, row);
                 } catch (Exception e) {
                     dealFillDataError(input, resultFuture, e);