去除多余日志logger和采用全局计数器

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;
@@ -257,9 +255,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

@@ -76,13 +76,10 @@ 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);
     }
 
-
     @Override
     protected void preInvoke(CRow input, ResultFuture<CRow> resultFuture){
 
@@ -93,7 +90,6 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, Resul
 
         AtomicLong networkLogCounter = new AtomicLong(0L);
         while (!CONN_STATUS.get()){//network is unhealth
-            //todo:统一计数
             if(networkLogCounter.getAndIncrement() % 1000 == 0){
                 LOG.info("network unhealth to block task");
             }
@@ -122,7 +118,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());
@@ -146,7 +142,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
             try {
                 latch.wait();
             } catch (InterruptedException e) {
-                logger.error("", e);
+                LOG.error("", e);
             }
         }