[Improve][Fink] NO_CDC source support checkpoint (apache#10094)

yzeng1618 · zengyi · web-flow · commit e7267cae4dd7 · 2025-11-26T21:58:34.000+08:00
Co-authored-by: zengyi &lt;zengyi@chinatelecom.cn&gt;
diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/execution/RuntimeEnvironment.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/execution/RuntimeEnvironment.java
@@ -61,13 +61,13 @@ static JobMode getJobMode(Config config) {
     }
 
     static boolean getEnableCheckpoint(Config config) {
-        boolean enableCheckpoint;
         Config envConfig = config.getConfig("env");
+        long checkpointInterval = -1;
         if (envConfig.hasPath(EnvCommonOptions.CHECKPOINT_INTERVAL.key())) {
-            enableCheckpoint = envConfig.getInt(EnvCommonOptions.CHECKPOINT_INTERVAL.key()) > 0;
-        } else {
-            enableCheckpoint = false;
+            checkpointInterval = envConfig.getLong(EnvCommonOptions.CHECKPOINT_INTERVAL.key());
+        } else if (envConfig.hasPath("execution.checkpoint.interval")) {
+            checkpointInterval = envConfig.getLong("execution.checkpoint.interval");
         }
-        return enableCheckpoint || getJobMode(config) == JobMode.STREAMING;
+        return checkpointInterval > 0 || getJobMode(config) == JobMode.STREAMING;
     }
 }
diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/execution/RuntimeEnvironmentTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/execution/RuntimeEnvironmentTest.java
@@ -45,5 +45,23 @@ void testEnableCheckpoint() {
                                 + "  checkpoint.interval = 10\n"
                                 + "}");
         Assertions.assertTrue(RuntimeEnvironment.getEnableCheckpoint(config));
+
+        config =
+                ConfigFactory.parseString(
+                        "env {\n"
+                                + "  parallelism = 1\n"
+                                + "  job.mode = \"BATCH\"\n"
+                                + "  execution.checkpoint.interval = 10\n"
+                                + "}");
+        Assertions.assertTrue(RuntimeEnvironment.getEnableCheckpoint(config));
+
+        config =
+                ConfigFactory.parseString(
+                        "env {\n"
+                                + "  parallelism = 1\n"
+                                + "  job.mode = \"BATCH\"\n"
+                                + "  checkpoint.interval = 0\n"
+                                + "}");
+        Assertions.assertFalse(RuntimeEnvironment.getEnableCheckpoint(config));
     }
 }
diff --git a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/AbstractFlinkRuntimeEnvironment.java b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/AbstractFlinkRuntimeEnvironment.java
@@ -46,6 +46,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.OptionalLong;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -56,6 +57,8 @@ public abstract class AbstractFlinkRuntimeEnvironment implements RuntimeEnvironm
     protected JobMode jobMode;
     protected String jobName = Constants.LOGO;
 
+    private static final long DEFAULT_CHECKPOINT_INTERVAL_MS = 10000L;
+
     protected AbstractFlinkRuntimeEnvironment(Config config) {
         this.initialize(config);
     }
