This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 d33bd82a4d [Fix][Zeta] Restore jobs from their own latest checkpoint
on master failover (#12387)
d33bd82a4d is described below
commit d33bd82a4df68f2d981923b57804ad35455c00ea
Author: 88fantasy <[email protected]>
AuthorDate: Fri Oct 23 00:30:27 2026 +0800
[Fix][Zeta] Restore jobs from their own latest checkpoint on master
failover (#12387)
---
.../server/checkpoint/CheckpointManager.java | 65 ++++++++++++++--
.../seatunnel/engine/server/master/JobMaster.java | 1 +
.../server/checkpoint/CheckpointManagerTest.java | 89 ++++++++++++++++++++++
3 files changed, 147 insertions(+), 8 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
index e1adaad108..5a85a16d17 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
@@ -107,6 +107,44 @@ public class CheckpointManager {
IMap<Object, Object> runningJobStateIMap,
SeaTunnelEngineContext engineContext,
CheckpointMonitorService checkpointMonitorService) {
+ this(
+ jobId,
+ isRestoreJob,
+ restoreMode,
+ restoreSourceJobId,
+ false,
+ nodeEngine,
+ jobMaster,
+ checkpointPlanMap,
+ checkpointConfig,
+ checkpointStorage,
+ executorService,
+ runningJobStateIMap,
+ engineContext,
+ checkpointMonitorService);
+ }
+
+ /**
+ * @param masterFailoverRestore true when the job is being restored after
a master node
+ * switch/restart. The latest completed checkpoint of this job itself
is preferred in that
+ * case, falling back to {@code restoreSourceJobId}/{@code
restoreMode} only when this job
+ * has not completed any checkpoint yet.
+ */
+ public CheckpointManager(
+ long jobId,
+ boolean isRestoreJob,
+ RestoreMode restoreMode,
+ Long restoreSourceJobId,
+ boolean masterFailoverRestore,
+ NodeEngine nodeEngine,
+ JobMaster jobMaster,
+ Map<Integer, CheckpointPlan> checkpointPlanMap,
+ CheckpointConfig checkpointConfig,
+ CheckpointStorage checkpointStorage,
+ ExecutorService executorService,
+ IMap<Object, Object> runningJobStateIMap,
+ SeaTunnelEngineContext engineContext,
+ CheckpointMonitorService checkpointMonitorService) {
this.jobId = jobId;
this.nodeEngine = nodeEngine;
this.jobMaster = jobMaster;
@@ -128,14 +166,25 @@ public class CheckpointManager {
try {
idCounter.start();
PipelineState pipelineState = null;
- if
(checkpointConfig.isCheckpointEnable()
- && isRestoreJob
- && restoreSourceJobId != null)
{
- pipelineState =
-
getLatestCheckpointStateByType(
-
String.valueOf(restoreSourceJobId),
-
String.valueOf(plan.getPipelineId()),
- restoreMode);
+ if
(checkpointConfig.isCheckpointEnable() && isRestoreJob) {
+ if (masterFailoverRestore) {
+ pipelineState =
+ checkpointStorage
+
.getLatestCheckpointByJobIdAndPipelineId(
+
String.valueOf(jobId),
+
String.valueOf(
+
plan
+
.getPipelineId()));
+ }
+ if (pipelineState == null
+ && restoreSourceJobId !=
null) {
+ pipelineState =
+
getLatestCheckpointStateByType(
+
String.valueOf(restoreSourceJobId),
+ String.valueOf(
+
plan.getPipelineId()),
+ restoreMode);
+ }
if (pipelineState != null) {
long checkpointId =
pipelineState.getCheckpointId();
idCounter.setCount(checkpointId + 1);
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 e3df6760d2..d9068ddf40 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
@@ -341,6 +341,7 @@ public class JobMaster {
jobImmutableInformation.isRestoreJob() || restart,
jobImmutableInformation.getRestoreMode(),
jobImmutableInformation.getRestoreSourceJobId(),
+ restart,
nodeEngine,
this,
checkpointPlanMap,
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManagerTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManagerTest.java
index a1aecbddd7..d19f2835a6 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManagerTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManagerTest.java
@@ -135,6 +135,76 @@ public class CheckpointManagerTest extends
AbstractSeaTunnelServerTest {
checkpointManager.getCheckpointCoordinator(1).getCheckpointIdCounter().get());
}
+ @Test
+ public void testMasterFailoverRestoreShouldLoadLatestCheckpointOfOwnJob()
throws Exception {
+ long jobId = (long) (Math.random() * 1000000L);
+ CheckpointStorage checkpointStorage = createCheckpointStorage();
+ storeCompletedCheckpoint(checkpointStorage, jobId, 7L);
+
+ // A normally submitted job has no restore source; after a master
switch it is
+ // re-initialized with restart = true and must still continue from its
own checkpoint.
+ CheckpointManager checkpointManager =
+ createCheckpointManager(
+ jobId,
+ true,
+ RestoreMode.NONE,
+ null,
+ true,
+ checkpointStorage,
+ new CheckpointConfig());
+
+ Assertions.assertEquals(
+ 8L,
checkpointManager.getCheckpointCoordinator(1).getCheckpointIdCounter().get());
+ }
+
+ @Test
+ public void
testMasterFailoverRestoreShouldPreferOwnLatestCheckpointOverSavepoint()
+ throws Exception {
+ long jobId = (long) (Math.random() * 1000000L);
+ CheckpointStorage checkpointStorage = createCheckpointStorage();
+ storeCompletedCheckpoint(checkpointStorage, jobId, 3L,
CheckpointType.SAVEPOINT_TYPE);
+ // checkpoint files are ordered by their store timestamp (millis)
+ Thread.sleep(5);
+ storeCompletedCheckpoint(checkpointStorage, jobId, 7L);
+
+ // Job started with -r and later hit a master switch: resume from
checkpoint 7, not
+ // from the savepoint it was originally restored from.
+ CheckpointManager checkpointManager =
+ createCheckpointManager(
+ jobId,
+ true,
+ RestoreMode.SAVEPOINT,
+ jobId,
+ true,
+ checkpointStorage,
+ new CheckpointConfig());
+
+ Assertions.assertEquals(
+ 8L,
checkpointManager.getCheckpointCoordinator(1).getCheckpointIdCounter().get());
+ }
+
+ @Test
+ public void
testMasterFailoverRestoreShouldFallBackToSourceJobWithoutOwnCheckpoint()
+ throws Exception {
+ long jobId = (long) (Math.random() * 1000000L);
+ long sourceJobId = jobId + 1;
+ CheckpointStorage checkpointStorage = createCheckpointStorage();
+ storeCompletedCheckpoint(checkpointStorage, sourceJobId, 5L);
+
+ CheckpointManager checkpointManager =
+ createCheckpointManager(
+ jobId,
+ true,
+ RestoreMode.CHECKPOINT,
+ sourceJobId,
+ true,
+ checkpointStorage,
+ new CheckpointConfig());
+
+ Assertions.assertEquals(
+ 6L,
checkpointManager.getCheckpointCoordinator(1).getCheckpointIdCounter().get());
+ }
+
private CheckpointStorage createCheckpointStorage() throws
CheckpointStorageException {
return FactoryUtil.discoverFactory(
Thread.currentThread().getContextClassLoader(),
@@ -196,6 +266,24 @@ public class CheckpointManagerTest extends
AbstractSeaTunnelServerTest {
Long restoreSourceJobId,
CheckpointStorage checkpointStorage,
CheckpointConfig checkpointConfig) {
+ return createCheckpointManager(
+ jobId,
+ isRestoreJob,
+ restoreMode,
+ restoreSourceJobId,
+ false,
+ checkpointStorage,
+ checkpointConfig);
+ }
+
+ private CheckpointManager createCheckpointManager(
+ long jobId,
+ boolean isRestoreJob,
+ RestoreMode restoreMode,
+ Long restoreSourceJobId,
+ boolean masterFailoverRestore,
+ CheckpointStorage checkpointStorage,
+ CheckpointConfig checkpointConfig) {
Map<Integer, CheckpointPlan> planMap = new HashMap<>();
planMap.put(1, CheckpointPlan.builder().pipelineId(1).build());
return new CheckpointManager(
@@ -203,6 +291,7 @@ public class CheckpointManagerTest extends
AbstractSeaTunnelServerTest {
isRestoreJob,
restoreMode,
restoreSourceJobId,
+ masterFailoverRestore,
nodeEngine,
null,
planMap,
