Merge branch '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,6 +51,7 @@
 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
@@ -64,10 +65,9 @@ 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;
     protected transient Counter parseErrorRecords;
 
@@ -125,7 +125,7 @@ protected void dealMissKey(CRow input, ResultFuture<CRow> resultFuture){
                 Row row = fillData(input.row(), null);
                 resultFuture.complete(Collections.singleton(new CRow(row, input.change())));
             } catch (Exception e) {
-                dealFillDataError(resultFuture, e, input);
+                dealFillDataError(input, resultFuture, e);
             }
         }else{
             resultFuture.complete(null);
@@ -206,7 +206,7 @@ private void invokeWithCache(Map<String, Object> inputParams, CRow input, Result
                         Row row = fillData(input.row(), val);
                         resultFuture.complete(Collections.singleton(new CRow(row, input.change())));
                     } catch (Exception e) {
-                        dealFillDataError(resultFuture, e, input);
+                        dealFillDataError(input, resultFuture, e);
                     }
                 } else if (ECacheContentType.MultiLine == val.getType()) {
                     try {
@@ -217,7 +217,7 @@ private void invokeWithCache(Map<String, Object> inputParams, CRow input, Result
                         }
                         resultFuture.complete(rowList);
                     } catch (Exception e) {
-                        dealFillDataError(resultFuture, e, input);
+                        dealFillDataError(input, resultFuture, e);
                     }
                 } else {
                     resultFuture.completeExceptionally(new RuntimeException("not support cache obj type " + val.getType()));
@@ -257,11 +257,14 @@ protected void cancelTimerWhenComplete(ResultFuture<CRow> resultFuture, Schedule
         }
     }
 
-    protected void dealFillDataError(ResultFuture<CRow> resultFuture, Exception e, Object sourceData) {
-        LOG.debug("source data {} join side table error ", sourceData);
-        LOG.debug("async buid row error..{}", e);
-        parseErrorRecords.inc();
-        resultFuture.complete(Collections.emptyList());
+    protected void dealFillDataError(CRow input, ResultFuture<CRow> resultFuture, Throwable e) {
+        if(FAIL_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
+            LOG.info("dealFillDataError", e);
+            parseErrorRecords.inc();
+            resultFuture.completeExceptionally(e);
+        } else {
+            dealMissKey(input, resultFuture);
+        }
     }
 
     @Override

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

@@ -20,7 +20,6 @@
 package com.dtstack.flink.sql.side.rdb.async;
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
-import com.dtstack.flink.sql.metric.MetricConstant;
 import com.dtstack.flink.sql.side.BaseAsyncReqRow;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.CacheMissVal;
@@ -34,10 +33,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.sql.SQLClient;
 import io.vertx.ext.sql.SQLConnection;
-import org.apache.calcite.sql.JoinType;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.Counter;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
@@ -88,12 +84,8 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private final static AtomicBoolean CONN_STATUS = new AtomicBoolean(true);
 
-    private final static AtomicLong FAIL_NUM = new AtomicLong(0);
-
     private Logger logger = LoggerFactory.getLogger(getClass());
 
-    private Counter counter = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_SIDE_PARSE_ERROR_RECORDS);
-
     public RdbAsyncReqRow(BaseSideInfo sideInfo) {
         super(sideInfo);
         init(sideInfo);
@@ -114,7 +106,11 @@ protected void preInvoke(CRow input, ResultFuture<CRow> resultFuture){
     @Override
     public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture) throws Exception {
 
+        AtomicLong networkLogCounter = new AtomicLong(0L);
         while (!CONN_STATUS.get()){//network is unhealth
+            if(networkLogCounter.getAndIncrement() % 1000 == 0){
+                LOG.info("network unhealth to block task");
+            }
             Thread.sleep(100);
         }
         Map<String, Object> params = formatInputParam(inputParams);
@@ -144,12 +140,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                             logger.error("getConnection error", conn.cause());
                         }
                         if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3L)){
-                            if(FAIL_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
-                                counter.inc();
-                                resultFuture.completeExceptionally(conn.cause());
-                            } else {
-                                dealMissKey(input, resultFuture);
-                            }
+                            dealFillDataError(input, resultFuture, conn.cause());
                             finishFlag.set(true);
                         }
                         conn.result().close();
@@ -161,8 +152,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                     handleQuery(conn.result(), inputParams, input, resultFuture);
                     finishFlag.set(true);
                 } catch (Exception e) {
-                    dealFillDataError(resultFuture, e, null);
-                    logger.error("", e);
+                    dealFillDataError(input, resultFuture, e);
                 } finally {
                     latch.countDown();
                 }
@@ -228,13 +218,7 @@ private void handleQuery(SQLConnection connection, Map<String, Object> inputPara
         JsonArray params = new JsonArray(Lists.newArrayList(inputParams.values()));
         connection.queryWithParams(sideInfo.getSqlCondition(), params, rs -> {
             if (rs.failed()) {
-                if(FAIL_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
-                    LOG.error("Cannot retrieve the data from the database", rs.cause());
-                    counter.inc();
-                    resultFuture.completeExceptionally(rs.cause());
-                } else {
-                    dealMissKey(input, resultFuture);
-                }
+                dealFillDataError(input, resultFuture, rs.cause());
                 return;
             }