Merge branch 'feat_1.8_restartStrategy' into 'v1.8.0_dev'

添加重启策略的指定参数



See merge request !228

Author: zoudaokoulife

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

@@ -63,12 +63,14 @@ public class ConfigConstrant {
     public static final String CHECKPOINTS_DIRECTORY_KEY = "state.checkpoints.dir";
     public static final String STATE_BACKEND_INCREMENTAL_KEY = "state.backend.incremental";
 
+    public static final String RESTOREENABLE = "restore.enable";
+
 
     // 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

@@ -101,11 +101,15 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
             }
         });
 
-        streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
-                ConfigConstrant.failureRate,
-                Time.of(ConfigConstrant.failureInterval, TimeUnit.MINUTES),
-                Time.of(ConfigConstrant.delayInterval, TimeUnit.SECONDS)
-        ));
+        if(isRestore(confProperties).get()){
+            streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
+                    ConfigConstrant.failureRate,
+                    Time.of(getFailureInterval(confProperties).get(), TimeUnit.MINUTES),
+                    Time.of(getDelayInterval(confProperties).get(), TimeUnit.SECONDS)
+            ));
+        } else {
+            streamEnv.setRestartStrategy(RestartStrategies.noRestart());
+        }
 
         // checkpoint config
         Optional<Boolean> checkpointingEnabled = isCheckpointingEnabled(confProperties);
@@ -163,6 +167,20 @@ public static Optional<Long> getAutoWatermarkInterval(Properties properties) {
         return StringUtils.isNotBlank(autoWatermarkInterval) ? Optional.of(Long.valueOf(autoWatermarkInterval)) : Optional.empty();
     }
 
+    public static Optional<Boolean> isRestore(Properties properties){
+        String restoreEnable = properties.getProperty(ConfigConstrant.RESTOREENABLE, "true");
+        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