This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 3b6d43bcc0 [Test][E2E] Wait for failover task graph before cancel
(#11236)
3b6d43bcc0 is described below
commit 3b6d43bcc06df6290bd5caacf4233e155aef2ccd
Author: Daniel <[email protected]>
AuthorDate: Wed Jul 1 20:46:30 2026 +0800
[Test][E2E] Wait for failover task graph before cancel (#11236)
---
.../SplitClusterPendingJobLifecycleFailoverIT.java | 67 ++++++++++++++++++++++
1 file changed, 67 insertions(+)
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
index 17033d47b4..e44f94dd13 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
@@ -28,8 +28,14 @@ import
org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
import org.apache.seatunnel.engine.common.config.server.ScheduleStrategy;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.core.job.PipelineStatus;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
+import org.apache.seatunnel.engine.server.dag.physical.PhysicalPlan;
+import org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex;
+import org.apache.seatunnel.engine.server.dag.physical.SubPlan;
+import org.apache.seatunnel.engine.server.execution.ExecutionState;
+import org.apache.seatunnel.engine.server.master.JobMaster;
import org.awaitility.Awaitility;
import org.jetbrains.annotations.NotNull;
@@ -148,6 +154,7 @@ public class SplitClusterPendingJobLifecycleFailoverIT {
3,
standbyMaster.getCluster().getMembers().size()));
assertJobStatusWithTimeout(pendingJobAfterFailover,
JobStatus.RUNNING, 180);
assertPendingQueueNotContainsJob(standbyMaster, pendingJobId);
+ assertRunningJobGraphWithTimeout(standbyMaster, pendingJobId, 120);
pendingJobAfterFailover.cancelJob();
assertJobStatusWithTimeout(pendingJobAfterFailover,
JobStatus.CANCELED, 120);
@@ -289,6 +296,66 @@ public class SplitClusterPendingJobLifecycleFailoverIT {
expectedStatus,
clientJobProxy.getJobStatus()));
}
+ /**
+ * Waits until failover restore has rebuilt the running graph on the
active master.
+ *
+ * <p>The top-level job status can turn RUNNING before every restored
pipeline and task vertex
+ * has reacquired slots and reported RUNNING. Cancelling earlier makes
this test depend on a
+ * restore/cancel race instead of pending-job scheduling.
+ */
+ private static void assertRunningJobGraphWithTimeout(
+ HazelcastInstanceImpl activeMaster, long jobId, long
timeoutSeconds) {
+ Awaitility.await()
+ .atMost(timeoutSeconds, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ JobMaster jobMaster = getJobMaster(activeMaster,
jobId);
+ Assertions.assertNotNull(
+ jobMaster,
+ "Job master should exist before checking
restored task states");
+ PhysicalPlan physicalPlan =
jobMaster.getPhysicalPlan();
+ Assertions.assertEquals(JobStatus.RUNNING,
physicalPlan.getJobStatus());
+ physicalPlan
+ .getPipelineList()
+ .forEach(
+
SplitClusterPendingJobLifecycleFailoverIT
+ ::assertRunningSubPlan);
+ });
+ }
+
+ /**
+ * Asserts that one restored pipeline and all of its task vertices are
fully running.
+ *
+ * <p>This keeps the following cancel assertion focused on the lifecycle
transition instead of
+ * racing against delayed task deployment after master failover.
+ */
+ private static void assertRunningSubPlan(SubPlan subPlan) {
+ Assertions.assertEquals(PipelineStatus.RUNNING,
subPlan.getPipelineState());
+ subPlan.getCoordinatorVertexList()
+
.forEach(SplitClusterPendingJobLifecycleFailoverIT::assertRunningVertex);
+ subPlan.getPhysicalVertexList()
+
.forEach(SplitClusterPendingJobLifecycleFailoverIT::assertRunningVertex);
+ }
+
+ /**
+ * Asserts that one task vertex has completed deployment and reported
RUNNING.
+ *
+ * <p>A restored vertex can lag behind the top-level job status while
resources are assigned, so
+ * the test must observe the vertex state directly before cancelling.
+ */
+ private static void assertRunningVertex(PhysicalVertex physicalVertex) {
+ Assertions.assertEquals(ExecutionState.RUNNING,
physicalVertex.getExecutionState());
+ }
+
+ /**
+ * Reads the current job master from the active SeaTunnel server embedded
in the test cluster.
+ */
+ private static JobMaster getJobMaster(HazelcastInstanceImpl activeMaster,
long jobId) {
+ SeaTunnelServer server =
+
activeMaster.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ return server.getCoordinatorService().getJobMaster(jobId);
+ }
+
private static HazelcastInstanceImpl waitAndFindActiveMaster(
HazelcastInstanceImpl masterNode1, HazelcastInstanceImpl
masterNode2) {
final HazelcastInstanceImpl[] activeMasterRef = new
HazelcastInstanceImpl[1];
