Merge branch 'hotfix_1.10_4.0.x_30906' into 1.10_release_4.0.x

Author: gituser

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

@@ -21,6 +21,9 @@
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.BaseAsyncReqRow;
 import com.dtstack.flink.sql.util.RowDataComplete;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
@@ -84,35 +87,33 @@ public void open(Configuration parameters) throws Exception {
     private void buildRedisClient(RedisSideTableInfo tableInfo){
         String url = redisSideTableInfo.getUrl();
         String password = redisSideTableInfo.getPassword();
-        if (password != null){
-            password = password + "@";
-        } else {
-            password = "";
-        }
+
         String database = redisSideTableInfo.getDatabase();
         if (database == null){
             database = "0";
         }
         switch (RedisType.parse(tableInfo.getRedisType())){
             case STANDALONE:
-                StringBuilder redisUri = new StringBuilder();
-                redisUri.append("redis://").append(password).append(url).append("/").append(database);
-                redisClient = RedisClient.create(redisUri.toString());
+                RedisURI redisURI = RedisURI.create("redis://" + url);
+                redisURI.setPassword(password);
+                redisURI.setDatabase(Integer.valueOf(database));
+                redisClient = RedisClient.create(redisURI);
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
             case SENTINEL:
-                StringBuilder sentinelUri = new StringBuilder();
-                sentinelUri.append("redis-sentinel://").append(password)
-                        .append(url).append("/").append(database).append("#").append(redisSideTableInfo.getMasterName());
-                redisClient = RedisClient.create(sentinelUri.toString());
+                RedisURI redisSentinelURI = RedisURI.create("redis-sentinel://" + url);
+                redisSentinelURI.setPassword(password);
+                redisSentinelURI.setDatabase(Integer.valueOf(database));
+                redisSentinelURI.setSentinelMasterId(redisSideTableInfo.getMasterName());
+                redisClient = RedisClient.create(redisSentinelURI);
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
             case CLUSTER:
-                StringBuilder clusterUri = new StringBuilder();
-                clusterUri.append("redis://").append(password).append(url);
-                clusterClient = RedisClusterClient.create(clusterUri.toString());
+                RedisURI clusterURI = RedisURI.create("redis://" + url);
+                clusterURI.setPassword(password);
+                clusterClient = RedisClusterClient.create(clusterURI);
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
             default: