[hotfix-1772][rdb] 调整rdb errorLimit初始值，当出现脏数据，任务立即停止.

[hotfix-1772][rdb] 调整rdb errorLimit初始值，当出现脏数据，任务立即停止.

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/exception/ExceptionTrace.java

@@ -20,21 +20,4 @@ public static String traceOriginalCause(Throwable e) {
         }
         return errorMsg;
     }
-
-    /**
-     * 根据异常的种类来判断是否需要强制跳过Flink的重启{@link SuppressRestartsException}
-      * @param e exception
-     * @param errorMsg 需要抛出的异常信息
-     */
-    public static void dealExceptionWithSuppressStart(Exception e, String errorMsg) {
-        if (e instanceof SuppressRestartsException) {
-            throw new SuppressRestartsException(
-                new Throwable(
-                    errorMsg
-                )
-            );
-        } else {
-            throw new RuntimeException(errorMsg);
-        }
-    }
 }

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/JDBCUpsertOutputFormat.java

@@ -210,16 +210,13 @@ private void checkConnectionOpen() {
     public synchronized void flush() {
         try {
             jdbcWriter.executeBatch(connection);
-            batchCount = 0;
-        } catch (Exception e) {
+        } catch (SQLException e) {
             String errorMsg = String.format(
                 "Writing records to JDBC failed. Cause: [%s]",
                 ExceptionTrace.traceOriginalCause(e));
-
-            ExceptionTrace.dealExceptionWithSuppressStart(
-                e,
-                errorMsg
-            );
+            throw new RuntimeException(errorMsg);
+        } finally {
+            batchCount = 0;
         }
     }
 

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbSinkParser.java

@@ -51,7 +51,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         rdbTableInfo.setAllReplace(MathUtil.getBoolean(props.get(RdbTableInfo.ALLREPLACE_KEY.toLowerCase()), false));
         rdbTableInfo.setDriverName(MathUtil.getString(props.get(RdbTableInfo.DRIVER_NAME)));
         rdbTableInfo.setFastCheck(MathUtil.getBoolean(props.getOrDefault(RdbTableInfo.FAST_CHECK.toLowerCase(), true)));
-        rdbTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(RdbTableInfo.ERROR_LIMIT.toLowerCase()), 1000L));
+        rdbTableInfo.setErrorLimit(MathUtil.getLongVal(props.get(RdbTableInfo.ERROR_LIMIT.toLowerCase()), 0L));
         rdbTableInfo.setCheckProperties();
 
         rdbTableInfo.check();

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/JDBCWriter.java

@@ -93,18 +93,12 @@ default void dealExecuteError(Connection connection,
 					row.toString(),
 					ExceptionTrace.traceOriginalCause(e)));
 		}
+		metricOutputFormat.outDirtyRecords.inc();
 
 		if (errorLimit > -1) {
 			if (metricOutputFormat.outDirtyRecords.getCount() > errorLimit) {
-				throw new SuppressRestartsException(
-					new Throwable(
-						String.format("dirty data Count: [%s]. Error cause: [%s]",
-							metricOutputFormat.outDirtyRecords.getCount(),
-							ExceptionTrace.traceOriginalCause(e)))
-				);
+				throw new SuppressRestartsException(e);
 			}
 		}
-
-		metricOutputFormat.outDirtyRecords.inc();
 	}
 }