Merge remote-tracking branch 'origin/feat_1.8_asyncException' into feat_1.8_asyncException_mergeDev

# Conflicts:
#	rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

Author: dapeng

core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java

@@ -51,7 +51,6 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
@@ -65,7 +64,6 @@ public abstract class BaseAsyncReqRow extends RichAsyncFunction<CRow, CRow> impl
     private static final Logger LOG = LoggerFactory.getLogger(BaseAsyncReqRow.class);
     private static final long serialVersionUID = 2098635244857937717L;
     private RuntimeContext runtimeContext;
-    private final static AtomicLong FAIL_NUM = new AtomicLong(0);
     private static int TIMEOUT_LOG_FLUSH_NUM = 10;
     private int timeOutNum = 0;
     protected BaseSideInfo sideInfo;
@@ -258,9 +256,9 @@ protected void cancelTimerWhenComplete(ResultFuture<CRow> resultFuture, Schedule
     }
 
     protected void dealFillDataError(CRow input, ResultFuture<CRow> resultFuture, Throwable e) {
-        if(FAIL_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
+        parseErrorRecords.inc();
+        if(parseErrorRecords.getCount() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
             LOG.info("dealFillDataError", e);
-            parseErrorRecords.inc();
             resultFuture.completeExceptionally(e);
         } else {
             dealMissKey(input, resultFuture);

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -84,8 +84,6 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private final static AtomicBoolean CONN_STATUS = new AtomicBoolean(true);
 
-    private Logger logger = LoggerFactory.getLogger(getClass());
-
     public RdbAsyncReqRow(BaseSideInfo sideInfo) {
         super(sideInfo);
         init(sideInfo);
@@ -137,7 +135,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                 try {
                     if(conn.failed()){
                         if(failCounter.getAndIncrement() % 1000 == 0){
-                            logger.error("getConnection error", conn.cause());
+                            LOG.error("getConnection error", conn.cause());
                         }
                         if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3L)){
                             dealFillDataError(input, resultFuture, conn.cause());
@@ -161,7 +159,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
             try {
                 latch.wait();
             } catch (InterruptedException e) {
-                logger.error("", e);
+                LOG.error("", e);
             }
         }