This is an automated email from the ASF dual-hosted git repository.
corgy-w 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 4aa843afc3 [Fix][Zeta] Stabilize coordinator restore test (#10877)
4aa843afc3 is described below
commit 4aa843afc33feb7c2bfed791526aaa51713f079c
Author: Jast <[email protected]>
AuthorDate: Wed May 13 09:56:42 2026 +0800
[Fix][Zeta] Stabilize coordinator restore test (#10877)
---
.../server/CoordinatorServiceJobCleanupTest.java | 21 ++++++++++++++++++++-
.../engine/server/CoordinatorServiceTest.java | 12 ++++++++++--
2 files changed, 30 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceJobCleanupTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceJobCleanupTest.java
index 5797ea6300..90400a6a82 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceJobCleanupTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceJobCleanupTest.java
@@ -31,6 +31,7 @@ import
org.apache.seatunnel.engine.serializer.protobuf.ProtoStuffSerializer;
import org.apache.seatunnel.engine.server.checkpoint.CompletedCheckpoint;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+import org.apache.seatunnel.engine.server.master.JobMaster;
import org.apache.seatunnel.engine.server.master.cleanup.JobCleanupRecord;
import org.apache.seatunnel.engine.server.operation.SubmitJobOperation;
import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;
@@ -46,10 +47,12 @@ import com.hazelcast.internal.serialization.Data;
import com.hazelcast.map.IMap;
import com.hazelcast.spi.exception.RetryableHazelcastException;
+import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -493,6 +496,8 @@ class CoordinatorServiceJobCleanupTest extends
AbstractSeaTunnelServerTest {
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_INFO);
IMap<Object, Object> runningJobStateIMap =
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_STATE);
+ IMap<Object, Long[]> runningJobStateTimestampsIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_STATE_TIMESTAMPS);
JobInfo jobInfo =
new JobInfo(
@@ -513,7 +518,21 @@ class CoordinatorServiceJobCleanupTest extends
AbstractSeaTunnelServerTest {
ReflectionUtils.setField(coordinatorService, "runningJobInfoIMap",
runningJobInfoIMap);
}
-
Assertions.assertTrue(coordinatorService.getPendingJobQueue().contains(jobId));
+ Long[] jobStateTimestamps = runningJobStateTimestampsIMap.get(jobId);
+ Assertions.assertNotNull(jobStateTimestamps);
+ Assertions.assertEquals(
+ initializationTimestamp,
jobStateTimestamps[JobStatus.INITIALIZING.ordinal()]);
+ Assertions.assertTrue(
+ coordinatorService.getPendingJobQueue().contains(jobId)
+ ||
getRunningJobMasterMap(coordinatorService).containsKey(jobId));
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, JobMaster> getRunningJobMasterMap(CoordinatorService
coordinatorService)
+ throws Exception {
+ Field field =
CoordinatorService.class.getDeclaredField("runningJobMasterMap");
+ field.setAccessible(true);
+ return (Map<Long, JobMaster>) field.get(coordinatorService);
}
private Set<Object> stateKeys(Object... keys) {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
index 2767c84ae2..195e0b878a 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
@@ -852,7 +852,7 @@ public class CoordinatorServiceTest {
createHazelcastInstanceWithJoinPortTryCount(
TestUtils.getClusterName(
"CoordinatorServiceTest_testRestoreUsesProvidedJobInfoInitializationTimestamp"),
- 100);
+ 1);
try {
SeaTunnelServer server =
instance.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
@@ -885,6 +885,8 @@ public class CoordinatorServiceTest {
instance.getMap(Constant.IMAP_RUNNING_JOB_INFO);
IMap<Object, Object> runningJobStateIMap =
instance.getMap(Constant.IMAP_RUNNING_JOB_STATE);
+ IMap<Object, Long[]> runningJobStateTimestampsIMap =
+ instance.getMap(Constant.IMAP_STATE_TIMESTAMPS);
runningJobInfoIMap.put(jobId, jobInfo);
runningJobStateIMap.put(jobId, JobStatus.RUNNING);
@@ -901,7 +903,13 @@ public class CoordinatorServiceTest {
coordinatorService, "runningJobInfoIMap",
runningJobInfoIMap);
}
-
Assertions.assertTrue(coordinatorService.getPendingJobQueue().contains(jobId));
+ Long[] jobStateTimestamps =
runningJobStateTimestampsIMap.get(jobId);
+ Assertions.assertNotNull(jobStateTimestamps);
+ Assertions.assertEquals(
+ initializationTimestamp,
jobStateTimestamps[JobStatus.INITIALIZING.ordinal()]);
+ Assertions.assertTrue(
+ coordinatorService.getPendingJobQueue().contains(jobId)
+ ||
getRunningJobMasterMap(coordinatorService).containsKey(jobId));
} finally {
instance.shutdown();
}
