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

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

@@ -20,6 +20,7 @@
 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;
@@ -36,6 +37,7 @@
 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;
@@ -86,10 +88,12 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private final static AtomicBoolean CONN_STATUS = new AtomicBoolean(true);
 
-    private final static AtomicLong TIMOUT_NUM = new AtomicLong(0);
+    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);
@@ -102,14 +106,6 @@ protected void init(BaseSideInfo sideInfo) {
         rdbSideTableInfo.setAsyncPoolSize(rdbPoolSize);
     }
 
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
-        LOG.info("rdb dim table config info: {} ", rdbSideTableInfo.toString());
-    }
-
-
     @Override
     protected void preInvoke(CRow input, ResultFuture<CRow> resultFuture){
 
@@ -148,7 +144,12 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                             logger.error("getConnection error", conn.cause());
                         }
                         if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3L)){
-                            outByJoinType(resultFuture, conn.cause());
+                            if(FAIL_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
+                                counter.inc();
+                                resultFuture.completeExceptionally(conn.cause());
+                            } else {
+                                dealMissKey(input, resultFuture);
+                            }
                             finishFlag.set(true);
                         }
                         conn.result().close();
@@ -160,6 +161,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);
                 } finally {
                     latch.countDown();
@@ -226,12 +228,13 @@ 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(TIMOUT_NUM.incrementAndGet() > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
-                   outByJoinType(resultFuture, rs.cause());
-                    return;
+                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);
                 }
-                LOG.error("Cannot retrieve the data from the database", rs.cause());
-                resultFuture.complete(null);
                 return;
             }
 
@@ -270,14 +273,6 @@ private void handleQuery(SQLConnection connection, Map<String, Object> inputPara
         });
     }
 
-    private void outByJoinType(ResultFuture<CRow> resultFuture, Throwable e){
-        if(sideInfo.getJoinType() == JoinType.LEFT){
-            resultFuture.complete(null);
-            return;
-        }
-        resultFuture.completeExceptionally(e);
-    }
-
     private Map<String, Object> formatInputParam(Map<String, Object> inputParam){
         Map<String, Object> result = Maps.newHashMap();
         inputParam.forEach((k,v) -> {