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 741336bdb9 [Test][Zeta] Stabilize pending job cancel status after
failover (#11258)
741336bdb9 is described below
commit 741336bdb985d8ba788be01218b4399f840236e2
Author: Daniel <[email protected]>
AuthorDate: Thu Jul 2 21:00:09 2026 +0800
[Test][Zeta] Stabilize pending job cancel status after failover (#11258)
Co-authored-by: DanielLeens <[email protected]>
---
.../SplitClusterPendingJobLifecycleFailoverIT.java | 23 +++++++++++++++++++---
1 file changed, 20 insertions(+), 3 deletions(-)
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 e44f94dd13..b7f4285c4e 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
@@ -27,6 +27,7 @@ import org.apache.seatunnel.engine.common.config.JobConfig;
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.JobResult;
import org.apache.seatunnel.engine.common.job.JobStatus;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
@@ -157,7 +158,7 @@ public class SplitClusterPendingJobLifecycleFailoverIT {
assertRunningJobGraphWithTimeout(standbyMaster, pendingJobId, 120);
pendingJobAfterFailover.cancelJob();
- assertJobStatusWithTimeout(pendingJobAfterFailover,
JobStatus.CANCELED, 120);
+ assertEventuallyCanceled(pendingJobAfterFailover);
engineClient.createJobClient().getJobProxy(holderJob.getJobId()).cancelJob();
} finally {
if (engineClient != null) {
@@ -232,13 +233,13 @@ public class SplitClusterPendingJobLifecycleFailoverIT {
assertPendingQueueState(activeMaster, pendingJobId, 1);
holderJob.cancelJob();
- assertJobStatusWithTimeout(holderJob, JobStatus.CANCELED, 120);
+ assertEventuallyCanceled(holderJob);
assertJobStatusWithTimeout(pendingJob, JobStatus.RUNNING, 180);
assertPendingQueueNotContainsJob(activeMaster, pendingJobId);
pendingJob.cancelJob();
- assertJobStatusWithTimeout(pendingJob, JobStatus.CANCELED, 120);
+ assertEventuallyCanceled(pendingJob);
} finally {
if (engineClient != null) {
engineClient.close();
@@ -296,6 +297,22 @@ public class SplitClusterPendingJobLifecycleFailoverIT {
expectedStatus,
clientJobProxy.getJobStatus()));
}
+ /**
+ * Failover and cancellation can leave the client-observed status at
CANCELING after the cancel
+ * request has already completed, so wait for the terminal job result
before asserting the final
+ * visible state.
+ */
+ private static void assertEventuallyCanceled(ClientJobProxy
clientJobProxy) {
+ JobResult jobResult = clientJobProxy.waitForJobCompleteV2();
+ Assertions.assertEquals(JobStatus.CANCELED, jobResult.getStatus());
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ JobStatus.CANCELED,
clientJobProxy.getJobStatus()));
+ }
+
/**
* Waits until failover restore has rebuilt the running graph on the
active master.
*
