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

Author: gituser

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java

@@ -71,7 +71,8 @@ public abstract class AbstractTableInfo implements Serializable {
      * error data limit. Task will failed once {@link AbstractDtRichOutputFormat#outDirtyRecords}
      * count over limit. Default 1000L;
      */
-    private Long errorLimit = 0L;
+    // TODO: 暂时默认值设置为10000，后面排期支持在产品上设置
+    private Long errorLimit = 10000L;
 
     public String[] getFieldTypes() {
         return fieldTypes;

hbase/hbase-side/hbase-all-side/pom.xml

@@ -33,6 +33,10 @@
                     <groupId>io.netty</groupId>
                     <artifactId>netty</artifactId>
                 </exclusion>
+                <exclusion>
+                    <artifactId>netty-all</artifactId>
+                    <groupId>io.netty</groupId>
+                </exclusion>
             </exclusions>
         </dependency>
     </dependencies>

hbase/pom.xml

@@ -38,8 +38,12 @@
                     <groupId>log4j</groupId>
                 </exclusion>
                 <exclusion>
+                    <artifactId>netty-all</artifactId>
                     <groupId>io.netty</groupId>
+                </exclusion>
+                <exclusion>
                     <artifactId>netty</artifactId>
+                    <groupId>io.netty</groupId>
                 </exclusion>
             </exclusions>
         </dependency>

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

@@ -37,17 +37,18 @@
 import io.lettuce.core.api.async.RedisKeyAsyncCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import io.lettuce.core.internal.HostAndPort;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.dataformat.BaseRow;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.Consumer;
 import java.util.regex.Matcher;
 
 /**
@@ -133,16 +134,34 @@ private void buildRedisClient(RedisSideTableInfo tableInfo){
                 async = connection.async();
                 break;
             case CLUSTER:
-                RedisURI clusterURI = RedisURI.create("redis://" + url);
-                clusterURI.setPassword(password);
-                clusterClient = RedisClusterClient.create(clusterURI);
+                List<RedisURI> clusterURIs = buildClusterURIs(url);
+                clusterClient = RedisClusterClient.create(clusterURIs);
                 clusterConnection = clusterClient.connect();
                 async = clusterConnection.async();
             default:
                 break;
         }
     }
 
+    private List<RedisURI> buildClusterURIs(String url) {
+        String password = redisSideTableInfo.getPassword();
+        String database = redisSideTableInfo.getDatabase();
+        String[] addresses = StringUtils.split(url, ",");
+        List<RedisURI> redisURIs = new ArrayList<>(addresses.length);
+        for (String addr : addresses){
+            HostAndPort hostAndPort = HostAndPort.parse(addr);
+            RedisURI redisURI = RedisURI.create(hostAndPort.hostText, hostAndPort.port);
+            if (StringUtils.isNotEmpty(password)) {
+                redisURI.setPassword(password);
+            }
+            if (StringUtils.isNotEmpty(database)) {
+                redisURI.setDatabase(Integer.valueOf(database));
+            }
+            redisURIs.add(redisURI);
+        }
+        return redisURIs;
+    }
+
     private RedisURI.Builder buildSentinelUri(
             String host,
             String port,
@@ -171,7 +190,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
             if (MapUtils.isNotEmpty(values)) {
                 try {
                     BaseRow row = fillData(input, values);
-                    dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.SingleLine, row));
+                    dealCacheData(key, CacheObj.buildCacheObj(ECacheContentType.SingleLine, values));
                     RowDataComplete.completeBaseRow(resultFuture, row);
                 } catch (Exception e) {
                     dealFillDataError(input, resultFuture, e);