side async param

Author: todd5167

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

@@ -47,12 +47,20 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     public static final String CACHE_MODE_KEY = "cacheMode";
 
+    public static final String ASYNC_CAP_KEY = "asyncCapacity";
+
+    public static final String ASYNC_TIMEOUT_KEY = "asyncTimeout";
+
     private String cacheType = "none";//None or LRU or ALL
 
     private int cacheSize = 10000;
 
     private long cacheTimeout = 60 * 1000;//
 
+    private int  asyncCapacity=100;
+
+    private int  asyncTimeout=10000;
+
     private boolean partitionedJoin = false;
 
     private String cacheMode="ordered";
@@ -107,4 +115,20 @@ public String getCacheMode() {
     public void setCacheMode(String cacheMode) {
         this.cacheMode = cacheMode;
     }
+
+    public int getAsyncCapacity() {
+        return asyncCapacity;
+    }
+
+    public void setAsyncCapacity(int asyncCapacity) {
+        this.asyncCapacity = asyncCapacity;
+    }
+
+    public int getAsyncTimeout() {
+        return asyncTimeout;
+    }
+
+    public void setAsyncTimeout(int asyncTimeout) {
+        this.asyncTimeout = asyncTimeout;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/side/operator/SideAsyncOperator.java

@@ -49,9 +49,6 @@ public class SideAsyncOperator {
     private static final String ORDERED = "ordered";
 
 
-    //TODO need to set by create table task
-    private static int asyncCapacity = 100;
-
     private static AsyncReqRow loadAsyncReq(String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,
                                             JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
@@ -70,10 +67,10 @@ public static DataStream getSideJoinDataStream(DataStream inputStream, String si
 
         //TODO How much should be set for the degree of parallelism? Timeout? capacity settings?
         if (ORDERED.equals(sideTableInfo.getCacheMode())){
-            return AsyncDataStream.orderedWait(inputStream, asyncDbReq, 10000, TimeUnit.MILLISECONDS, asyncCapacity)
+            return AsyncDataStream.orderedWait(inputStream, asyncDbReq, sideTableInfo.getAsyncTimeout(), TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
                     .setParallelism(sideTableInfo.getParallelism());
         }else {
-            return AsyncDataStream.unorderedWait(inputStream, asyncDbReq, 10000, TimeUnit.MILLISECONDS, asyncCapacity)
+            return AsyncDataStream.unorderedWait(inputStream, asyncDbReq, sideTableInfo.getAsyncTimeout(), TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
                     .setParallelism(sideTableInfo.getParallelism());
         }
 

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

@@ -95,6 +95,22 @@ protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> p
                 }
                 sideTableInfo.setCacheMode(cachemode.toLowerCase());
             }
+
+            if(props.containsKey(SideTableInfo.ASYNC_CAP_KEY.toLowerCase())){
+                Integer asyncCap = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_CAP_KEY.toLowerCase()));
+                if(asyncCap < 0){
+                    throw new RuntimeException("asyncCapacity size need > 0.");
+                }
+                sideTableInfo.setAsyncCapacity(asyncCap);
+            }
+
+            if(props.containsKey(SideTableInfo.ASYNC_TIMEOUT_KEY.toLowerCase())){
+                Integer asyncTimeout = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_TIMEOUT_KEY.toLowerCase()));
+                if (asyncTimeout<0){
+                    throw new RuntimeException("asyncTimeout size need > 0.");
+                }
+                sideTableInfo.setAsyncTimeout(asyncTimeout);
+            }
         }
     }
 }

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java

@@ -62,6 +62,7 @@ public long extractTimestamp(Row row) {
             Long extractTime=eveTime;
 
             lastTime = extractTime + timezone.getOffset(extractTime);
+
             eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - convertTimeZone(extractTime))/1000));
 
             return lastTime;

docs/mysqlSide.md

@@ -53,7 +53,8 @@
     * cacheSize: 缓存的条目数量
     * cacheTTLMs:缓存的过期时间(ms)
     * cacheMode: (unordered|ordered)异步加载是有序还是无序,默认有序。
-  
+    * asyncCapacity:异步请求容量，默认1000
+    * asyncTimeout：异步请求超时时间，默认10000毫秒
 
 ## 5.样例
 ```
@@ -72,6 +73,8 @@ create table sideTable(
     cacheSize ='10000',
     cacheTTLMs ='60000',
     cacheMode='unordered',
+    asyncCapacity='1000',
+    asyncTimeout='10000'
     parallelism ='1',
     partitionedJoin='false'
  );