@@ -77,19 +80,23 @@ public StreamExecutionEnvironment getStreamExecutionEnvironment() {
     }
 
     protected void setCheckpoint() {
-        if (jobMode == JobMode.BATCH) {
-            log.warn(
-                    "Disabled Checkpointing. In flink execution environment, checkpointing is not supported and not needed when executing jobs in BATCH mode");
+        OptionalLong intervalOpt = resolveCheckpointInterval(true);
+        boolean hasExplicitInterval = intervalOpt.isPresent();
+        boolean positiveInterval = intervalOpt.isPresent() && intervalOpt.getAsLong() > 0;
+        long interval = intervalOpt.orElse(DEFAULT_CHECKPOINT_INTERVAL_MS);
+
+        if (jobMode == JobMode.BATCH && !positiveInterval) {
+            log.info(
+                    "Checkpoint is disabled for batch job because 'checkpoint.interval' is not set or <= 0.");
+            return;
         }
-        long interval;
-        if (config.hasPath(EnvCommonOptions.CHECKPOINT_INTERVAL.key())) {
-            interval = config.getLong(EnvCommonOptions.CHECKPOINT_INTERVAL.key());
-        } else if (config.hasPath(ConfigKeyName.CHECKPOINT_INTERVAL)) {
+
+        if (hasExplicitInterval && !positiveInterval) {
             log.warn(
-                    "the parameter 'execution.checkpoint.interval' will be deprecated, please use common parameter 'checkpoint.interval' to set it");
-            interval = config.getLong(ConfigKeyName.CHECKPOINT_INTERVAL);
-        } else {
-            interval = 10000L;
+                    "checkpoint.interval is set to {} which is not positive, fallback to default {} ms for streaming job.",
+                    interval,
+                    DEFAULT_CHECKPOINT_INTERVAL_MS);
+            interval = DEFAULT_CHECKPOINT_INTERVAL_MS;
         }
 
         CheckpointConfig checkpointConfig = environment.getCheckpointConfig();
@@ -195,8 +202,28 @@ protected void createStreamEnvironment() {
         }
 
         if (this.jobMode.equals(JobMode.BATCH)) {
-            environment.setRuntimeMode(RuntimeExecutionMode.BATCH);
+            OptionalLong intervalOpt = resolveCheckpointInterval(false);
+            if (intervalOpt.isPresent() && intervalOpt.getAsLong() > 0) {
+                log.info(
+                        "Flink batch runtime does not support checkpoint-based restore; 'checkpoint.interval' > 0 will make this batch job run in streaming runtime.");
+            } else {
+                environment.setRuntimeMode(RuntimeExecutionMode.BATCH);
+            }
+        }
+    }
+
+    protected OptionalLong resolveCheckpointInterval(boolean warnLegacy) {
+        if (config.hasPath(EnvCommonOptions.CHECKPOINT_INTERVAL.key())) {
+            return OptionalLong.of(config.getLong(EnvCommonOptions.CHECKPOINT_INTERVAL.key()));
+        }
+        if (config.hasPath(ConfigKeyName.CHECKPOINT_INTERVAL)) {
+            if (warnLegacy) {
+                log.warn(
+                        "the parameter 'execution.checkpoint.interval' will be deprecated, please use common parameter 'checkpoint.interval' to set it");
+            }
+            return OptionalLong.of(config.getLong(ConfigKeyName.CHECKPOINT_INTERVAL));
         }
+        return OptionalLong.empty();
     }
 
     private void setTimeCharacteristic() {
diff --git a/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java b/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java
@@ -52,6 +52,7 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -135,11 +136,17 @@ public void execute() throws TaskExecuteException {
                 "Flink Execution Plan: {}",
                 flinkRuntimeEnvironment.getStreamExecutionEnvironment().getExecutionPlan());
         LOGGER.info("Flink job name: {}", flinkRuntimeEnvironment.getJobName());
-        if (!flinkRuntimeEnvironment.isStreaming()) {
-            flinkRuntimeEnvironment
-                    .getStreamExecutionEnvironment()
-                    .setRuntimeMode(RuntimeExecutionMode.BATCH);
-            LOGGER.info("Flink job Mode: {}", JobMode.BATCH);
+        if (flinkRuntimeEnvironment.getJobMode() == JobMode.BATCH) {
+            OptionalLong checkpointInterval =
+                    flinkRuntimeEnvironment.resolveCheckpointInterval(false);
+            boolean enableCheckpointForBatch =
+                    checkpointInterval.isPresent() && checkpointInterval.getAsLong() > 0;
+            if (!enableCheckpointForBatch) {
+                flinkRuntimeEnvironment
+                        .getStreamExecutionEnvironment()
+                        .setRuntimeMode(RuntimeExecutionMode.BATCH);
+                LOGGER.info("Flink job Mode: {}", JobMode.BATCH);
+            }
         }
         try {
             final long jobStartTime = System.currentTimeMillis();
diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java
@@ -21,7 +21,6 @@
 import org.apache.seatunnel.e2e.common.TestSuiteBase;
 import org.apache.seatunnel.e2e.common.container.EngineType;
 import org.apache.seatunnel.e2e.common.container.TestContainer;
-import org.apache.seatunnel.e2e.common.container.TestContainerId;
 import org.apache.seatunnel.e2e.common.container.flink.AbstractTestFlinkContainer;
 import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
 import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
@@ -208,9 +207,10 @@ public void testZetaStreamingCheckpointNoInterval(TestContainer container)
     public void testFlinkCheckpointEnable(AbstractTestFlinkContainer container)
             throws IOException, InterruptedException {
         /**
-         * In flink execution environment, checkpoint is not supported and not needed when executing
-         * jobs in BATCH mode. So it is only necessary to determine whether flink has enabled
-         * checkpoint by configuring tasks with 'checkpoint.interval'.
+         * In Flink execution environment, batch jobs normally do not enable checkpointing. When
+         * 'checkpoint.interval' is configured for a batch job, SeaTunnel will submit it in
+         * streaming runtime with the same checkpoint interval. This test verifies that Flink has
+         * enabled checkpointing and uses the configured interval.
          */
         Container.ExecResult enableExecResult =
                 container.executeJob(
@@ -228,19 +228,12 @@ public void testFlinkCheckpointEnable(AbstractTestFlinkContainer container)
                                         jobId)),
                         String.class,
                         Object.class);
-        /**
-         * when the checkpoint interval is 0x7fffffffffffffff, indicates that checkpoint is
-         * disabled. reference {@link
-         * org.apache.flink.runtime.jobgraph.JobGraph#isCheckpointingEnabled()}
-         */
-        if (container.identifier().equals(TestContainerId.FLINK_1_13)
-                || container.identifier().equals(TestContainerId.FLINK_1_14)
-                || container.identifier().equals(TestContainerId.FLINK_1_15)
-                || container.identifier().equals(TestContainerId.FLINK_1_16)) {
-            Assertions.assertEquals(Long.MAX_VALUE, jobConfig.getOrDefault("interval", 0L));
-        } else {
-            Assertions.assertEquals(0, jobConfig.getOrDefault("interval", 0));
-        }
+        Object intervalObject = jobConfig.get("interval");
+        Assertions.assertNotNull(intervalObject);
+        long interval = ((Number) intervalObject).longValue();
+        // the value here should be consistent with `checkpoint.interval` in
+        // batch_fakesource_to_localfile_checkpoint_enable.conf
+        Assertions.assertEquals(1000L, interval);
     }
 
     @TestTemplate

Original file line number	Diff line number	Diff line change
`@@ -61,13 +61,13 @@ static JobMode getJobMode(Config config) {`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`static boolean getEnableCheckpoint(Config config) {`
`64`		`- boolean enableCheckpoint;`
`65`	`64`	`Config envConfig = config.getConfig("env");`
	`65`	`+ long checkpointInterval = -1;`
`66`	`66`	`if (envConfig.hasPath(EnvCommonOptions.CHECKPOINT_INTERVAL.key())) {`
`67`		`- enableCheckpoint = envConfig.getInt(EnvCommonOptions.CHECKPOINT_INTERVAL.key()) > 0;`
`68`		`- } else {`
`69`		`- enableCheckpoint = false;`
	`67`	`+ checkpointInterval = envConfig.getLong(EnvCommonOptions.CHECKPOINT_INTERVAL.key());`
	`68`	`+ } else if (envConfig.hasPath("execution.checkpoint.interval")) {`
	`69`	`+ checkpointInterval = envConfig.getLong("execution.checkpoint.interval");`
`70`	`70`	`}`
`71`		`- return enableCheckpoint \|\| getJobMode(config) == JobMode.STREAMING;`
	`71`	`+ return checkpointInterval > 0 \|\| getJobMode(config) == JobMode.STREAMING;`
`72`	`72`	`}`
`73`	`73`	`}`