save checkpoint when sql.checkpoint.cleanup.mode is null

Author: yanxi0227

core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java

@@ -101,16 +101,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
         }
 
         String cleanupModeStr = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY);
-        if(cleanupModeStr != null){//设置在cancel job情况下checkpoint是否被保存
-            if("true".equalsIgnoreCase(cleanupModeStr)){
-                env.getCheckpointConfig().enableExternalizedCheckpoints(
-                        CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
-            }else if("false".equalsIgnoreCase(cleanupModeStr)){
-                env.getCheckpointConfig().enableExternalizedCheckpoints(
-                        CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-            }else{
-                throw new RuntimeException("not support value of cleanup mode :" + cleanupModeStr);
-            }
+        if ("true".equalsIgnoreCase(cleanupModeStr)){
+            env.getCheckpointConfig().enableExternalizedCheckpoints(
+                    CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
+        } else if("false".equalsIgnoreCase(cleanupModeStr) || cleanupModeStr == null){
+            env.getCheckpointConfig().enableExternalizedCheckpoints(
+                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        } else{
+            throw new RuntimeException("not support value of cleanup mode :" + cleanupModeStr);
         }
 
         String backendPath = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_DATAURI_KEY);