[hotfix-36409] 维表设置允许错误数限制不生效问题

xiuzhu9527 · xiuzhu9527 · commit e052b0b4fc40 · 2021-04-16T11:24:15.000+08:00
diff --git a/cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java b/cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java
@@ -91,6 +91,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         cassandraSideTableInfo.setConnectTimeoutMillis(MathUtil.getIntegerVal(props.get(CONNECT_TIMEOUT_MILLIS_KEY.toLowerCase())));
         cassandraSideTableInfo.setPoolTimeoutMillis(MathUtil.getIntegerVal(props.get(POOL_TIMEOUT_MILLIS_KEY.toLowerCase())));
 
+        if (MathUtil.getLongVal(props.get(cassandraSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            cassandraSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(cassandraSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         return cassandraSideTableInfo;
     }
 
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java b/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java
@@ -68,7 +68,7 @@ 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;
+    private Long errorLimit = Long.MAX_VALUE;
 
     public String[] getFieldTypes() {
         return fieldTypes;
diff --git a/elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/table/Elasticsearch6SideParser.java b/elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/table/Elasticsearch6SideParser.java
@@ -71,6 +71,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
             elasticsearch6SideTableInfo.setUserName(MathUtil.getString(props.get(KEY_ES6_USERNAME.toLowerCase())));
             elasticsearch6SideTableInfo.setPassword(MathUtil.getString(props.get(KEY_ES6_PASSWORD.toLowerCase())));
         }
+
+        if (MathUtil.getLongVal(props.get(elasticsearch6SideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            elasticsearch6SideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(elasticsearch6SideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         elasticsearch6SideTableInfo.check();
         return elasticsearch6SideTableInfo;
     }
diff --git a/elasticsearch7/elasticsearch7-side/elasticsearch7-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/table/Elasticsearch7SideParser.java b/elasticsearch7/elasticsearch7-side/elasticsearch7-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/table/Elasticsearch7SideParser.java
@@ -71,6 +71,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
             elasticsearch7SideTableInfo.setUserName(MathUtil.getString(props.get(KEY_ES7_USERNAME.toLowerCase())));
             elasticsearch7SideTableInfo.setPassword(MathUtil.getString(props.get(KEY_ES7_PASSWORD.toLowerCase())));
         }
+
+        if (MathUtil.getLongVal(props.get(elasticsearch7SideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            elasticsearch7SideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(elasticsearch7SideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         elasticsearch7SideTableInfo.check();
         return elasticsearch7SideTableInfo;
     }
diff --git a/hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/table/HbaseSideParser.java b/hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/table/HbaseSideParser.java
@@ -72,6 +72,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         hbaseTableInfo.setPreRowKey(MathUtil.getBoolean(props.get(PRE_ROW_KEY.toLowerCase()), false));
         hbaseTableInfo.setCacheType((String) props.get(CACHE));
         hbaseTableInfo.setKerberosAuthEnable(MathUtil.getBoolean(props.get(KERBEROS_ENABLE), false));
+
+        if (MathUtil.getLongVal(props.get(hbaseTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            hbaseTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(hbaseTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         props.entrySet().stream()
                 .filter(entity -> entity.getKey().contains("."))
                 .map(entity -> hbaseTableInfo.getHbaseConfig().put(entity.getKey(), String.valueOf(entity.getValue())))
diff --git a/impala/impala-side/impala-side-core/src/main/java/com/dtstack/flink/sql/side/impala/table/ImpalaSideParser.java b/impala/impala-side/impala-side-core/src/main/java/com/dtstack/flink/sql/side/impala/table/ImpalaSideParser.java
@@ -65,6 +65,9 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         impalaSideTableInfo.setFastCheck(MathUtil.getBoolean(props.getOrDefault(RdbSideTableInfo.FAST_CHECK.toLowerCase(), true)));
         impalaSideTableInfo.setCheckProperties();
 
+        if (MathUtil.getLongVal(props.get(impalaSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            impalaSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(impalaSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
 
         //set authmech params
         Integer authMech = MathUtil.getIntegerVal(props.get(ImpalaSideTableInfo.AUTHMECH_KEY.toLowerCase()));
diff --git a/kudu/kudu-side/kudu-side-core/src/main/java/com/dtstack/flink/sql/side/kudu/table/KuduSideParser.java b/kudu/kudu-side/kudu-side-core/src/main/java/com/dtstack/flink/sql/side/kudu/table/KuduSideParser.java
@@ -80,6 +80,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         );
         kuduSideTableInfo.judgeKrbEnable();
 
+        if (MathUtil.getLongVal(props.get(kuduSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            kuduSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(kuduSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         return kuduSideTableInfo;
 
     }
diff --git a/mongo/mongo-side/mongo-side-core/src/main/java/com/dtstack/flink/sql/side/mongo/table/MongoSideParser.java b/mongo/mongo-side/mongo-side-core/src/main/java/com/dtstack/flink/sql/side/mongo/table/MongoSideParser.java
@@ -59,6 +59,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         mongoSideTableInfo.setDatabase(MathUtil.getString(props.get(DATABASE_KEY.toLowerCase())));
         mongoSideTableInfo.setUserName(MathUtil.getString(props.get(USER_NAME_KEY.toLowerCase())));
         mongoSideTableInfo.setPassword(MathUtil.getString(props.get(PASSWORD_KEY.toLowerCase())));
+
+        if (MathUtil.getLongVal(props.get(mongoSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            mongoSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(mongoSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
         mongoSideTableInfo.check();
         return mongoSideTableInfo;
     }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideParser.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideParser.java
@@ -52,14 +52,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         rdbTableInfo.setSchema(MathUtil.getString(props.get(RdbSideTableInfo.SCHEMA_KEY.toLowerCase())));
         rdbTableInfo.setDriverName(MathUtil.getString(props.get(RdbSideTableInfo.DRIVER_NAME)));
         rdbTableInfo.setFastCheck(MathUtil.getBoolean(props.getOrDefault(RdbSideTableInfo.FAST_CHECK.toLowerCase(), false)));
-        rdbTableInfo.setErrorLimit(
-            MathUtil.getLongVal(
-                props.getOrDefault(
-                    RdbSideTableInfo.ERROR_LIMIT.toLowerCase(),
-                    1000L
-                )
-            )
-        );
+
+        if (MathUtil.getLongVal(props.get(RdbSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            rdbTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(RdbSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
 
         rdbTableInfo.setCheckProperties();
 
diff --git a/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideParser.java b/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideParser.java
@@ -51,6 +51,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         redisSideTableInfo.setMasterName(MathUtil.getString(props.get(RedisSideTableInfo.MASTER_NAME.toLowerCase())));
         redisSideTableInfo.setRedisType(MathUtil.getString(props.get(RedisSideTableInfo.REDIS_TYPE.toLowerCase())));
 
+        if (MathUtil.getLongVal(props.get(RedisSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {
+            redisSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(RedisSideTableInfo.ERROR_LIMIT.toLowerCase())));
+        }
+
+        redisSideTableInfo.check();
         return redisSideTableInfo;
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -91,6 +91,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S`
`91`	`91`	`cassandraSideTableInfo.setConnectTimeoutMillis(MathUtil.getIntegerVal(props.get(CONNECT_TIMEOUT_MILLIS_KEY.toLowerCase())));`
`92`	`92`	`cassandraSideTableInfo.setPoolTimeoutMillis(MathUtil.getIntegerVal(props.get(POOL_TIMEOUT_MILLIS_KEY.toLowerCase())));`
`93`	`93`
	`94`	`+ if (MathUtil.getLongVal(props.get(cassandraSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {`
	`95`	`+ cassandraSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(cassandraSideTableInfo.ERROR_LIMIT.toLowerCase())));`
	`96`	`+ }`
	`97`	`+`
`94`	`98`	`return cassandraSideTableInfo;`
`95`	`99`	`}`
`96`	`100`
Original file line number	Diff line number	Diff line change
`@@ -71,6 +71,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S`
`71`	`71`	`elasticsearch6SideTableInfo.setUserName(MathUtil.getString(props.get(KEY_ES6_USERNAME.toLowerCase())));`
`72`	`72`	`elasticsearch6SideTableInfo.setPassword(MathUtil.getString(props.get(KEY_ES6_PASSWORD.toLowerCase())));`
`73`	`73`	`}`
	`74`	`+`
	`75`	`+ if (MathUtil.getLongVal(props.get(elasticsearch6SideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {`
	`76`	`+ elasticsearch6SideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(elasticsearch6SideTableInfo.ERROR_LIMIT.toLowerCase())));`
	`77`	`+ }`
	`78`	`+`
`74`	`79`	`elasticsearch6SideTableInfo.check();`
`75`	`80`	`return elasticsearch6SideTableInfo;`
`76`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -80,6 +80,10 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S`
`80`	`80`	`);`
`81`	`81`	`kuduSideTableInfo.judgeKrbEnable();`
`82`	`82`
	`83`	`+ if (MathUtil.getLongVal(props.get(kuduSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {`
	`84`	`+ kuduSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(kuduSideTableInfo.ERROR_LIMIT.toLowerCase())));`
	`85`	`+ }`
	`86`	`+`
`83`	`87`	`return kuduSideTableInfo;`
`84`	`88`
`85`	`89`	`}`
Original file line number	Diff line number	Diff line change
`@@ -51,6 +51,11 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S`
`51`	`51`	`redisSideTableInfo.setMasterName(MathUtil.getString(props.get(RedisSideTableInfo.MASTER_NAME.toLowerCase())));`
`52`	`52`	`redisSideTableInfo.setRedisType(MathUtil.getString(props.get(RedisSideTableInfo.REDIS_TYPE.toLowerCase())));`
`53`	`53`
	`54`	`+ if (MathUtil.getLongVal(props.get(RedisSideTableInfo.ERROR_LIMIT.toLowerCase())) != null) {`
	`55`	`+ redisSideTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(RedisSideTableInfo.ERROR_LIMIT.toLowerCase())));`
	`56`	`+ }`
	`57`	`+`
	`58`	`+ redisSideTableInfo.check();`
`54`	`59`	`return redisSideTableInfo;`
`55`	`60`	`}`
`56`	`61`	`}`