This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12030-3a7c1226a5370aabcc1f851893acbc783f28f580 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 20900262eeea837b1321138a9bccb0437abefc54 Author: David Zollo <[email protected]> AuthorDate: Wed Sep 9 05:06:11 2026 +0000 [Test][E2E] Add regression coverage for task stuck in CANCELING when worker crashes mid-cancel (#12030) --- .../engine/e2e/SplitClusterFaultToleranceIT.java | 246 ++++++++++++++++++++- 1 file changed, 237 insertions(+), 9 deletions(-) diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterFaultToleranceIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterFaultToleranceIT.java index 71d76ed713..665bb340a1 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterFaultToleranceIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterFaultToleranceIT.java @@ -29,7 +29,9 @@ import org.apache.seatunnel.engine.common.config.ConfigProvider; import org.apache.seatunnel.engine.common.config.JobConfig; import org.apache.seatunnel.engine.common.config.SeaTunnelConfig; 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; import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; import org.apache.seatunnel.engine.server.dag.physical.PhysicalPlan; @@ -54,6 +56,7 @@ import lombok.extern.slf4j.Slf4j; import java.io.File; import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; @@ -333,6 +336,240 @@ public class SplitClusterFaultToleranceIT { } } + /** + * Regression test for the CANCELING-stuck-forever bug fixed by <a + * href="https://github.com/apache/seatunnel/pull/10729";>#10729</a> ("[Fix][Zeta] prevent cancel + * stuck and downgrade tmp cleanup failure"). Before that fix, {@code + * PhysicalVertex#noticeTaskExecutionServiceCancel} sent a {@code CancelTaskOperation} to the + * worker owning the task and retried while that worker remained a cluster member, but once the + * worker actually left the cluster before acking the cancel, the retry loop simply exited + * without ever resolving the task's state: the vertex - and therefore its pipeline and job - + * stayed CANCELING forever, because nothing else in that method drove it to a terminal state. + * The fix added a fallback right after the retry loop: if the loop exits with the ack still + * missing and the vertex is still CANCELING, mark it CANCELED locally. + * + * <p>{@code CoordinatorService#failedTaskOnMemberRemoved} also matches CANCELING tasks when a + * member is lost, but it cannot be the one to save this race: {@code PhysicalVertex#cancel}, + * {@code #updateTaskState} and {@code #stateProcess} are all {@code synchronized} on the vertex + * itself, and the whole call chain down into {@code noticeTaskExecutionServiceCancel} runs + * without releasing that lock. A competing {@code failedTaskOnMemberRemoved} call has to go + * through the same {@code synchronized updateTaskState}, so it can only run after the + * cancelling thread has already returned - by which point the fixed code has already resolved + * the vertex to CANCELED, and the end-state consistency check inside {@code updateTaskState} + * rejects any later attempt to move a terminal-state task to FAILED. So the fixed method's own + * fallback is the sole, deterministic mechanism that unsticks this specific race, which is why + * this test asserts CANCELED rather than FAILED as the outcome. + * + * <p>No existing fault-tolerance test combines "cancel requested" with "worker crashes before + * the cancel ack arrives": {@link #testStreamJobRunOk()} cancels a job on a fully healthy + * cluster, and {@link #testStreamJobRestoreInWorkerDown()} kills a worker on a job that is + * simply RUNNING with no cancel in flight, then cancels only after restore has completed. This + * test constructs the narrow race directly: it white-box polls {@code + * PhysicalVertex#getExecutionState()} (same in-JVM technique as {@code + * SplitClusterPendingJobLifecycleFailoverIT#getJobMaster}) in a tight loop right after issuing + * the cancel, and shuts the worker down the instant a task vertex is first observed CANCELING - + * landing squarely inside the RPC-in-flight window instead of guessing at timing. + */ + @Test + public void testStreamJobCancelResolvesWhenWorkerCrashesBeforeCancelAck() throws Exception { + String testCaseName = "testStreamJobCancelResolvesWhenWorkerCrashesBeforeCancelAck"; + String testClusterName = "SplitClusterFaultToleranceIT_" + testCaseName; + HazelcastInstanceImpl masterNode = null; + HazelcastInstanceImpl workerNode = null; + SeaTunnelClient engineClient = null; + + SeaTunnelConfig seaTunnelConfig = getSeaTunnelConfig(testClusterName); + SeaTunnelConfig masterNodeConfig = getSeaTunnelConfig(testClusterName); + SeaTunnelConfig workerNodeConfig = getSeaTunnelConfig(testClusterName); + + try { + masterNode = SeaTunnelServerStarter.createMasterHazelcastInstance(masterNodeConfig); + + workerNode = SeaTunnelServerStarter.createWorkerHazelcastInstance(workerNodeConfig); + + // waiting all node added to cluster + HazelcastInstanceImpl finalMasterNode = masterNode; + Awaitility.await() + .atMost(10000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + 2, finalMasterNode.getCluster().getMembers().size())); + + Common.setDeployMode(DeployMode.CLIENT); + JobConfig jobConfig = new JobConfig(); + jobConfig.setName(testCaseName); + + ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig(); + clientConfig.setClusterName(TestUtils.getClusterName(testClusterName)); + engineClient = new SeaTunnelClient(clientConfig); + ClientJobExecutionEnvironment jobExecutionEnv = + engineClient.createExecutionContext( + TestUtils.getResource("pending_jobs_streaming_lifecycle.conf"), + jobConfig, + seaTunnelConfig); + ClientJobProxy clientJobProxy = jobExecutionEnv.execute(); + long jobId = clientJobProxy.getJobId(); + + Awaitility.await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + JobStatus.RUNNING, clientJobProxy.getJobStatus())); + // Wait for every task vertex to actually finish deploying before cancelling, so the + // CANCELING transition observed below is unambiguously caused by the cancel request, + // not mixed up with an in-flight DEPLOYING -> RUNNING transition. + assertAllVerticesRunning(masterNode, jobId, 60); + + // CoordinatorService#cancelJob runs JobMaster#cancelJob synchronously on its own + // executor, and the client's cancelJob() blocks (PassiveCompletableFuture#join) until + // that whole per-vertex cancel resolves. Issue it on its own thread so this thread + // stays free to tight-poll for CANCELING and react to it immediately. + CompletableFuture<Void> cancelInvocation = + CompletableFuture.runAsync(clientJobProxy::cancelJob); + + boolean observedCanceling = + awaitAnyVertexCanceling(masterNode, jobId, 10, TimeUnit.SECONDS); + Assertions.assertTrue( + observedCanceling, + "Should have observed a task vertex enter CANCELING before its cancel ack " + + "could possibly arrive; otherwise this test never exercised the " + + "race it targets"); + + // The vertex is CANCELING and still waiting on a CancelTaskOperation ack from + // workerNode. Kill workerNode right now, before that ack can arrive, landing inside + // the exact window PR #10729 fixed: cancel requested, CANCELING set, worker gone + // before the terminal callback. + workerNode.shutdown(); + + CompletableFuture<JobResult> waitForCompleteFuture = + CompletableFuture.supplyAsync(clientJobProxy::waitForJobCompleteV2); + assertEventuallyCanceled(clientJobProxy, waitForCompleteFuture); + + Assertions.assertDoesNotThrow( + () -> cancelInvocation.get(30, TimeUnit.SECONDS), + "cancelJob() should return once the vertex resolves locally, not hang"); + } finally { + if (engineClient != null) { + engineClient.close(); + } + + if (workerNode != null) { + workerNode.shutdown(); + } + + if (masterNode != null) { + masterNode.shutdown(); + } + } + } + + /** + * Waits until every coordinator and task vertex of the job's single pipeline has itself + * reported RUNNING, not just the top-level job status. + */ + private static void assertAllVerticesRunning( + HazelcastInstanceImpl masterNode, long jobId, long timeoutSeconds) { + Awaitility.await() + .atMost(timeoutSeconds, TimeUnit.SECONDS) + .untilAsserted( + () -> { + JobMaster jobMaster = getJobMaster(masterNode, jobId); + Assertions.assertNotNull( + jobMaster, + "Job master should exist before checking task vertex states"); + PhysicalPlan physicalPlan = jobMaster.getPhysicalPlan(); + Assertions.assertEquals(JobStatus.RUNNING, physicalPlan.getJobStatus()); + physicalPlan + .getPipelineList() + .forEach(SplitClusterFaultToleranceIT::assertRunningSubPlan); + }); + } + + private static void assertRunningSubPlan(SubPlan subPlan) { + Assertions.assertEquals(PipelineStatus.RUNNING, subPlan.getPipelineState()); + subPlan.getCoordinatorVertexList() + .forEach(SplitClusterFaultToleranceIT::assertRunningVertex); + subPlan.getPhysicalVertexList().forEach(SplitClusterFaultToleranceIT::assertRunningVertex); + } + + private static void assertRunningVertex(PhysicalVertex physicalVertex) { + Assertions.assertEquals(ExecutionState.RUNNING, physicalVertex.getExecutionState()); + } + + /** + * Tight-polls (no fixed sleep) every coordinator and task vertex of the job for CANCELING, + * returning as soon as any one is observed. The window between a task vertex entering CANCELING + * and its cancel ack arriving from a healthy worker can be sub-millisecond, well under + * Awaitility's default ~100ms poll interval, so this deliberately busy-spins instead. + */ + private static boolean awaitAnyVertexCanceling( + HazelcastInstanceImpl masterNode, long jobId, long timeout, TimeUnit unit) { + long deadline = System.nanoTime() + unit.toNanos(timeout); + while (System.nanoTime() < deadline) { + JobMaster jobMaster = getJobMaster(masterNode, jobId); + if (jobMaster != null) { + PhysicalPlan physicalPlan = jobMaster.getPhysicalPlan(); + if (physicalPlan != null) { + for (SubPlan subPlan : physicalPlan.getPipelineList()) { + if (isAnyVertexCanceling(subPlan.getCoordinatorVertexList()) + || isAnyVertexCanceling(subPlan.getPhysicalVertexList())) { + return true; + } + } + } + } + Thread.yield(); + } + return false; + } + + private static boolean isAnyVertexCanceling(List<PhysicalVertex> vertices) { + for (PhysicalVertex vertex : vertices) { + if (ExecutionState.CANCELING.equals(vertex.getExecutionState())) { + return true; + } + } + return false; + } + + /** + * Reads the current job master from the active SeaTunnel server embedded in the test cluster. + */ + private static JobMaster getJobMaster(HazelcastInstanceImpl masterNode, long jobId) { + SeaTunnelServer server = + masterNode.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME); + return server.getCoordinatorService().getJobMaster(jobId); + } + + /** + * Waits for the job to reach a terminal state and asserts it is CANCELED, using the + * already-in-flight {@code waitForJobCompleteV2()} future so this assertion cannot itself hang + * forever if the CANCELING-stuck regression this test guards against were to reappear. + */ + private static void assertEventuallyCanceled( + ClientJobProxy clientJobProxy, CompletableFuture<JobResult> waitForCompleteFuture) { + Awaitility.await() + .atMost(60, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> { + Assertions.assertTrue( + waitForCompleteFuture.isDone(), + "Job should reach a terminal state instead of staying stuck " + + "in CANCELING after its worker crashed mid-cancel"); + Assertions.assertEquals( + JobStatus.CANCELED, waitForCompleteFuture.get().getStatus()); + }); + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + JobStatus.CANCELED, clientJobProxy.getJobStatus())); + } + @Test public void testBatchJobRestoreInWorkerDown() throws Exception { String testCaseName = "testBatchJobRestoreInWorkerDown"; @@ -1369,15 +1606,6 @@ public class SplitClusterFaultToleranceIT { 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];
