重启测试增加failure.interval和delay.interval参数

Author: dapeng

README.md

@@ -149,6 +149,9 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
         * taskmanager.slots：per_job模式下指定每个taskmanager对应的slot数量(默认1)
         * savePointPath：任务恢复点的路径（默认无）
         * allowNonRestoredState：指示保存点是否允许非还原状态的标志（默认false）
+        * restore.enable：是否失败重启（默认是true）
+        * failure.interval：衡量失败率的时间段，单位分钟（默认6m）
+        * delay.interval：连续两次重启尝试间的间隔，单位是秒（默认10s）
         * [prometheus 相关参数](docs/prometheus.md) per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
 
 

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -69,8 +69,8 @@ public class ConfigConstrant {
     // restart plocy
     public static final int failureRate = 3;
 
-    public static final int failureInterval = 6; //min
+    public static final String FAILUREINTERVAL = "failure.interval"; //min
 
-    public static final int delayInterval = 10; //sec
+    public static final String  DELAYINTERVAL= "delay.interval"; //sec
 
 }

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -104,8 +104,8 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
         if(isRestore(confProperties).get()){
             streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
                     ConfigConstrant.failureRate,
-                    Time.of(ConfigConstrant.failureInterval, TimeUnit.MINUTES),
-                    Time.of(ConfigConstrant.delayInterval, TimeUnit.SECONDS)
+                    Time.of(getFailureInterval(confProperties).get(), TimeUnit.MINUTES),
+                    Time.of(getDelayInterval(confProperties).get(), TimeUnit.SECONDS)
             ));
         } else {
             streamEnv.setRestartStrategy(RestartStrategies.noRestart());
@@ -172,6 +172,15 @@ public static Optional<Boolean> isRestore(Properties properties){
         return Optional.of(Boolean.valueOf(restoreEnable));
     }
 
+    public static Optional<Integer> getDelayInterval(Properties properties){
+        String delayInterval = properties.getProperty(ConfigConstrant.DELAYINTERVAL, "10");
+        return Optional.of(Integer.valueOf(delayInterval));
+    }
+    public static Optional<Integer> getFailureInterval(Properties properties){
+        String failureInterval = properties.getProperty(ConfigConstrant.FAILUREINTERVAL, "6");
+        return Optional.of(Integer.valueOf(failureInterval));
+    }
+
     /**
      * #ProcessingTime(默认), IngestionTime, EventTime
      * @param properties