fix redis 密码问题

Author: dapeng

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

@@ -21,6 +21,8 @@
 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.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,37 +86,36 @@ 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);
+                redisUri.append("redis://").append(url).append("/").append(database);
                 redisClient = RedisClient.create(redisUri.toString());
                 connection = redisClient.connect();
                 async = connection.async();
+                ((RedisAsyncCommands)async).auth(password);
                 break;
             case SENTINEL:
                 StringBuilder sentinelUri = new StringBuilder();
-                sentinelUri.append("redis-sentinel://").append(password)
+                sentinelUri.append("redis-sentinel://")
                         .append(url).append("/").append(database).append("#").append(redisSideTableInfo.getMasterName());
                 redisClient = RedisClient.create(sentinelUri.toString());
                 connection = redisClient.connect();
                 async = connection.async();
+                ((RedisAsyncCommands)async).auth(password);
                 break;
             case CLUSTER:
                 StringBuilder clusterUri = new StringBuilder();
-                clusterUri.append("redis://").append(password).append(url);
+                clusterUri.append("redis://").append(url);
                 clusterClient = RedisClusterClient.create(clusterUri.toString());
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
+                ((RedisAdvancedClusterAsyncCommands)async).auth(password);
             default:
                 break;
         }