This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 53c11a519e [Fix][Zeta] Job stuck permanently after master failover,
unable to complete (affects BATCH / bounded source / job shutdown phase)
(#10836)
53c11a519e is described below
commit 53c11a519e35bed2b260be5ad6503ec460e99eb2
Author: zhiwei.niu <[email protected]>
AuthorDate: Fri May 8 02:11:52 2026 +0800
[Fix][Zeta] Job stuck permanently after master failover, unable to complete
(affects BATCH / bounded source / job shutdown phase) (#10836)
Co-authored-by: Doyeon Kim <[email protected]>
---
.../e2e/CheckpointCoordinatorFailoverIT.java | 377 ++++++++++++++++
...fake_to_localfile_master_failover_template.conf | 154 +++++++
...fake_to_localfile_master_failover_template.conf | 189 ++++++++
.../server/checkpoint/CheckpointCoordinator.java | 234 +++++++---
.../seatunnel/engine/server/master/JobMaster.java | 6 +
.../checkpoint/CheckpointCoordinatorTest.java | 488 +++++++++++++++++++++
6 files changed, 1387 insertions(+), 61 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
new file mode 100644
index 0000000000..a176aed193
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.e2e;
+
+import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.ImmutablePair;
+
+import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.common.utils.FileUtils;
+import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
+import org.apache.seatunnel.engine.client.job.ClientJobProxy;
+import org.apache.seatunnel.engine.common.Constant;
+import org.apache.seatunnel.engine.common.config.ConfigProvider;
+import org.apache.seatunnel.engine.common.config.JobConfig;
+import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
+import org.apache.seatunnel.engine.server.checkpoint.IMapCheckpointIDCounter;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.instance.impl.HazelcastInstanceImpl;
+import com.hazelcast.map.IMap;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Slf4j
+public class CheckpointCoordinatorFailoverIT {
+
+ private static final String BATCH_TEMPLATE_CONF =
+ "batch_fake_to_localfile_master_failover_template.conf";
+
+ private static final String STREAM_TEMPLATE_CONF =
+ "stream_fake_to_localfile_master_failover_template.conf";
+
+ private static final String DYNAMIC_TEST_CASE_NAME =
"dynamic_test_case_name";
+
+ /** Must match the parallelism value set in the conf templates
(env.parallelism). */
+ private static final int SOURCE_PARALLELISM = 5;
+
+ @Test
+ public void testBatchJobCompletesAfterMasterFailover() throws Exception {
+ String testCaseName = "testBatchJobCompletesAfterMasterFailover";
+ String testClusterName =
+
"CheckpointCoordinatorFailoverIT_testBatchJobCompletesAfterMasterFailover";
+ // Per-source row.num must match
batch_fake_to_localfile_master_failover_template.conf.
+ // All sources use the same configuration (row.num=500) for stable
failover timing.
+ long rowNumPerSource = 500;
+ int sourceCount = 5;
+ final long expectedTotalRows = rowNumPerSource * sourceCount *
SOURCE_PARALLELISM;
+
+ HazelcastInstanceImpl masterNode1 = null;
+ HazelcastInstanceImpl masterNode2 = null;
+ SeaTunnelClient engineClient = null;
+
+ SeaTunnelConfig config1 = ConfigProvider.locateAndGetSeaTunnelConfig();
+
config1.getHazelcastConfig().setClusterName(TestUtils.getClusterName(testClusterName));
+ config1.getEngineConfig().getHttpConfig().setEnabled(false);
+
+ SeaTunnelConfig config2 = ConfigProvider.locateAndGetSeaTunnelConfig();
+
config2.getHazelcastConfig().setClusterName(TestUtils.getClusterName(testClusterName));
+ config2.getEngineConfig().getHttpConfig().setEnabled(false);
+
+ try {
+ masterNode1 =
SeaTunnelServerStarter.createHazelcastInstance(config1);
+ masterNode2 =
SeaTunnelServerStarter.createHazelcastInstance(config2);
+
+ HazelcastInstanceImpl finalMaster1 = masterNode1;
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 2,
finalMaster1.getCluster().getMembers().size()));
+
+ Common.setDeployMode(DeployMode.CLUSTER);
+ ImmutablePair<String, String> testResources =
+ createTestResources(testCaseName, BATCH_TEMPLATE_CONF);
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName(testCaseName);
+
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+
clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+ ClientJobExecutionEnvironment jobExecutionEnv =
+ engineClient.createExecutionContext(
+ testResources.getRight(), jobConfig, config1);
+ ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
+
+ long jobId = clientJobProxy.getJobId();
+ long triggerThreshold = expectedTotalRows / 4;
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
clientJobProxy.getJobStatus());
+ long observedRows =
+
FileUtils.getFileLineNumberFromDir(testResources.getLeft());
+ Assertions.assertTrue(
+ observedRows > triggerThreshold,
+ String.format(
+ "Waiting for sufficient output
before failover "
+ + "(rows=%d,
threshold=%d)",
+ observedRows,
triggerThreshold));
+ });
+
+ log.info(
+ "Job {} is RUNNING with over {} rows written. "
+ + "Triggering master failover by shutting down
masterNode1.",
+ jobId,
+ triggerThreshold);
+
+ masterNode1.shutdown();
+ masterNode1 = null;
+
+ HazelcastInstanceImpl finalMaster2 = masterNode2;
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 1,
finalMaster2.getCluster().getMembers().size()));
+
+ Awaitility.await()
+ .atMost(5, TimeUnit.MINUTES)
+ .pollInterval(3, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ JobStatus status =
clientJobProxy.getJobStatus();
+ Assertions.assertTrue(
+ status == JobStatus.RUNNING || status
== JobStatus.FINISHED,
+ "Waiting for job status to recover
after master failover, "
+ + "current status: "
+ + status);
+ });
+ Assertions.assertEquals(JobStatus.FINISHED,
clientJobProxy.waitForJobComplete());
+
+ long actualRows =
FileUtils.getFileLineNumberFromDir(testResources.getLeft());
+ Assertions.assertTrue(
+ actualRows >= expectedTotalRows,
+ String.format(
+ "Expected at least %d rows after failover, but got
%d",
+ expectedTotalRows, actualRows));
+ } finally {
+ if (engineClient != null) {
+ engineClient.close();
+ }
+ if (masterNode1 != null) {
+ masterNode1.shutdown();
+ }
+ if (masterNode2 != null) {
+ masterNode2.shutdown();
+ }
+ }
+ }
+
+ @Test
+ public void testStreamJobContinuesAfterMasterFailover() throws Exception {
+ String testCaseName = "testStreamJobContinuesAfterMasterFailover";
+ String testClusterName =
+
"CheckpointCoordinatorFailoverIT_testStreamJobContinuesAfterMasterFailover";
+ // Per-source row.num must match
stream_fake_to_localfile_master_failover_template.conf.
+ long[] rowNumPerSource = {100, 150, 200, 250, 300};
+ long maxBoundedRows = 0;
+ for (long rows : rowNumPerSource) {
+ maxBoundedRows += rows * SOURCE_PARALLELISM;
+ }
+
+ HazelcastInstanceImpl masterNode1 = null;
+ HazelcastInstanceImpl masterNode2 = null;
+ SeaTunnelClient engineClient = null;
+
+ SeaTunnelConfig config1 = ConfigProvider.locateAndGetSeaTunnelConfig();
+
config1.getHazelcastConfig().setClusterName(TestUtils.getClusterName(testClusterName));
+ config1.getEngineConfig().getHttpConfig().setEnabled(false);
+
+ SeaTunnelConfig config2 = ConfigProvider.locateAndGetSeaTunnelConfig();
+
config2.getHazelcastConfig().setClusterName(TestUtils.getClusterName(testClusterName));
+ config2.getEngineConfig().getHttpConfig().setEnabled(false);
+
+ try {
+ masterNode1 =
SeaTunnelServerStarter.createHazelcastInstance(config1);
+ masterNode2 =
SeaTunnelServerStarter.createHazelcastInstance(config2);
+
+ HazelcastInstanceImpl finalMaster1 = masterNode1;
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 2,
finalMaster1.getCluster().getMembers().size()));
+
+ Common.setDeployMode(DeployMode.CLUSTER);
+ ImmutablePair<String, String> testResources =
+ createTestResources(testCaseName, STREAM_TEMPLATE_CONF);
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName(testCaseName);
+
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+
clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+ ClientJobExecutionEnvironment jobExecutionEnv =
+ engineClient.createExecutionContext(
+ testResources.getRight(), jobConfig, config1);
+ ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
+
+ long jobId = clientJobProxy.getJobId();
+
+ // Trigger failover after ~1/4 of the bounded data has been
written; FakeSource
+ // in STREAMING mode is UNBOUNDED, so total rows are still bounded
by row.num
+ // per split but the source itself never finishes.
+ long triggerThreshold = maxBoundedRows / 4;
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
clientJobProxy.getJobStatus());
+ Assertions.assertTrue(
+
FileUtils.getFileLineNumberFromDir(testResources.getLeft())
+ > triggerThreshold);
+ });
+
+ log.info(
+ "Over {} rows written for streaming job {}. "
+ + "Triggering master failover by shutting down
masterNode1.",
+ triggerThreshold,
+ jobId);
+
+ masterNode1.shutdown();
+ masterNode1 = null;
+
+ HazelcastInstanceImpl finalMaster2 = masterNode2;
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 1,
finalMaster2.getCluster().getMembers().size()));
+
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .pollInterval(3, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
clientJobProxy.getJobStatus()));
+
+ // Verify at least one pipeline's checkpoint id strictly grows on
the new master.
+ IMap<String, Long> ckIdMap =
masterNode2.getMap(Constant.IMAP_CHECKPOINT_ID);
+ Map<Integer, Long> checkpointBefore = new HashMap<>();
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ checkpointBefore.clear();
+ for (int pipelineId = 1;
+ pipelineId <= rowNumPerSource.length;
+ pipelineId++) {
+ String ckIdKey =
+
IMapCheckpointIDCounter.convertLongIntToBase64(
+ jobId, pipelineId);
+ Long value = ckIdMap.get(ckIdKey);
+ if (value != null) {
+ checkpointBefore.put(pipelineId,
value);
+ }
+ }
+ Assertions.assertFalse(
+ checkpointBefore.isEmpty(),
+ "Waiting for checkpoint ids after
failover");
+ });
+
+ AtomicInteger observedPipelineId = new AtomicInteger(-1);
+ AtomicLong ckIdBefore = new AtomicLong(-1);
+ AtomicLong ckIdAfter = new AtomicLong(-1);
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ boolean grew = false;
+ for (Map.Entry<Integer, Long> entry :
checkpointBefore.entrySet()) {
+ int pid = entry.getKey();
+ long before = entry.getValue();
+ String ckIdKey =
+
IMapCheckpointIDCounter.convertLongIntToBase64(
+ jobId, pid);
+ Long current = ckIdMap.get(ckIdKey);
+ if (current != null && current > before) {
+ observedPipelineId.set(pid);
+ ckIdBefore.set(before);
+ ckIdAfter.set(current);
+ grew = true;
+ break;
+ }
+ }
+ Assertions.assertTrue(
+ grew,
+ "Checkpoint id should grow after
failover for at least"
+ + " one pipeline");
+ });
+ Assertions.assertTrue(
+ ckIdAfter.get() > ckIdBefore.get(),
+ String.format(
+ "Checkpoint id must continue to grow on the new
master for at least"
+ + " one pipeline (pipelineId=%d,
before=%d, after=%d)",
+ observedPipelineId.get(), ckIdBefore.get(),
ckIdAfter.get()));
+
+ clientJobProxy.cancelJob();
+ } finally {
+ if (engineClient != null) {
+ engineClient.close();
+ }
+ if (masterNode1 != null) {
+ masterNode1.shutdown();
+ }
+ if (masterNode2 != null) {
+ masterNode2.shutdown();
+ }
+ }
+ }
+
+ private ImmutablePair<String, String> createTestResources(
+ @NonNull String testCaseName, String templateConf) throws
IOException {
+ Map<String, String> valueMap = new HashMap<>();
+ valueMap.put(DYNAMIC_TEST_CASE_NAME, testCaseName);
+
+ String targetDir = "/tmp/hive/warehouse/" + testCaseName;
+ targetDir = targetDir.replace("/", File.separator);
+ FileUtils.createNewDir(targetDir);
+
+ String targetConfigFilePath =
+ File.separator
+ + "tmp"
+ + File.separator
+ + "test_conf"
+ + File.separator
+ + testCaseName
+ + ".conf";
+ TestUtils.createTestConfigFileFromTemplate(templateConf, valueMap,
targetConfigFilePath);
+
+ return new ImmutablePair<>(targetDir, targetConfigFilePath);
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/batch_fake_to_localfile_master_failover_template.conf
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/batch_fake_to_localfile_master_failover_template.conf
new file mode 100644
index 0000000000..e839786e59
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/batch_fake_to_localfile_master_failover_template.conf
@@ -0,0 +1,154 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ job.mode = BATCH
+ checkpoint.interval = 2000
+ parallelism = 5
+}
+
+source {
+ FakeSource {
+ plugin_output = table1
+ row.num = 500
+ split.num = 5
+ split.read-interval = 500
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table2
+ row.num = 500
+ split.num = 5
+ split.read-interval = 500
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table3
+ row.num = 500
+ split.num = 5
+ split.read-interval = 500
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table4
+ row.num = 500
+ split.num = 5
+ split.read-interval = 500
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table5
+ row.num = 500
+ split.num = 5
+ split.read-interval = 500
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ LocalFile {
+ plugin_input = table1
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table2
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table3
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table4
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table5
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/stream_fake_to_localfile_master_failover_template.conf
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/stream_fake_to_localfile_master_failover_template.conf
new file mode 100644
index 0000000000..e754ec3896
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/stream_fake_to_localfile_master_failover_template.conf
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ job.mode = STREAMING
+ checkpoint.interval = 2000
+ parallelism = 5
+}
+
+source {
+ FakeSource {
+ plugin_output = table1
+ row.num = 100
+ split.num = 2
+ split.read-interval = 500
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_boolean = boolean
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table2
+ row.num = 150
+ split.num = 3
+ split.read-interval = 750
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_boolean = boolean
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table3
+ row.num = 200
+ split.num = 4
+ split.read-interval = 1000
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_boolean = boolean
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table4
+ row.num = 250
+ split.num = 5
+ split.read-interval = 1500
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_boolean = boolean
+ }
+ }
+ }
+
+ FakeSource {
+ plugin_output = table5
+ row.num = 300
+ split.num = 6
+ split.read-interval = 2000
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_boolean = boolean
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ LocalFile {
+ plugin_input = table1
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table2
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table3
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table4
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+
+ LocalFile {
+ plugin_input = table5
+ path = "/tmp/hive/warehouse/${dynamic_test_case_name}"
+ field_delimiter = "\t"
+ row_delimiter = "\n"
+ file_name_expression = "${transactionId}_${now}"
+ file_format_type = "text"
+ filename_time_format = "yyyy.MM.dd"
+ is_enable_transaction = true
+ save_mode = "error"
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
index 26329c334c..9bbc6a39f2 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
@@ -158,6 +158,8 @@ public class CheckpointCoordinator {
private final String checkpointStateImapKey;
+ private final String readyToCloseImapKey;
+
@SneakyThrows
public CheckpointCoordinator(
CheckpointManager manager,
@@ -178,6 +180,7 @@ public class CheckpointCoordinator {
this.jobId = jobId;
this.pipelineId = plan.getPipelineId();
this.checkpointStateImapKey = "checkpoint_state_" + jobId + "_" +
pipelineId;
+ this.readyToCloseImapKey = checkpointStateImapKey + "_ready_to_close";
this.runningJobStateIMap = runningJobStateIMap;
this.plan = plan;
this.coordinatorConfig = checkpointConfig;
@@ -206,18 +209,18 @@ public class CheckpointCoordinator {
this.closedIdleTask = new CopyOnWriteArraySet<>();
LOG.info(
- "Create CheckpointCoordinator for job({}@{}) with plan({})",
- pipelineId,
+ "Create CheckpointCoordinator, job id: {}, pipeline id: {},
plan: {}",
jobId,
+ pipelineId,
plan);
if (pipelineState != null) {
this.latestCompletedCheckpoint =
serializer.deserialize(pipelineState.getStates(),
CompletedCheckpoint.class);
this.latestCompletedCheckpoint.setRestored(true);
LOG.info(
- "Restore job({}@{}) with checkpoint({}), data: {}",
- pipelineId,
+ "Restore checkpoint, job id: {}, pipeline id: {},
checkpoint id: {}, data: {} ",
jobId,
+ pipelineId,
latestCompletedCheckpoint.getCheckpointId(),
latestCompletedCheckpoint);
}
@@ -375,7 +378,7 @@ public class CheckpointCoordinator {
if (completedCheckpoint != null) {
try {
LOG.info(
- "start notify checkpoint completed, job id: {},
pipeline id: {}, checkpoint id:{}",
+ "start notify checkpoint completed, job id: {},
pipeline id: {}, checkpoint id: {}.",
completedCheckpoint.getJobId(),
completedCheckpoint.getPipelineId(),
completedCheckpoint.getCheckpointId());
@@ -427,18 +430,79 @@ public class CheckpointCoordinator {
protected void readyToClose(TaskLocation taskLocation) {
readyToCloseStartingTask.add(taskLocation);
+ updateReadyToCloseStartingTask();
if (readyToCloseStartingTask.size() ==
plan.getStartingSubtasks().size()) {
tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
}
}
+ private Set<TaskLocation> loadReadyToCloseStartingTask() {
+ try {
+ Object stored = runningJobStateIMap.get(readyToCloseImapKey);
+ if (stored instanceof Set) {
+ Set<TaskLocation> result = (Set<TaskLocation>) stored;
+ LOG.info(
+ "Loaded readyToCloseStartingTask from IMap, job id:
{}, pipeline id: {}, value: {}",
+ jobId,
+ pipelineId,
+ result);
+ return result;
+ }
+ return null;
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to load readyToCloseStartingTask from IMap, job
id: {}, pipeline id: {}.",
+ jobId,
+ pipelineId,
+ e);
+ throw new RuntimeException(
+ "Failed to load readyToCloseStartingTask from IMap, key: "
+ + readyToCloseImapKey,
+ e);
+ }
+ }
+
+ private void updateReadyToCloseStartingTask() {
+ try {
+ RetryUtils.retryWithException(
+ () -> {
+ runningJobStateIMap.compute(
+ readyToCloseImapKey,
+ (k, exist) -> {
+ Set<TaskLocation> merged =
+ exist instanceof Set
+ ? new
HashSet<>((Set<TaskLocation>) exist)
+ : new HashSet<>();
+ merged.addAll(readyToCloseStartingTask);
+ return merged;
+ });
+ return null;
+ },
+ new RetryUtils.RetryMaterial(
+ Constant.OPERATION_RETRY_TIME,
+ true,
+ ExceptionUtil::isOperationNeedRetryException,
+ Constant.OPERATION_RETRY_SLEEP));
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to persist readyToCloseStartingTask to IMap after
retries, key: {}."
+ + " Failing the job to avoid an unrecoverable
stuck state on master failover.",
+ readyToCloseImapKey,
+ e);
+ throw new RuntimeException(
+ "Failed to persist readyToCloseStartingTask to IMap, key: "
+ + readyToCloseImapKey,
+ e);
+ }
+ }
+
protected void readyToCloseIdleTask(TaskLocation taskLocation) {
if (plan.getStartingSubtasks().contains(taskLocation)) {
throw new UnsupportedOperationException("Unsupported close
starting task");
}
LOG.info(
- "Received close idle task[{}]({}/{}). {}",
+ "Received close idle task, task id: {}, pipeline id: {}, job
id: {}, detail: {}",
taskLocation.getTaskID(),
taskLocation.getPipelineId(),
taskLocation.getJobId(),
@@ -447,7 +511,7 @@ public class CheckpointCoordinator {
if (readyToCloseIdleTask.contains(taskLocation)
|| closedIdleTask.contains(taskLocation)) {
LOG.warn(
- "task[{}]({}/{}) already in closed. {}",
+ "task already in closed, task id: {}, pipeline id: {},
job id: {}, detail: {}",
taskLocation.getTaskID(),
taskLocation.getPipelineId(),
taskLocation.getJobId(),
@@ -461,7 +525,7 @@ public class CheckpointCoordinator {
// close all subtask in the same task group
subTaskList.add(subTask);
LOG.info(
- "Add task[{}]({}/{}) to prepare close list",
+ "Add task to prepare close list, task id: {},
pipeline id: {}, job id: {}",
subTask.getTaskID(),
subTask.getPipelineId(),
subTask.getJobId());
@@ -477,7 +541,7 @@ public class CheckpointCoordinator {
readyToCloseIdleTask.remove(taskLocation);
closedIdleTask.add(taskLocation);
LOG.info(
- "Completed close task[{}]({}/{})",
+ "Completed close task, task id: {}, pipeline id: {},
job id: {}",
taskLocation.getTaskID(),
taskLocation.getPipelineId(),
taskLocation.getJobId());
@@ -486,18 +550,38 @@ public class CheckpointCoordinator {
}
protected void restoreCoordinator(boolean alreadyStarted) {
- LOG.info("received restore CheckpointCoordinator with alreadyStarted =
{}", alreadyStarted);
+ LOG.info("received restore CheckpointCoordinator with alreadyStarted:
{}", alreadyStarted);
errorByPhysicalVertex = new AtomicReference<>();
checkpointCoordinatorFuture = new CompletableFuture<>();
updateStatus(CheckpointCoordinatorStatus.RUNNING);
+
+ Set<TaskLocation> restoredReadyToClose =
loadReadyToCloseStartingTask();
+
cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_RESET);
shutdown = false;
+
+ if (restoredReadyToClose != null && !restoredReadyToClose.isEmpty()) {
+ readyToCloseStartingTask.addAll(restoredReadyToClose);
+ LOG.info(
+ "Restored readyToCloseStartingTask, restored count: {}, "
+ + "total starting subtasks: {}, job id: {},
pipeline id: {}",
+ readyToCloseStartingTask.size(),
+ plan.getStartingSubtasks().size(),
+ jobId,
+ pipelineId);
+ }
+
if (alreadyStarted) {
isAllTaskReady.set(true);
if (!notifyCompleted(latestCompletedCheckpoint)) {
return;
}
- tryTriggerPendingCheckpoint(CHECKPOINT_TYPE);
+ if (readyToCloseStartingTask.size() ==
plan.getStartingSubtasks().size()) {
+ // All sources already finished before failover; complete the
job now.
+
tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+ } else {
+ tryTriggerPendingCheckpoint(CHECKPOINT_TYPE);
+ }
} else {
isAllTaskReady.set(false);
}
@@ -517,13 +601,16 @@ public class CheckpointCoordinator {
long interval = currentTimestamp - latestTriggerTimestamp.get();
if (interval <= 0) {
LOG.error(
- "The time on your server may not be incremental which
can lead checkpoint to stop. The latestTriggerTimestamp: ({}), but the
currentTimestamp: ({})",
+ "The time on your server may not be incremental which
can lead checkpoint to stop. "
+ + "The latestTriggerTimestamp: ({}), but the
currentTimestamp: ({})",
latestTriggerTimestamp.get(),
currentTimestamp);
}
if (interval < coordinatorConfig.getCheckpointInterval()) {
LOG.info(
- "skip trigger checkpoint because the last trigger
timestamp is {} and current timestamp is {}, the interval is less than config.",
+ "skip trigger checkpoint "
+ + "because the last trigger timestamp is ({})
and current timestamp is ({}), "
+ + "the interval is less than config.",
latestTriggerTimestamp.get(),
currentTimestamp);
scheduleTriggerPendingCheckpoint(
@@ -539,7 +626,9 @@ public class CheckpointCoordinator {
long minPauseDelay =
coordinatorConfig.getCheckpointMinPause() -
timeSinceLastCompleted;
LOG.info(
- "skip trigger checkpoint because the last
completed timestamp is {} and current timestamp is {}, the time since
completion ({} ms) is less than min-pause ({} ms).",
+ "skip trigger checkpoint "
+ + "because the last completed timestamp is
{} and current timestamp is {}, "
+ + "the time since completion ({} ms) is
less than min-pause ({} ms).",
lastCompletedTime,
currentTimestamp,
timeSinceLastCompleted,
@@ -552,13 +641,13 @@ public class CheckpointCoordinator {
synchronized (lock) {
if (isCompleted() || isShutdown()) {
LOG.warn(
- String.format(
- "can't trigger checkpoint with type: %s,
because checkpoint coordinator already have last completed checkpoint: (%s) or
shutdown (%b).",
- checkpointType,
- latestCompletedCheckpoint != null
- ?
latestCompletedCheckpoint.getCheckpointType()
- : "null",
- shutdown));
+ "can't trigger checkpoint with type: {}, because
checkpoint coordinator"
+ + " already have last completed checkpoint:
({}) or shutdown ({}).",
+ checkpointType,
+ latestCompletedCheckpoint != null
+ ? latestCompletedCheckpoint.getCheckpointType()
+ : "null",
+ shutdown);
return;
}
@@ -601,7 +690,7 @@ public class CheckpointCoordinator {
@SneakyThrows
public PassiveCompletableFuture<CompletedCheckpoint> startSavepoint() {
- LOG.info(String.format("Start save point for Job (%s)", jobId));
+ LOG.info("start save point for job id: {}.", jobId);
if (shutdown || isCompleted()) {
return completableFutureWithError(
CheckpointCloseReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
@@ -628,9 +717,10 @@ public class CheckpointCoordinator {
}
savepointPendingCheckpoint = savepoint.join();
LOG.info(
- String.format(
- "The save point checkpointId is %s",
- savepointPendingCheckpoint.getCheckpointId()));
+ "save point checkpoint is created, job id: {}, pipeline id:
{}, checkpoint id: {}.",
+ jobId,
+ pipelineId,
+ savepointPendingCheckpoint.getCheckpointId());
return savepointPendingCheckpoint.getCompletableFuture();
}
@@ -645,7 +735,9 @@ public class CheckpointCoordinator {
CompletableFuture<PendingCheckpoint> pendingCompletableFuture) {
pendingCompletableFuture.thenAccept(
pendingCheckpoint -> {
- LOG.info("wait checkpoint completed: {}",
pendingCheckpoint.getCheckpointId());
+ LOG.info(
+ "wait checkpoint id: {} completed.",
+ pendingCheckpoint.getCheckpointId());
PassiveCompletableFuture<CompletedCheckpoint>
completableFuture =
pendingCheckpoint.getCompletableFuture();
completableFuture.whenCompleteAsync(
@@ -897,6 +989,12 @@ public class CheckpointCoordinator {
closedIdleTask.clear();
pendingCounter.set(0);
schemaChanging.set(false);
+ // Only remove the persisted ready-to-close state when the
coordinator truly ends
+ // (completed/failed/cancelled). During a reset (master failover),
the IMap entry
+ // must be preserved so restoreCoordinator() can recover from it.
+ if (closedReason !=
CheckpointCloseReason.CHECKPOINT_COORDINATOR_RESET) {
+ runningJobStateIMap.remove(readyToCloseImapKey);
+ }
scheduler.shutdownNow();
scheduler =
Executors.newScheduledThreadPool(
@@ -919,12 +1017,12 @@ public class CheckpointCoordinator {
final long checkpointId = ackOperation.getBarrier().getId();
final PendingCheckpoint pendingCheckpoint =
pendingCheckpoints.get(checkpointId);
if (pendingCheckpoint == null) {
- LOG.info("skip already ack checkpoint {}", checkpointId);
+ LOG.info("skip already ack checkpoint id: {}", checkpointId);
return;
}
TaskLocation location = ackOperation.getTaskLocation();
LOG.debug(
- "task[{}]({}/{}) ack. {}",
+ "task ack, task id: {}, pipeline id: {}, job id: {}, barrier:
{}",
location.getTaskID(),
location.getPipelineId(),
location.getJobId(),
@@ -954,10 +1052,11 @@ public class CheckpointCoordinator {
public synchronized void completePendingCheckpoint(CompletedCheckpoint
completedCheckpoint) {
LOG.debug(
- "pending checkpoint({}/{}@{}) completed! cost: {}, trigger:
{}, completed: {}",
- completedCheckpoint.getCheckpointId(),
- completedCheckpoint.getPipelineId(),
+ "pending checkpoint completed, job id: {}, pipeline id: {},
checkpoint id: {}, "
+ + "cost: {}, trigger: {}, completed: {}",
completedCheckpoint.getJobId(),
+ completedCheckpoint.getPipelineId(),
+ completedCheckpoint.getCheckpointId(),
completedCheckpoint.getCompletedTimestamp()
- completedCheckpoint.getCheckpointTimestamp(),
completedCheckpoint.getCheckpointTimestamp(),
@@ -997,10 +1096,10 @@ public class CheckpointCoordinator {
sneakyThrow(e);
}
LOG.info(
- "pending checkpoint({}/{}@{}) notify finished!",
- completedCheckpoint.getCheckpointId(),
+ "pending checkpoint notify finished, job id: {}, pipeline id:
{}, checkpoint id: {}!",
+ completedCheckpoint.getJobId(),
completedCheckpoint.getPipelineId(),
- completedCheckpoint.getJobId());
+ completedCheckpoint.getCheckpointId());
latestCompletedCheckpoint = completedCheckpoint;
if (checkpointMonitorService != null) {
long stateSize =
CheckpointMonitorService.calculateStateSize(completedCheckpoint);
@@ -1105,11 +1204,10 @@ public class CheckpointCoordinator {
RetryUtils.retryWithException(
() -> {
LOG.info(
- String.format(
- "Turn %s state from %s to %s",
- checkpointStateImapKey,
-
runningJobStateIMap.get(checkpointStateImapKey),
- targetStatus));
+ "Turn {} state from {} to {}",
+ checkpointStateImapKey,
+
runningJobStateIMap.get(checkpointStateImapKey),
+ targetStatus);
runningJobStateIMap.set(checkpointStateImapKey,
targetStatus);
return null;
},
@@ -1120,58 +1218,68 @@ public class CheckpointCoordinator {
Constant.OPERATION_RETRY_SLEEP));
} catch (Exception e) {
LOG.warn(
- String.format(
- "Set %s state %s to IMap failed, skip do it",
- checkpointStateImapKey, targetStatus));
+ "Set {} state {} to IMap failed, skip do it",
+ checkpointStateImapKey,
+ targetStatus);
}
}
protected void scheduleSchemaChangeBeforeCheckpoint() {
if (schemaChanging.compareAndSet(false, true)) {
LOG.info(
- "stop trigger general-checkpoint({}@{}) because schema
change in progress.",
- pipelineId,
- jobId);
- LOG.info("schedule schema-change-before checkpoint({}@{}).",
pipelineId, jobId);
+ "stop trigger general-checkpoint "
+ + "because schema change in progress, job id: {},
pipeline id: {}.",
+ jobId,
+ pipelineId);
+ LOG.info(
+ "schedule schema-change-before checkpoint, job id: {},
pipeline id: {}.",
+ jobId,
+ pipelineId);
scheduleTriggerPendingCheckpoint(CheckpointType.SCHEMA_CHANGE_BEFORE_POINT_TYPE,
0);
} else {
LOG.warn(
- "schema-change-before checkpoint({}@{}) is already
scheduled.",
- pipelineId,
- jobId);
+ "schema-change-before checkpoint is already scheduled, job
id: {}, pipeline id: {}.",
+ jobId,
+ pipelineId);
}
}
protected void scheduleSchemaChangeAfterCheckpoint() {
if (schemaChanging.get()) {
- LOG.info("schedule schema-change-after checkpoint({}@{}).",
pipelineId, jobId);
+ LOG.info(
+ "schedule schema-change-after checkpoint, job id: {},
pipeline id: {}.",
+ jobId,
+ pipelineId);
scheduleTriggerPendingCheckpoint(CheckpointType.SCHEMA_CHANGE_AFTER_POINT_TYPE,
0);
} else {
LOG.warn(
- "schema-change-after checkpoint({}@{}) is already
scheduled.",
- pipelineId,
- jobId);
+ "schema-change-after checkpoint is already scheduled, job
id: {}, pipeline id: {}.",
+ jobId,
+ pipelineId);
}
}
protected void completeSchemaChangeAfterCheckpoint(CompletedCheckpoint
checkpoint) {
if (schemaChanging.compareAndSet(true, false)) {
LOG.info(
- "completed schema-change-after checkpoint({}/{}@{}).",
- checkpoint.getCheckpointId(),
+ "completed schema-change-after checkpoint, job id: {},
pipeline id: {}, "
+ + "checkpoint id: {}.",
+ jobId,
pipelineId,
- jobId);
+ checkpoint.getCheckpointId());
LOG.info(
- "recover trigger general-checkpoint({}/{}@{}).",
- checkpoint.getCheckpointId(),
+ "recover trigger general-checkpoint, job id: {}, pipeline
id: {}, "
+ + "checkpoint id: {}.",
+ jobId,
pipelineId,
- jobId);
+ checkpoint.getCheckpointId());
scheduleTriggerPendingCheckpoint(coordinatorConfig.getCheckpointInterval());
} else {
throw new IllegalStateException(
String.format(
- "schema-change-after checkpoint(%s/%s@%s) is
already completed.",
- checkpoint.getCheckpointId(), pipelineId, jobId));
+ "schema-change-after checkpoint is already
completed, "
+ + "job id: %s, pipeline id: %s, checkpoint
id: %s.",
+ jobId, pipelineId, checkpoint.getCheckpointId()));
}
}
@@ -1179,6 +1287,10 @@ public class CheckpointCoordinator {
return checkpointStateImapKey;
}
+ public String getReadyToCloseImapKey() {
+ return readyToCloseImapKey;
+ }
+
/** Only for test */
@VisibleForTesting
public PendingCheckpoint getSavepointPendingCheckpoint() {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
index 961832660c..6280225aff 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
@@ -661,6 +661,12 @@ public class JobMaster {
.getCheckpointCoordinator(pipeline.getPipelineId())
.getCheckpointStateImapKey();
runningJobStateIMap.remove(checkpointStateImapKey);
+
+ String readyToCloseImapKey =
+ checkpointManager
+
.getCheckpointCoordinator(pipeline.getPipelineId())
+ .getReadyToCloseImapKey();
+ runningJobStateIMap.remove(readyToCloseImapKey);
});
runningJobStateIMap.remove(jobId);
runningJobInfoIMap.remove(jobId);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorTest.java
index beb0ce381e..2ac5be59f7 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorTest.java
@@ -53,11 +53,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -375,6 +378,491 @@ public class CheckpointCoordinatorTest
executor.shutdownNow();
}
+ @Test
+ void testReadyToClosePartialProgressPersistedAndRestoredCorrectly() {
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ try {
+ TaskLocation task1 = new TaskLocation(new TaskGroupLocation(1L, 1,
1), 1, 1);
+ TaskLocation task2 = new TaskLocation(new TaskGroupLocation(1L, 1,
2), 2, 2);
+ TaskLocation task3 = new TaskLocation(new TaskGroupLocation(1L, 1,
3), 3, 3);
+ Set<TaskLocation> allStarting = new HashSet<>(Arrays.asList(task1,
task2, task3));
+
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(allStarting)
+ .startingSubtasks(allStarting)
+ .build();
+
+ IMap<Object, Object> realIMap =
+
nodeEngine.getHazelcastInstance().getMap(IMAP_RUNNING_JOB_STATE);
+ String readyToCloseKey = "checkpoint_state_1_1_ready_to_close";
+ realIMap.remove(readyToCloseKey);
+
+ CheckpointManager mockManager =
Mockito.mock(CheckpointManager.class);
+ CheckpointStorage mockStorage =
Mockito.mock(CheckpointStorage.class);
+ CheckpointIDCounter mockIdCounter =
Mockito.mock(CheckpointIDCounter.class);
+
+ // Phase 1: partial readyToClose (1 of 3)
+ CheckpointCoordinator coord1 =
+ new CheckpointCoordinator(
+ mockManager,
+ mockStorage,
+ checkpointConfig,
+ 1L,
+ plan,
+ mockIdCounter,
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy1 = Mockito.spy(coord1);
+ Mockito.doNothing()
+ .when(spy1)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ spy1.readyToClose(task1);
+
+ Object stored = realIMap.get(readyToCloseKey);
+ Assertions.assertInstanceOf(Set.class, stored);
+ Assertions.assertEquals(
+ 1,
+ ((Set<?>) stored).size(),
+ "After 1 of 3 readyToClose, IMap should contain exactly 1
task");
+ Mockito.verify(spy1, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+
+ // Phase 2: restore with partial progress → should trigger normal
checkpoint
+ CheckpointCoordinator coord2 =
+ new CheckpointCoordinator(
+ mockManager,
+ mockStorage,
+ checkpointConfig,
+ 1L,
+ plan,
+ mockIdCounter,
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy2 = Mockito.spy(coord2);
+ Mockito.doReturn(true).when(spy2).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(spy2)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ spy2.restoreCoordinator(true);
+
+
Mockito.verify(spy2).tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+ Mockito.verify(spy2, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+
+ // Phase 3: all tasks readyToClose (3 of 3)
+ spy1.readyToClose(task2);
+ spy1.readyToClose(task3);
+
+ stored = realIMap.get(readyToCloseKey);
+ Assertions.assertInstanceOf(Set.class, stored);
+ Assertions.assertEquals(
+ 3,
+ ((Set<?>) stored).size(),
+ "After 3 of 3 readyToClose, IMap should contain all 3
tasks");
+
Mockito.verify(spy1).tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+
+ // Phase 4: restore with full progress → should trigger
COMPLETED_POINT
+ CheckpointCoordinator coord3 =
+ new CheckpointCoordinator(
+ mockManager,
+ mockStorage,
+ checkpointConfig,
+ 1L,
+ plan,
+ mockIdCounter,
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy3 = Mockito.spy(coord3);
+ Mockito.doReturn(true).when(spy3).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(spy3)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ spy3.restoreCoordinator(true);
+
+
Mockito.verify(spy3).tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+ Mockito.verify(spy3, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void
testRestoreCoordinatorShouldBeIdempotentWithPartialReadyToCloseProgress() {
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ try {
+ TaskLocation task1 = new TaskLocation(new TaskGroupLocation(1L, 1,
1), 1, 1);
+ TaskLocation task2 = new TaskLocation(new TaskGroupLocation(1L, 1,
2), 2, 2);
+ TaskLocation task3 = new TaskLocation(new TaskGroupLocation(1L, 1,
3), 3, 3);
+ Set<TaskLocation> allStarting = new HashSet<>(Arrays.asList(task1,
task2, task3));
+
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(allStarting)
+ .startingSubtasks(allStarting)
+ .build();
+
+ IMap<Object, Object> realIMap =
+
nodeEngine.getHazelcastInstance().getMap(IMAP_RUNNING_JOB_STATE);
+ String readyToCloseKey = "checkpoint_state_1_1_ready_to_close";
+ realIMap.put(readyToCloseKey, new
HashSet<>(Collections.singleton(task1)));
+
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinator(
+ Mockito.mock(CheckpointManager.class),
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy = Mockito.spy(coordinator);
+ Mockito.doReturn(true).when(spy).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(spy)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ Assertions.assertDoesNotThrow(() -> spy.restoreCoordinator(true));
+ Assertions.assertDoesNotThrow(() -> spy.restoreCoordinator(true));
+
+ Mockito.verify(spy, Mockito.times(2))
+
.tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+ Mockito.verify(spy, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+
+ Set<TaskLocation> restoredReadyToClose =
+ (Set<TaskLocation>)
+ ReflectionUtils.getField(spy,
"readyToCloseStartingTask")
+ .orElse(Collections.emptySet());
+ Assertions.assertEquals(
+ 1,
+ restoredReadyToClose.size(),
+ "Repeated restore should not duplicate partial
readyToClose progress");
+ Assertions.assertTrue(restoredReadyToClose.contains(task1));
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void
testRestoreCoordinatorShouldFallbackToCheckpointWhenReadyToCloseKeyMissing() {
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ try {
+ TaskLocation task1 = new TaskLocation(new TaskGroupLocation(1L, 1,
1), 1, 1);
+ TaskLocation task2 = new TaskLocation(new TaskGroupLocation(1L, 1,
2), 2, 2);
+ Set<TaskLocation> allStarting = new HashSet<>(Arrays.asList(task1,
task2));
+
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(allStarting)
+ .startingSubtasks(allStarting)
+ .build();
+
+ IMap<Object, Object> realIMap =
+
nodeEngine.getHazelcastInstance().getMap(IMAP_RUNNING_JOB_STATE);
+ String readyToCloseKey = "checkpoint_state_1_1_ready_to_close";
+ realIMap.remove(readyToCloseKey);
+
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinator(
+ Mockito.mock(CheckpointManager.class),
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy = Mockito.spy(coordinator);
+ Mockito.doReturn(true).when(spy).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(spy)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ Assertions.assertDoesNotThrow(() -> spy.restoreCoordinator(true));
+
Mockito.verify(spy).tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+ Mockito.verify(spy, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void
testRestoreCoordinatorShouldFallbackToCheckpointWhenReadyToCloseValueCorrupted()
{
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ try {
+ TaskLocation task1 = new TaskLocation(new TaskGroupLocation(1L, 1,
1), 1, 1);
+ TaskLocation task2 = new TaskLocation(new TaskGroupLocation(1L, 1,
2), 2, 2);
+ Set<TaskLocation> allStarting = new HashSet<>(Arrays.asList(task1,
task2));
+
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(allStarting)
+ .startingSubtasks(allStarting)
+ .build();
+
+ IMap<Object, Object> realIMap =
+
nodeEngine.getHazelcastInstance().getMap(IMAP_RUNNING_JOB_STATE);
+ String readyToCloseKey = "checkpoint_state_1_1_ready_to_close";
+ realIMap.set(readyToCloseKey, "corrupted_ready_to_close_payload");
+
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinator(
+ Mockito.mock(CheckpointManager.class),
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ executorService,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy = Mockito.spy(coordinator);
+ Mockito.doReturn(true).when(spy).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(spy)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ Assertions.assertDoesNotThrow(() -> spy.restoreCoordinator(true));
+
Mockito.verify(spy).tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+ Mockito.verify(spy, Mockito.never())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void testUpdateReadyToCloseStartingTaskWithTenTaskLocationsConcurrently()
throws Exception {
+ ExecutorService coordinatorExecutor = Executors.newCachedThreadPool();
+ ExecutorService concurrentExecutor = Executors.newFixedThreadPool(10);
+ try {
+ Set<TaskLocation> allStarting = new HashSet<>();
+ for (int i = 1; i <= 10; i++) {
+ allStarting.add(new TaskLocation(new TaskGroupLocation(1L, 1,
i), i, i));
+ }
+
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(allStarting)
+ .startingSubtasks(allStarting)
+ .build();
+
+ IMap<Object, Object> realIMap = Mockito.mock(IMap.class);
+ String readyToCloseKey = "checkpoint_state_1_1_ready_to_close";
+ Map<Object, Object> simulatedImapStorage = new
ConcurrentHashMap<>();
+ Mockito.when(realIMap.get(Mockito.any()))
+ .thenAnswer(invocation ->
simulatedImapStorage.get(invocation.getArgument(0)));
+
+ CountDownLatch firstSnapshotReady = new CountDownLatch(1);
+ CountDownLatch fullSnapshotWritten = new CountDownLatch(1);
+ CountDownLatch allowFirstSnapshotWrite = new CountDownLatch(1);
+ AtomicBoolean blockedFirstSnapshot = new AtomicBoolean(false);
+
+ AtomicInteger computeCallCount = new AtomicInteger(0);
+ AtomicInteger setCallCount = new AtomicInteger(0);
+
+ Mockito.doAnswer(
+ invocation -> {
+ Object key = invocation.getArgument(0);
+ java.util.function.BiFunction<Object, Object,
Object>
+ remappingFunction =
invocation.getArgument(1);
+ if (!readyToCloseKey.equals(key)) {
+ return simulatedImapStorage.compute(key,
remappingFunction);
+ }
+ computeCallCount.incrementAndGet();
+ if (blockedFirstSnapshot.compareAndSet(false,
true)) {
+ firstSnapshotReady.countDown();
+ if (!allowFirstSnapshotWrite.await(30,
TimeUnit.SECONDS)) {
+ throw new AssertionError(
+ "Timed out while waiting to
release first snapshot write");
+ }
+ }
+ Object result =
+ simulatedImapStorage.compute(key,
remappingFunction);
+ @SuppressWarnings("unchecked")
+ Set<TaskLocation> snapshot =
(Set<TaskLocation>) result;
+ if (snapshot.size() == allStarting.size()) {
+ fullSnapshotWritten.countDown();
+ }
+ return result;
+ })
+ .when(realIMap)
+ .compute(Mockito.any(), Mockito.any());
+
+ Mockito.doAnswer(
+ invocation -> {
+ Object key = invocation.getArgument(0);
+ Object value = invocation.getArgument(1);
+ if (readyToCloseKey.equals(key)) {
+ setCallCount.incrementAndGet();
+ }
+ simulatedImapStorage.put(key, value);
+ return null;
+ })
+ .when(realIMap)
+ .set(Mockito.any(), Mockito.any());
+
+ CheckpointCoordinator coordinator =
+ new CheckpointCoordinator(
+ Mockito.mock(CheckpointManager.class),
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ coordinatorExecutor,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator spy = Mockito.spy(coordinator);
+ Mockito.doNothing()
+ .when(spy)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+
+ CountDownLatch finishLatch = new
CountDownLatch(allStarting.size());
+ List<Future<?>> futures = new ArrayList<>(allStarting.size());
+ List<TaskLocation> taskLocations = new ArrayList<>(allStarting);
+ TaskLocation firstTask = taskLocations.get(0);
+
+ Future<?> firstFuture =
+ concurrentExecutor.submit(
+ () -> {
+ try {
+ spy.readyToClose(firstTask);
+ } finally {
+ finishLatch.countDown();
+ }
+ });
+ futures.add(firstFuture);
+
+ Assertions.assertTrue(
+ firstSnapshotReady.await(30, TimeUnit.SECONDS),
+ "Should observe the first snapshot write");
+
+ for (int i = 1; i < taskLocations.size(); i++) {
+ TaskLocation taskLocation = taskLocations.get(i);
+ Future<?> future =
+ concurrentExecutor.submit(
+ () -> {
+ try {
+ spy.readyToClose(taskLocation);
+ } finally {
+ finishLatch.countDown();
+ }
+ });
+ futures.add(future);
+ }
+
+ Assertions.assertTrue(
+ fullSnapshotWritten.await(30, TimeUnit.SECONDS),
+ "Should observe full snapshot write before releasing first
write");
+ allowFirstSnapshotWrite.countDown();
+ Assertions.assertTrue(
+ finishLatch.await(30, TimeUnit.SECONDS),
+ "All concurrent readyToClose calls should finish in time");
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+
+ Object persistedValue = simulatedImapStorage.get(readyToCloseKey);
+ Assertions.assertInstanceOf(
+ Set.class, persistedValue, "Persisted ready-to-close state
should be a Set");
+ Set<TaskLocation> persisted = (Set<TaskLocation>) persistedValue;
+ Assertions.assertEquals(10, persisted.size());
+ Assertions.assertTrue(persisted.containsAll(allStarting));
+ Mockito.verify(spy, Mockito.atLeastOnce())
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+
+ Assertions.assertTrue(
+ computeCallCount.get() >= 10,
+ "Production code must use atomic compute() for
ready-to-close persistence,"
+ + " actual compute calls: "
+ + computeCallCount.get());
+ Assertions.assertEquals(
+ 0,
+ setCallCount.get(),
+ "Production code must NOT use non-atomic set() for
ready-to-close persistence;"
+ + " set() would cause lost-update under
concurrency");
+
+ // Restore from persisted IMap state and verify merged set is
fully recoverable.
+ CheckpointCoordinator restoredCoordinator =
+ new CheckpointCoordinator(
+ Mockito.mock(CheckpointManager.class),
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ coordinatorExecutor,
+ realIMap,
+ false,
+ null);
+ CheckpointCoordinator restoreSpy =
Mockito.spy(restoredCoordinator);
+
Mockito.doReturn(true).when(restoreSpy).notifyCompleted(Mockito.any());
+ Mockito.doNothing()
+ .when(restoreSpy)
+
.tryTriggerPendingCheckpoint(Mockito.any(CheckpointType.class));
+ restoreSpy.restoreCoordinator(true);
+
+ Set<TaskLocation> restoredReadyToClose =
+ (Set<TaskLocation>)
+ ReflectionUtils.getField(restoreSpy,
"readyToCloseStartingTask")
+ .orElse(Collections.emptySet());
+ Assertions.assertEquals(
+ allStarting,
+ restoredReadyToClose,
+ "Restored ready-to-close state should recover the full
merged set");
+ Mockito.verify(restoreSpy)
+
.tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
+ } finally {
+ concurrentExecutor.shutdownNow();
+ coordinatorExecutor.shutdownNow();
+ }
+ }
+
// ------------------------------------------------------------------
// Regression tests: notifyCompleted returns false → callers bail out
// ------------------------------------------------------------------
