Merge remote-tracking branch 'origin/feat_async_mergeTest' into 1.8_test_3.10.x

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/Main.java
#	core/src/main/java/com/dtstack/flink/sql/side/AsyncReqRow.java
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

Author: dapeng

clickhouse/clickhouse-side/clickhouse-all-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-side/clickhouse-async-side/pom.xml

@@ -41,7 +41,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

clickhouse/clickhouse-sink/pom.xml

@@ -40,7 +40,7 @@
                         <configuration>
                             <artifactSet>
                                 <excludes>
-
+                                    <exclude>org.slf4j</exclude>
                                 </excludes>
                             </artifactSet>
                             <filters>

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

@@ -51,6 +51,9 @@ public abstract class AsyncReqRow extends RichAsyncFunction<CRow, CRow> implemen
     private static final Logger LOG = LoggerFactory.getLogger(AsyncReqRow.class);
     private static final long serialVersionUID = 2098635244857937717L;
 
+    private static int TIMEOUT_LOG_FLUSH_NUM = 10;
+    private int timeOutNum = 0;
+
     protected SideInfo sideInfo;
     protected transient Counter parseErrorRecords;
 
@@ -119,13 +122,16 @@ protected void dealCacheData(String key, CacheObj missKeyObj) {
 
     @Override
     public void timeout(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
-        StreamRecordQueueEntry<CRow> future = (StreamRecordQueueEntry<CRow>)resultFuture;
-        try {
-            if (null == future.get()) {
-                resultFuture.completeExceptionally(new TimeoutException("Async function call has timed out."));
-            }
-        } catch (Exception e) {
-            resultFuture.completeExceptionally(new Exception(e));
+
+        //TODO 需要添加数据指标
+        if(timeOutNum % TIMEOUT_LOG_FLUSH_NUM == 0){
+            LOG.info("Async function call has timed out. input:{}, timeOutNum:{}",input.toString(), timeOutNum);
+        }
+        timeOutNum ++;
+        if(timeOutNum > sideInfo.getSideTableInfo().getAsyncTimeoutNumLimit()){
+            resultFuture.completeExceptionally(new Exception("Async function call timedoutNum beyond limit."));
+        } else {
+            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 = Integer.MAX_VALUE;
+
     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.setAsyncTimeoutNumLimit(asyncTimeoutNum);
+                }
+            }
         }
     }
 }