This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 20900262ee [Test][E2E] Add regression coverage for task stuck in
CANCELING when worker crashes mid-cancel (#12030)
20900262ee is described below
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];
