添加重启策略的指定参数

Author: dapeng

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

@@ -63,6 +63,8 @@ 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;

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(ConfigConstrant.failureInterval, TimeUnit.MINUTES),
+                    Time.of(ConfigConstrant.delayInterval, TimeUnit.SECONDS)
+            ));
+        } else {
+            streamEnv.setRestartStrategy(RestartStrategies.noRestart());
+        }
 
         // checkpoint config
         Optional<Boolean> checkpointingEnabled = isCheckpointingEnabled(confProperties);
@@ -163,6 +167,11 @@ 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));
+    }
+
     /**
      * #ProcessingTime(默认), IngestionTime, EventTime
      * @param properties