fix 版本不兼容的问题

Author: dapeng

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

@@ -21,6 +21,7 @@
 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;
@@ -93,29 +94,28 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
         }
         switch (RedisType.parse(tableInfo.getRedisType())){
             case STANDALONE:
-                StringBuilder redisUri = new StringBuilder();
-                redisUri.append("redis://").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();
-                ((RedisAsyncCommands)async).auth(password);
                 break;
             case SENTINEL:
-                StringBuilder sentinelUri = new StringBuilder();
-                sentinelUri.append("redis-sentinel://")
-                        .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();
-                ((RedisAsyncCommands)async).auth(password);
                 break;
             case CLUSTER:
-                StringBuilder clusterUri = new StringBuilder();
-                clusterUri.append("redis://").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();
-                ((RedisAdvancedClusterAsyncCommands)async).auth(password);
             default:
                 break;
         }