添加记述和超过次数任务退出

Author: dapeng

core/src/main/java/com/dtstack/flink/sql/table/AbstractSideTableParser.java

@@ -108,12 +108,6 @@ protected void parseCacheProp(AbstractSideTableInfo sideTableInfo, Map<String, O
                 }
                 sideTableInfo.setAsyncTimeout(asyncTimeout);
             }
-            if(props.containsKey(AbstractSideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase())){
-                Integer asyncTimeoutNum = MathUtil.getIntegerVal(props.get(AbstractSideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase()));
-                if (asyncTimeoutNum > 0){
-                    sideTableInfo.setAsyncTimeoutNumLimit(asyncTimeoutNum);
-                }
-            }
 
             if(props.containsKey(AbstractSideTableInfo.ASYNC_FAIL_MAX_NUM_KEY.toLowerCase())){
                 Long asyncFailNum = MathUtil.getLongVal(props.get(AbstractSideTableInfo.ASYNC_FAIL_MAX_NUM_KEY.toLowerCase()));

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;
@@ -31,6 +32,7 @@
 import io.vertx.ext.sql.SQLConnection;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.lang3.StringUtils;
+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;
@@ -77,10 +79,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);
     }
@@ -123,7 +127,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();
@@ -135,6 +144,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();
@@ -200,12 +210,13 @@ private void handleQuery(SQLConnection connection,Map<String, Object> inputParam
         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;
             }
 
@@ -244,12 +255,4 @@ private void handleQuery(SQLConnection connection,Map<String, Object> inputParam
         });
     }
 
-    private void outByJoinType(ResultFuture<CRow> resultFuture, Throwable e){
-        if(sideInfo.getJoinType() == JoinType.LEFT){
-            resultFuture.complete(null);
-            return;
-        }
-        resultFuture.completeExceptionally(e);
-    }
-
 }