rdb 连接重试配置项

Author: dapeng

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

@@ -55,7 +55,9 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
     public static final String ASYNC_TIMEOUT_KEY = "asyncTimeout";
 
     public static final String ASYNC_FAIL_MAX_NUM_KEY = "asyncFailMaxNum";
-    
+
+    public static final String CONNECT_RETRY_MAX_NUM_KEY = "connectRetryMaxNum";
+
     public static final String ASYNC_REQ_POOL_KEY = "asyncPoolSize";
 
     private String cacheType = "none";
@@ -79,6 +81,8 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
 
     private Long asyncFailMaxNum;
 
+    private Integer connectRetryMaxNum;
+
     private List<PredicateInfo>  predicateInfoes = Lists.newArrayList();
 
     public RowTypeInfo getRowTypeInfo(){
@@ -164,6 +168,7 @@ public void setAsyncFailMaxNum(Long asyncFailMaxNum) {
         this.asyncFailMaxNum = asyncFailMaxNum;
     }
 
+
     public int getAsyncPoolSize() {
         return asyncPoolSize;
     }
@@ -172,6 +177,14 @@ public void setAsyncPoolSize(int asyncPoolSize) {
         this.asyncPoolSize = asyncPoolSize;
     }
 
+
+    public Integer getConnectRetryMaxNum(Integer defaultValue) {
+        return Objects.isNull(connectRetryMaxNum) ? defaultValue : connectRetryMaxNum;
+    }
+
+    public void setConnectRetryMaxNum(Integer connectRetryMaxNum) {
+        this.connectRetryMaxNum = connectRetryMaxNum;
+    }
     @Override
     public String toString() {
         return "Cache Info{" +

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

@@ -110,10 +110,10 @@ protected void parseCacheProp(AbstractSideTableInfo sideTableInfo, Map<String, O
                 sideTableInfo.setAsyncTimeout(asyncTimeout);
             }
 
-            if(props.containsKey(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);
+            if(props.containsKey(AbstractSideTableInfo.CONNECT_RETRY_MAX_NUM_KEY.toLowerCase())){
+                Integer connectRetryMaxNum = MathUtil.getIntegerVal(props.get(AbstractSideTableInfo.CONNECT_RETRY_MAX_NUM_KEY.toLowerCase()));
+                if (connectRetryMaxNum > 0){
+                    sideTableInfo.setConnectRetryMaxNum(connectRetryMaxNum);
                 }
             }
         }

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

@@ -137,7 +137,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                         if(failCounter.getAndIncrement() % 1000 == 0){
                             LOG.error("getConnection error", conn.cause());
                         }
-                        if(failCounter.get() >= sideInfo.getSideTableInfo().getAsyncFailMaxNum(3L)){
+                        if(failCounter.get() >= sideInfo.getSideTableInfo().getConnectRetryMaxNum(3)){
                             dealFillDataError(input, resultFuture, conn.cause());
                             finishFlag.set(true);
                         }