async的超时次数配置

Author: dapeng

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

@@ -125,11 +125,13 @@ public void timeout(Row input, ResultFuture<Row> resultFuture) throws Exception
 
         //TODO 需要添加数据指标
         if(timeOutNum % TIMEOUT_LOG_FLUSH_NUM == 0){
-            LOG.info("Async function call has timed out. input:" + input.toString());
+            LOG.info("Async function call has timed out. input:{}, timeOutNum:{}",input.toString(), timeOutNum);
+        }
+        timeOutNum ++;
+        if(timeOutNum > sideInfo.getSideTableInfo().getAsyncTimeoutNumLimit()){
+            resultFuture.complete(null);
         }
 
-        timeOutNum++;
-        resultFuture.complete(null);
     }
 
 

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

@@ -53,6 +53,8 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     public static final String ASYNC_TIMEOUT_KEY = "asyncTimeout";
 
+    public static final String ASYNC_TIMEOUT_NUM_KEY = "asyncTimeoutNum";
+
     private String cacheType = "none";//None or LRU or ALL
 
     private int cacheSize = 10000;
@@ -63,6 +65,8 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     private int  asyncTimeout=10000;
 
+    private int asyncTimeoutNumLimit = 0;
+
     private boolean partitionedJoin = false;
 
     private String cacheMode="ordered";
@@ -143,4 +147,13 @@ public void setPredicateInfoes(List<PredicateInfo> predicateInfoes) {
     public List<PredicateInfo> getPredicateInfoes() {
         return predicateInfoes;
     }
+
+    public int getAsyncTimeoutNumLimit() {
+        return asyncTimeoutNumLimit;
+    }
+
+    public void setAsyncTimeoutNumLimit(int asyncTimeoutNumLimit) {
+        this.asyncTimeoutNumLimit = asyncTimeoutNumLimit;
+    }
+
 }

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

@@ -108,6 +108,12 @@ protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> p
                 }
                 sideTableInfo.setAsyncTimeout(asyncTimeout);
             }
+            if(props.containsKey(SideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase())){
+                Integer asyncTimeoutNum = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_TIMEOUT_NUM_KEY.toLowerCase()));
+                if (asyncTimeoutNum > 0){
+                    sideTableInfo.setAsyncTimeout(asyncTimeoutNum);
+                }
+            }
         }
     }
 }