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

Author: gituser

redis5/pom.xml

@@ -15,6 +15,18 @@
         <module>redis5-sink</module>
         <module>redis5-side</module>
     </modules>
-
+    <dependencies>
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.core</artifactId>
+            <version>1.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+            <version>2.9.0</version>
+        </dependency>
+    </dependencies>
 
 </project>

redis5/redis5-side/pom.xml

@@ -11,20 +11,6 @@
     <modelVersion>4.0.0</modelVersion>
     <artifactId>sql.side.redis</artifactId>
     <name>redis-side</name>
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>redis.clients</groupId>
-            <artifactId>jedis</artifactId>
-            <version>2.8.0</version>
-        </dependency>
-    </dependencies>
-
 
     <modules>
         <module>redis-side-core</module>

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

@@ -30,7 +30,6 @@
 import com.google.common.collect.Maps;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
@@ -48,20 +47,27 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.Calendar;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 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 +78,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));
@@ -112,9 +118,9 @@ protected void reloadCache() {
     public void flatMap(BaseRow input, Collector<BaseRow> out) throws Exception {
         GenericRow genericRow = (GenericRow) input;
         Map<String, Object> inputParams = Maps.newHashMap();
-        for(Integer conValIndex : sideInfo.getEqualValIndex()){
+        for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = genericRow.getField(conValIndex);
-            if(equalObj == null){
+            if (equalObj == null) {
                 if (sideInfo.getJoinType() == JoinType.LEFT) {
                     BaseRow data = fillData(input, null);
                     RowDataComplete.collectBaseRow(out, data);
@@ -128,11 +134,11 @@ public void flatMap(BaseRow input, Collector<BaseRow> out) throws Exception {
 
         Map<String, String> cacheMap = cacheRef.get().get(key);
 
-        if (cacheMap == null){
-            if(sideInfo.getJoinType() == JoinType.LEFT){
+        if (cacheMap == null) {
+            if (sideInfo.getJoinType() == JoinType.LEFT) {
                 BaseRow data = fillData(input, null);
                 RowDataComplete.collectBaseRow(out, data);
-            }else{
+            } else {
                 return;
             }
 
@@ -147,7 +153,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);
@@ -182,41 +188,54 @@ private JedisCommands getJedis(RedisSideTableInfo tableInfo) {
         String url = tableInfo.getUrl();
         String password = tableInfo.getPassword();
         String database = tableInfo.getDatabase() == null ? "0" : tableInfo.getDatabase();
+        String masterName = tableInfo.getMasterName();
         int timeout = tableInfo.getTimeout();
-        if (timeout == 0){
-            timeout = 1000;
+        if (timeout == 0) {
+            timeout = 10000;
         }
 
-        String[] nodes = StringUtils.split(url, ",");
-        String[] firstIpPort = StringUtils.split(nodes[0], ":");
-        String firstIp = firstIpPort[0];
-        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;
-        }
+        String[] nodes = url.split(",");
         JedisCommands jedis = null;
         GenericObjectPoolConfig poolConfig = setPoolConfig(tableInfo.getMaxTotal(), tableInfo.getMaxIdle(), tableInfo.getMinIdle());
-        switch (RedisType.parse(tableInfo.getRedisType())){
-            //单机
+        switch (RedisType.parse(tableInfo.getRedisType())) {
             case STANDALONE:
-                pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
-                jedis = pool.getResource();
+                String firstIp = null;
+                String firstPort = null;
+                Matcher standalone = HOST_PORT_PATTERN.matcher(nodes[0]);
+                if (standalone.find()) {
+                    firstIp = standalone.group("host").trim();
+                    firstPort = standalone.group("port").trim();
+                }
+                if (Objects.nonNull(firstIp)) {
+                    pool = new JedisPool(poolConfig, firstIp, Integer.parseInt(firstPort), timeout, password, Integer.parseInt(database));
+                    jedis = pool.getResource();
+                } else {
+                    throw new IllegalArgumentException(
+                            String.format("redis url error. current url [%s]", nodes[0]));
+                }
                 break;
-            //哨兵
             case SENTINEL:
-                jedisSentinelPool = new JedisSentinelPool(tableInfo.getMasterName(), ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
+                Set<String> ipPorts = new HashSet<>(Arrays.asList(nodes));
+                jedisSentinelPool = new JedisSentinelPool(masterName, ipPorts, poolConfig, timeout, password, Integer.parseInt(database));
                 jedis = jedisSentinelPool.getResource();
                 break;
-            //集群
             case CLUSTER:
-                jedis = new JedisCluster(addresses, timeout, timeout,1, poolConfig);
+                Set<HostAndPort> addresses = new HashSet<>();
+                // 对ipv6 支持
+                for (String node : nodes) {
+                    Matcher matcher = HOST_PORT_PATTERN.matcher(node);
+                    if (matcher.find()) {
+                        String host = matcher.group("host").trim();
+                        String portStr = matcher.group("port").trim();
+                        if (org.apache.commons.lang3.StringUtils.isNotBlank(host) && org.apache.commons.lang3.StringUtils.isNotBlank(portStr)) {
+                            // 转化为int格式的端口
+                            int port = Integer.parseInt(portStr);
+                            addresses.add(new HostAndPort(host, port));
+                        }
+                    }
+                }
+                jedis = new JedisCluster(addresses, timeout, timeout, 10, password, poolConfig);
+                break;
             default:
                 break;
         }
@@ -244,35 +263,32 @@ private JedisCommands getJedisWithRetry(int retryNum) {
         return null;
     }
 
-    private Set<String> getRedisKeys(RedisType redisType, JedisCommands jedis, String keyPattern){
-        if(!redisType.equals(RedisType.CLUSTER)){
+    private Set<String> getRedisKeys(RedisType redisType, JedisCommands jedis, String keyPattern) {
+        if (!redisType.equals(RedisType.CLUSTER)) {
             return ((Jedis) jedis).keys(keyPattern);
         }
         Set<String> keys = new TreeSet<>();
-        Map<String, JedisPool> clusterNodes = ((JedisCluster)jedis).getClusterNodes();
-        for(String k : clusterNodes.keySet()){
+        Map<String, JedisPool> clusterNodes = ((JedisCluster) jedis).getClusterNodes();
+        for (String k : clusterNodes.keySet()) {
             JedisPool jp = clusterNodes.get(k);
-            Jedis connection = jp.getResource();
-            try {
+            try (Jedis connection = jp.getResource()) {
                 keys.addAll(connection.keys(keyPattern));
-            } catch (Exception e){
-                LOG.error("Getting keys error: {}", e);
-            } finally {
-                connection.close();
+            } catch (Exception e) {
+                LOG.error("Getting keys error", e);
             }
         }
         return keys;
     }
 
-    private GenericObjectPoolConfig setPoolConfig(String maxTotal, String maxIdle, String minIdle){
+    private GenericObjectPoolConfig setPoolConfig(String maxTotal, String maxIdle, String minIdle) {
         GenericObjectPoolConfig config = new GenericObjectPoolConfig();
-        if (maxTotal != null){
+        if (maxTotal != null) {
             config.setMaxTotal(Integer.parseInt(maxTotal));
         }
-        if (maxIdle != null){
+        if (maxIdle != null) {
             config.setMaxIdle(Integer.parseInt(maxIdle));
         }
-        if (minIdle != null){
+        if (minIdle != null) {
             config.setMinIdle(Integer.parseInt(minIdle));
         }
         return config;

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

@@ -69,7 +69,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));
@@ -95,7 +95,7 @@ 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();
@@ -167,21 +167,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());
             }
         });
     }

redis5/redis5-side/redis-side-core/pom.xml

@@ -12,14 +12,7 @@
     </parent>
 
     <artifactId>sql.side.redis.core</artifactId>
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-    </dependencies>
+
     <packaging>jar</packaging>
 
 

redis5/redis5-sink/pom.xml

@@ -16,20 +16,6 @@
     <name>redis-sink</name>
     <url>http://maven.apache.org</url>
 
-    <dependencies>
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.core</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>redis.clients</groupId>
-            <artifactId>jedis</artifactId>
-            <version>2.9.0</version>
-        </dependency>
-    </dependencies>
-
     <build>
         <plugins>
             <plugin>