查询根据失败次数和joinType输出

Author: dapeng

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

@@ -68,13 +68,11 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
 
     private int  asyncTimeout=10000;
 
-    private int asyncTimeoutNumLimit = Integer.MAX_VALUE;
-
     private boolean partitionedJoin = false;
 
     private String cacheMode="ordered";
 
-    private Integer asyncFailMaxNum;
+    private Long asyncFailMaxNum;
 
     private List<PredicateInfo>  predicateInfoes = Lists.newArrayList();
 
@@ -153,19 +151,11 @@ public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
 
-    public int getAsyncTimeoutNumLimit() {
-        return asyncTimeoutNumLimit;
-    }
-
-    public void setAsyncTimeoutNumLimit(int asyncTimeoutNumLimit) {
-        this.asyncTimeoutNumLimit = asyncTimeoutNumLimit;
-    }
-
-    public Integer getAsyncFailMaxNum(Integer defaultValue) {
+    public Long getAsyncFailMaxNum(Long defaultValue) {
         return Objects.isNull(asyncFailMaxNum) ? defaultValue : asyncFailMaxNum;
     }
 
-    public void setAsyncFailMaxNum(Integer asyncFailMaxNum) {
+    public void setAsyncFailMaxNum(Long asyncFailMaxNum) {
         this.asyncFailMaxNum = asyncFailMaxNum;
     }
 

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

@@ -149,7 +149,7 @@ public void timeout(CRow input, ResultFuture<CRow> resultFuture) throws Exceptio
             resultFuture.complete(null);
             return;
         }
-        if(timeOutNum > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Integer.MAX_VALUE)){
+        if(timeOutNum > sideInfo.getSideTableInfo().getAsyncFailMaxNum(Long.MAX_VALUE)){
             resultFuture.completeExceptionally(new Exception("Async function call timedoutNum beyond limit."));
             return;
         }

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

@@ -116,7 +116,7 @@ protected void parseCacheProp(AbstractSideTableInfo sideTableInfo, Map<String, O
             }
 
             if(props.containsKey(AbstractSideTableInfo.ASYNC_FAIL_MAX_NUM_KEY.toLowerCase())){
-                Integer asyncFailNum = MathUtil.getIntegerVal(props.get(AbstractSideTableInfo.ASYNC_FAIL_MAX_NUM_KEY.toLowerCase()));
+                Long asyncFailNum = MathUtil.getLongVal(props.get(AbstractSideTableInfo.ASYNC_FAIL_MAX_NUM_KEY.toLowerCase()));
                 if (asyncFailNum > 0){
                     sideTableInfo.setAsyncFailMaxNum(asyncFailNum);
                 }

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

@@ -30,6 +30,7 @@
 import io.vertx.ext.sql.SQLClient;
 import io.vertx.ext.sql.SQLConnection;
 import com.google.common.collect.Lists;
+import org.apache.calcite.sql.JoinType;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
@@ -38,16 +39,19 @@
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import scala.Int;
 
 import java.sql.Timestamp;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -81,6 +85,8 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private final static AtomicBoolean CONN_STATUS = new AtomicBoolean(true);
 
+    private final static AtomicLong TIMOUT_NUM = new AtomicLong(0);
+
     private Logger logger = LoggerFactory.getLogger(getClass());
 
     public RdbAsyncReqRow(BaseSideInfo sideInfo) {
@@ -114,7 +120,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, Resul
     }
 
     private void connectWithRetry(Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture, SQLClient rdbSqlClient) {
-        AtomicInteger failCounter = new AtomicInteger(0);
+        AtomicLong failCounter = new AtomicLong(0);
         AtomicBoolean finishFlag = new AtomicBoolean(false);
         while(!finishFlag.get()){
             CountDownLatch latch = new CountDownLatch(1);
@@ -124,8 +130,8 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                         if(failCounter.getAndIncrement() % 1000 == 0){
                             logger.error("getConnection error", conn.cause());
                         }
-                        if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3)){
-                            resultFuture.completeExceptionally(conn.cause());
+                        if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3L)){
+                            outByJoinType(resultFuture, conn.cause());
                             finishFlag.set(true);
                         }
                         conn.result().close();
@@ -202,8 +208,12 @@ 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;
+                }
                 LOG.error("Cannot retrieve the data from the database", rs.cause());
-                resultFuture.completeExceptionally(rs.cause());
+                resultFuture.complete(null);
                 return;
             }
 
@@ -242,4 +252,12 @@ 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);
+    }
+
 }