This is an automated email from the ASF dual-hosted git repository.
nielifeng 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 229e006bd3 [Fix][Zeta] Handle missing checkpoint state during restore
(#11565)
229e006bd3 is described below
commit 229e006bd33251304c7e5609af0b917214b0f3b4
Author: lifeng <[email protected]>
AuthorDate: Wed Jul 29 14:37:18 2026 +0800
[Fix][Zeta] Handle missing checkpoint state during restore (#11565)
Signed-off-by: lifeng <[email protected]>
---
.../server/checkpoint/CheckpointCoordinator.java | 4 +-
.../checkpoint/CheckpointCoordinatorTest.java | 58 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 2 deletions(-)
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 ba557d3fb4..7a0fcd9983 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
@@ -1396,8 +1396,8 @@ public class CheckpointCoordinator {
CheckpointCoordinatorStatus status =
(CheckpointCoordinatorStatus)
runningJobStateIMap.get(checkpointStateImapKey);
return
latestCompletedCheckpoint.getCheckpointType().isFinalCheckpoint()
- && (status.equals(CheckpointCoordinatorStatus.FINISHED)
- || status.equals(CheckpointCoordinatorStatus.SUSPEND))
+ && (CheckpointCoordinatorStatus.FINISHED.equals(status)
+ || CheckpointCoordinatorStatus.SUSPEND.equals(status))
&& !latestCompletedCheckpoint.isRestored();
}
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 2e80d89d12..ff54efee8b 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
@@ -909,6 +909,64 @@ public class CheckpointCoordinatorTest
null);
}
+ @Test
+ void testIsNoErrorCompletedUsesCheckpointStateMatrix() {
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ try {
+ CheckpointCoordinator coordinator =
buildMinimalCoordinator(executorService);
+ @SuppressWarnings("unchecked")
+ IMap<Object, Object> runningJobStateIMap =
+ (IMap<Object, Object>)
+ ReflectionUtils.getField(coordinator,
"runningJobStateIMap")
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+
"runningJobStateIMap field not found"));
+ CheckpointCoordinatorStatus[] states = {
+ null,
+ CheckpointCoordinatorStatus.RUNNING,
+ CheckpointCoordinatorStatus.CANCELED,
+ CheckpointCoordinatorStatus.FAILED,
+ CheckpointCoordinatorStatus.FINISHED,
+ CheckpointCoordinatorStatus.SUSPEND
+ };
+ for (CheckpointType checkpointType : CheckpointType.values()) {
+ for (boolean restored : new boolean[] {false, true}) {
+ CompletedCheckpoint completedCheckpoint =
+ new CompletedCheckpoint(
+ 1L,
+ 1,
+ 1L,
+ System.currentTimeMillis(),
+ checkpointType,
+ System.currentTimeMillis(),
+ new HashMap<>(),
+ new HashMap<>());
+ completedCheckpoint.setRestored(restored);
+ ReflectionUtils.setField(
+ coordinator, "latestCompletedCheckpoint",
completedCheckpoint);
+
+ for (CheckpointCoordinatorStatus status : states) {
+
Mockito.when(runningJobStateIMap.get(Mockito.any())).thenReturn(status);
+ boolean expected =
+ checkpointType.isFinalCheckpoint()
+ && (status ==
CheckpointCoordinatorStatus.FINISHED
+ || status ==
CheckpointCoordinatorStatus.SUSPEND)
+ && !restored;
+ Assertions.assertEquals(
+ expected,
+ coordinator.isNoErrorCompleted(),
+ String.format(
+ "checkpointType=%s, status=%s,
restored=%s",
+ checkpointType, status, restored));
+ }
+ }
+ }
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
/**
* Regression: when {@code notifyCompleted()} fails (returns {@code
false}), {@code
* completePendingCheckpoint} must return immediately without decrementing
{@code
