[feat-34267][redis] redis side support ipv6.

Author: FlechazoW

redis5/redis5-side/redis-all-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAllReqRow.java

@@ -55,13 +55,18 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
 /**
  * @author yanxi
  */
 public class RedisAllReqRow extends BaseAllReqRow {
 
     private static final long serialVersionUID = 7578879189085344807L;
 
+    private static final Pattern HOST_PORT_PATTERN = Pattern.compile("(?<host>(.*)):(?<port>\\d+)*");
+
     private static final Logger LOG = LoggerFactory.getLogger(RedisAllReqRow.class);
 
     private static final int CONN_RETRY_NUM = 3;
@@ -72,9 +77,9 @@ public class RedisAllReqRow extends BaseAllReqRow {
 
     private RedisSideTableInfo tableInfo;
 
-    private AtomicReference<Map<String, Map<String, String>>> cacheRef = new AtomicReference<>();
+    private final AtomicReference<Map<String, Map<String, String>>> cacheRef = new AtomicReference<>();
 
-    private RedisSideReqRow redisSideReqRow;
+    private final RedisSideReqRow redisSideReqRow;
 
     public RedisAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(new RedisAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
@@ -147,7 +152,7 @@ private void loadData(Map<String, Map<String, String>> tmpCache) throws SQLExcep
         JedisCommands jedis = null;
         try {
             StringBuilder keyPattern = new StringBuilder(tableInfo.getTableName());
-            for (String key : tableInfo.getPrimaryKeys()) {
+            for (int i = 0; i < tableInfo.getPrimaryKeys().size(); i++) {
                 keyPattern.append("_").append("*");
             }
             jedis = getJedisWithRetry(CONN_RETRY_NUM);
@@ -193,14 +198,22 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         String firstPort = firstIpPort[1];
         Set<HostAndPort> addresses = new HashSet<>();
         Set<String> ipPorts = new HashSet<>();
-        for (String ipPort : nodes) {
-            ipPorts.add(ipPort);
-            String[] ipPortPair = ipPort.split(":");
-            addresses.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
-        }
-        if (timeout == 0){
-            timeout = 1000;
+
+        // 对ipv6 支持
+        for (String node : nodes) {
+            ipPorts.add(node);
+            Matcher matcher = HOST_PORT_PATTERN.matcher(node);
+            if (matcher.find()) {
+                String host = matcher.group("host").trim();
+                String portStr = matcher.group("port").trim();
+                if (StringUtils.isNotBlank(host) && StringUtils.isNotBlank(portStr)) {
+                    // 转化为int格式的端口
+                    int port = Integer.parseInt(portStr);
+                    addresses.add(new HostAndPort(host, port));
+                }
+            }
         }
+
         JedisCommands jedis = null;
         GenericObjectPoolConfig poolConfig = setPoolConfig(tableInfo.getMaxTotal(), tableInfo.getMaxIdle(), tableInfo.getMinIdle());
         switch (RedisType.parse(tableInfo.getRedisType())){

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

@@ -66,7 +66,7 @@ public class RedisAsyncReqRow extends BaseAsyncReqRow {
 
     private RedisSideTableInfo redisSideTableInfo;
 
-    private RedisSideReqRow redisSideReqRow;
+    private final RedisSideReqRow redisSideReqRow;
 
     public RedisAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(new RedisAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
@@ -92,15 +92,15 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
             case STANDALONE:
                 RedisURI redisURI = RedisURI.create("redis://" + url);
                 redisURI.setPassword(password);
-                redisURI.setDatabase(Integer.valueOf(database));
+                redisURI.setDatabase(Integer.parseInt(database));
                 redisClient = RedisClient.create(redisURI);
                 connection = redisClient.connect();
                 async = connection.async();
                 break;
             case SENTINEL:
                 RedisURI redisSentinelURI = RedisURI.create("redis-sentinel://" + url);
                 redisSentinelURI.setPassword(password);
-                redisSentinelURI.setDatabase(Integer.valueOf(database));
+                redisSentinelURI.setDatabase(Integer.parseInt(database));
                 redisSentinelURI.setSentinelMasterId(redisSideTableInfo.getMasterName());
                 redisClient = RedisClient.create(redisSentinelURI);
                 connection = redisClient.connect();
@@ -129,21 +129,18 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
             return;
         }
         RedisFuture<Map<String, String>> future = ((RedisHashAsyncCommands) async).hgetall(key);
-        future.thenAccept(new Consumer<Map<String, String>>() {
-            @Override
-            public void accept(Map<String, String> values) {
-                if (MapUtils.isNotEmpty(values)) {
-                    try {
-                        BaseRow row = fillData(input, values);
-                        dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
-                        RowDataComplete.completeBaseRow(resultFuture, row);
-                    } catch (Exception e) {
-                        dealFillDataError(input, resultFuture, e);
-                    }
-                } else {
-                    dealMissKey(input, resultFuture);
-                    dealCacheData(key, CacheMissVal.getMissKeyObj());
+        future.thenAccept(values -> {
+            if (MapUtils.isNotEmpty(values)) {
+                try {
+                    BaseRow row = fillData(input, values);
+                    dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
+                    RowDataComplete.completeBaseRow(resultFuture, row);
+                } catch (Exception e) {
+                    dealFillDataError(input, resultFuture, e);
                 }
+            } else {
+                dealMissKey(input, resultFuture);
+                dealCacheData(key, CacheMissVal.getMissKeyObj());
             }
         });
     }