Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

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

@@ -46,7 +46,10 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.Consumer;
+import java.util.regex.Matcher;
+
 /**
  * @author yanxi
  */
@@ -98,11 +101,34 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
                 async = connection.async();
                 break;
             case SENTINEL:
-                RedisURI redisSentinelURI = RedisURI.create("redis-sentinel://" + url);
-                redisSentinelURI.setPassword(password);
-                redisSentinelURI.setDatabase(Integer.valueOf(database));
-                redisSentinelURI.setSentinelMasterId(redisSideTableInfo.getMasterName());
-                redisClient = RedisClient.create(redisSentinelURI);
+                String[] urlSplit = StringUtils.split(url, ",");
+                RedisURI.Builder builder = null;
+                for (String item : urlSplit) {
+                    Matcher mather = RedisSideReqRow.HOST_PORT_PATTERN.matcher(item);
+                    if (mather.find()) {
+                        builder = buildSentinelUri(
+                                mather.group("host"),
+                                mather.group("port"),
+                                builder
+                        );
+                    } else {
+                        throw new IllegalArgumentException(
+                                String.format("Illegal format with redis url [%s]", item)
+                        );
+                    }
+                }
+
+                if (Objects.nonNull(builder)) {
+                    builder
+                            .withPassword(tableInfo.getPassword())
+                            .withDatabase(Integer.parseInt(tableInfo.getDatabase()))
+                            .withSentinelMasterId(tableInfo.getMasterName());
+                } else {
+                    throw new NullPointerException("build redis uri error!");
+                }
+
+                RedisURI uri = builder.build();
+                redisClient = RedisClient.create(uri);
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
@@ -117,6 +143,18 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
         }
     }
 
+    private RedisURI.Builder buildSentinelUri(
+            String host,
+            String port,
+            RedisURI.Builder builder) {
+        if (Objects.nonNull(builder)) {
+            builder.withSentinel(host, Integer.parseInt(port));
+        } else {
+            builder = RedisURI.Builder.sentinel(host, Integer.parseInt(port));
+        }
+        return builder;
+    }
+
     @Override
     public BaseRow fillData(BaseRow input, Object sideInput) {
         return redisSideReqRow.fillData(input, sideInput);

redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java

@@ -32,6 +32,7 @@
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 /**
  * redis fill row data
@@ -44,6 +45,8 @@ public class RedisSideReqRow implements ISideReqRow, Serializable {
 
     private static final long serialVersionUID = 3751171828444748982L;
 
+    public static final Pattern HOST_PORT_PATTERN = Pattern.compile("(?<host>(.*)):(?<port>\\d+)*");
+
     private BaseSideInfo sideInfo;
 
     private RedisSideTableInfo sideTableInfo;