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 50256f12c3 [Fix][Zeta] Preserve cancellation after worker failure
(#12377)
50256f12c3 is described below
commit 50256f12c3092203f05ea98c30aea83a55b4266f
Author: Jast <[email protected]>
AuthorDate: Mon Oct 26 04:56:29 2026 +0000
[Fix][Zeta] Preserve cancellation after worker failure (#12377)
Co-authored-by: zhangshenghang <[email protected]>
Co-authored-by: jast <[email protected]>
Co-authored-by: multica-agent <[email protected]>
Co-authored-by: Claude Code <[email protected]>
---
.../common/junit/ContainerTestingExtension.java | 17 ++
.../junit/ContainerTestingExtensionTest.java | 100 ++++++++++
.../engine/server/dag/physical/PhysicalPlan.java | 7 +
.../engine/server/dag/physical/SubPlan.java | 102 ++++++++--
.../dag/physical/StateTransitionCleanupTest.java | 207 +++++++++++++++++++++
5 files changed, 417 insertions(+), 16 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtension.java
b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtension.java
index 0dac40cdcc..b8e1d7c9f1 100644
---
a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtension.java
+++
b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtension.java
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.extension.AfterAllCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.platform.commons.support.AnnotationSupport;
+import org.opentest4j.TestAbortedException;
import java.lang.annotation.Annotation;
import java.util.ArrayList;
@@ -73,6 +74,22 @@ public class ContainerTestingExtension implements
BeforeAllCallback, AfterAllCal
AnnotationUtil.filterDisabledContainers(
containersFactories.get(0).create(),
context.getRequiredTestInstance().getClass());
+ if (testContainers.isEmpty()) {
+ // The engine selection of a PR build (RUN_ENGINES,
RUN_ALL_CONTAINER and
+ // RUN_ZETA_CONTAINER) can exclude every container this suite is
able to run, e.g. a
+ // Flink/Spark-only suite on a Zeta-only run. Skip the whole suite
instead of letting
+ // every @TestTemplate fail with JUnit's "provided a non-empty
stream"
+ // PreconditionViolationException.
+ throw new TestAbortedException(
+ String.format(
+ "Skip %s: no enabled TestContainer under the
current engine selection"
+ + " (RUN_ENGINES=%s, RUN_ALL_CONTAINER=%s,
RUN_ZETA_CONTAINER"
+ + "=%s).",
+ context.getRequiredTestClass().getName(),
+ System.getenv("RUN_ENGINES"),
+ System.getenv("RUN_ALL_CONTAINER"),
+ System.getenv("RUN_ZETA_CONTAINER")));
+ }
Set<TestContainerId> sharedContainerIds =
getSharedContainerIds(context);
List<SharedTestContainerResource> sharedResources = new ArrayList<>();
try {
diff --git
a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtensionTest.java
b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtensionTest.java
new file mode 100644
index 0000000000..9c383bd3b4
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtensionTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.common.junit;
+
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.container.TestContainerId;
+import org.apache.seatunnel.e2e.common.container.TestContainersFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.opentest4j.TestAbortedException;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ContainerTestingExtensionTest {
+
+ private final ContainerTestingExtension extension = new
ContainerTestingExtension();
+
+ @Test
+ void shouldAbortSuiteWhenNoContainerIsDiscovered() {
+ assertThrows(
+ TestAbortedException.class,
+ () -> extension.beforeAll(contextFor(new EmptyFactorySuite())),
+ "A suite without any discovered container should be skipped");
+ }
+
+ @Test
+ void shouldAbortSuiteWhenEveryContainerIsDisabledOnTheSuite() {
+ assertThrows(
+ TestAbortedException.class,
+ () -> extension.beforeAll(contextFor(new ZetaDisabledSuite())),
+ "A suite whose containers are all disabled by
@DisabledOnContainer should be"
+ + " skipped");
+ }
+
+ @Test
+ void shouldRunSuiteWhenAtLeastOneContainerIsEnabled() {
+ assertDoesNotThrow(() -> extension.beforeAll(contextFor(new
ZetaSuite())));
+ }
+
+ private ExtensionContext contextFor(Object testInstance) {
+ ExtensionContext context = mock(ExtensionContext.class);
+ when(context.getRequiredTestInstance()).thenReturn(testInstance);
+ doReturn(testInstance.getClass()).when(context).getRequiredTestClass();
+ when(context.getStore(any(ExtensionContext.Namespace.class)))
+ .thenReturn(mock(ExtensionContext.Store.class));
+ return context;
+ }
+
+ private static TestContainer container(TestContainerId identifier) {
+ TestContainer container = mock(TestContainer.class);
+ when(container.identifier()).thenReturn(identifier);
+ return container;
+ }
+
+ /** Simulates a suite left without containers by the PR engine selection.
*/
+ private static final class EmptyFactorySuite {
+ @TestContainers
+ private final TestContainersFactory containersFactory =
Collections::emptyList;
+ }
+
+ /** Simulates a Flink/Spark-only suite on a Zeta-only PR run. */
+ @DisabledOnContainer(
+ value = {},
+ type = EngineType.SEATUNNEL)
+ private static final class ZetaDisabledSuite {
+ @TestContainers
+ private final TestContainersFactory containersFactory =
+ () ->
Collections.singletonList(container(TestContainerId.SEATUNNEL));
+ }
+
+ private static final class ZetaSuite {
+ @TestContainers
+ private final TestContainersFactory containersFactory =
+ () ->
Collections.singletonList(container(TestContainerId.SEATUNNEL));
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
index e4d4f7f162..ffa605efd3 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
@@ -405,6 +405,13 @@ public class PhysicalPlan {
case FAILING:
case CANCELING:
jobMaster.neverNeedRestore();
+ if (JobStatus.CANCELING.equals(jobStatus)) {
+ // Mark every pipeline as cancel-requested before any
cancelPipeline() can
+ // block on checkpoint cancellation:
CoordinatorService.makeTasksFailed()
+ // may otherwise complete the remaining RUNNING vertices
as FAILED while
+ // the cancel loop has not reached their pipeline yet
(#12353).
+ getPipelineList().forEach(SubPlan::markCancelRequested);
+ }
getPipelineList().forEach(SubPlan::cancelPipeline);
break;
case FAILED:
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
index fa4c2e3af0..3347ea4c33 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
@@ -17,6 +17,8 @@
package org.apache.seatunnel.engine.server.dag.physical;
+import
org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+
import org.apache.seatunnel.api.common.multitable.MultiTableFailureHelper;
import org.apache.seatunnel.api.options.EnvCommonOptions;
import org.apache.seatunnel.common.utils.ExceptionUtils;
@@ -105,6 +107,16 @@ public class SubPlan {
private volatile PipelineStatus currPipelineStatus;
+ /**
+ * Set as soon as cancellation of this pipeline is requested, before any
blocking work in {@link
+ * #cancelPipeline()}. While the pipeline state has not left {@code
RUNNING} yet (the cancel
+ * loop has not reached this pipeline, or {@code cancelPipeline()} is
still cancelling the
+ * checkpoint coordinator), {@code CoordinatorService.makeTasksFailed()}
may complete the
+ * remaining vertices as FAILED; this flag lets the failed-task callback
and the end-state
+ * computation honor the in-flight cancel so the pipeline resolves to
CANCELED (#12353).
+ */
+ private volatile boolean cancelRequested = false;
+
public volatile boolean isRunning = false;
private Map<TaskGroupLocation, SlotProfile> slotProfiles;
@@ -216,7 +228,17 @@ public class SubPlan {
failedTaskNum.incrementAndGet();
errorByPhysicalVertex.compareAndSet(
null, executionState.getThrowableMsg());
- updatePipelineState(PipelineStatus.FAILING);
+ // A worker failure must not overwrite a
cancellation that has
+ // already started: keep CANCELING until the
end-state
+ // computation so the pipeline resolves to
CANCELED instead of
+ // FAILED (#12353). The request may still be in
flight while the
+ // pipeline state has not left RUNNING (e.g.
cancelPipeline() is
+ // still cancelling the checkpoint coordinator),
hence the
+ // cancelRequested check.
+ if (!cancelRequested
+ &&
!PipelineStatus.CANCELING.equals(getPipelineState())) {
+ updatePipelineState(PipelineStatus.FAILING);
+ }
}
if (finishedTaskNum.incrementAndGet()
@@ -241,14 +263,12 @@ public class SubPlan {
}
private PipelineStatus getPipelineEndState() {
- PipelineStatus pipelineStatus = null;
- if (failedTaskNum.get() > 0) {
- pipelineStatus = PipelineStatus.FAILED;
+ PipelineStatus pipelineStatus = determinePipelineEndState();
+ if (PipelineStatus.FAILED.equals(pipelineStatus)) {
// we don't care the checkpoint error reason when the task is
// failed.
jobMaster.getCheckpointManager().cancelCheckpoint(getPipelineId()).join();
- } else if (canceledTaskNum.get() > 0) {
- pipelineStatus = PipelineStatus.CANCELED;
+ } else if (PipelineStatus.CANCELED.equals(pipelineStatus)) {
CheckpointCoordinatorState checkpointCoordinatorState =
jobMaster.getCheckpointManager().cancelCheckpoint(getPipelineId()).join();
if (CheckpointCoordinatorStatus.FAILED.equals(
@@ -257,17 +277,7 @@ public class SubPlan {
errorByPhysicalVertex.compareAndSet(
null, checkpointCoordinatorState.getThrowableMsg());
}
-
- // Because the pipeline state must update by tasks, If the
pipeline can not get enough
- // slot, the pipeline state will turn to Failing and then cancel
all tasks in this
- // pipeline.
- // Because the tasks never run, so the tasks will complete with
CANCELED. But the actual
- // status of the pipeline should be FAILED
- if (getPipelineState().equals(PipelineStatus.FAILING)) {
- pipelineStatus = PipelineStatus.FAILED;
- }
} else {
- pipelineStatus = PipelineStatus.FINISHED;
CheckpointCoordinatorState checkpointCoordinatorState =
jobMaster
.getCheckpointManager()
@@ -289,6 +299,49 @@ public class SubPlan {
return pipelineStatus;
}
+ /**
+ * Decides the terminal {@link PipelineStatus} of this pipeline from the
task counters and the
+ * current pipeline state.
+ *
+ * <p>Precedence: once cancellation has started and a task has actually
failed or been canceled,
+ * {@code CANCELING} wins over the failure and the pipeline resolves to
{@code CANCELED} — the
+ * cancellation requested by the user must not be overwritten by a worker
failure racing it
+ * (#12353). Cancellation counts as started as soon as it has been
requested ({@code
+ * cancelRequested}), even while the pipeline state has not left {@code
RUNNING} yet because the
+ * cancel loop has not reached this pipeline or {@code cancelPipeline()}
is still cancelling the
+ * checkpoint coordinator. A pipeline whose tasks all completed naturally
still resolves to
+ * {@code FINISHED} even if the pipeline state has not left {@code
CANCELING} yet, so a finished
+ * pipeline is never reclassified as {@code CANCELED} (which would make it
restore-eligible).
+ * Otherwise {@code failedTaskNum} wins over {@code canceledTaskNum}, with
the {@code FAILING}
+ * carve-out (e.g. slot allocation failure before any task ran) still
mapping canceled tasks to
+ * {@code FAILED}.
+ *
+ * <p>Runs on the vertex-callback executor, outside the {@code
synchronized} monitor of this
+ * SubPlan, reading the pipeline state and the task counters without
holding it.
+ */
+ @VisibleForTesting
+ PipelineStatus determinePipelineEndState() {
+ if ((failedTaskNum.get() > 0 || canceledTaskNum.get() > 0)
+ && (cancelRequested ||
PipelineStatus.CANCELING.equals(getPipelineState()))) {
+ // Once cancellation has started, a worker failure must not
overwrite the
+ // cancellation requested by the user.
+ return PipelineStatus.CANCELED;
+ }
+ if (failedTaskNum.get() > 0) {
+ return PipelineStatus.FAILED;
+ }
+ if (canceledTaskNum.get() > 0) {
+ // Because the pipeline state must update by tasks, if the
pipeline cannot get enough
+ // slots, the pipeline state turns to FAILING and then cancels all
tasks in this
+ // pipeline. Because the tasks never run, they complete with
CANCELED, but the actual
+ // status of the pipeline should be FAILED.
+ return PipelineStatus.FAILING.equals(getPipelineState())
+ ? PipelineStatus.FAILED
+ : PipelineStatus.CANCELED;
+ }
+ return PipelineStatus.FINISHED;
+ }
+
private boolean checkNeedRestore(PipelineStatus pipelineStatus) {
return canRestorePipeline()
&& !PipelineStatus.FINISHED.equals(pipelineStatus)
@@ -403,7 +456,21 @@ public class SubPlan {
}
}
+ /**
+ * Marks this pipeline as cancel-requested before any vertex can be failed
by member removal.
+ * Must be called before {@code cancelPipeline()} starts any blocking work
(checkpoint
+ * cancellation), so a concurrent {@code
CoordinatorService.makeTasksFailed()} cannot end the
+ * pipeline as FAILED while the cancel is still in flight (#12353).
+ */
+ public void markCancelRequested() {
+ this.cancelRequested = true;
+ }
+
public synchronized void cancelPipeline() {
+ // Record the request before cancelCheckpointCoordinator() can block,
so the
+ // failed-task callback and the end-state computation honor this
cancel even though
+ // the pipeline state has not left RUNNING yet (#12353).
+ markCancelRequested();
cancelCheckpointCoordinator();
if (!getPipelineState().isEndState()) {
updatePipelineState(PipelineStatus.CANCELING);
@@ -428,6 +495,9 @@ public class SubPlan {
finishedTaskNum.set(0);
canceledTaskNum.set(0);
failedTaskNum.set(0);
+ // The restored run starts from a clean slate: a cancel request from
the previous
+ // run must not leak into it (#12353).
+ cancelRequested = false;
coordinatorVertexList.forEach(PhysicalVertex::reset);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
index b0bbd7204a..4a5a115158 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
@@ -24,6 +24,7 @@ import org.apache.seatunnel.engine.common.config.EngineConfig;
import org.apache.seatunnel.engine.common.config.JobConfig;
import org.apache.seatunnel.engine.common.job.JobStatus;
import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture;
+import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import org.apache.seatunnel.engine.core.dag.logical.LogicalDag;
import org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.JobInfo;
@@ -31,6 +32,8 @@ import org.apache.seatunnel.engine.core.job.PipelineStatus;
import org.apache.seatunnel.engine.server.AbstractSeaTunnelServerTest;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.TestUtils;
+import
org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinatorState;
+import
org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinatorStatus;
import org.apache.seatunnel.engine.server.checkpoint.CheckpointManager;
import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
@@ -49,20 +52,28 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junitpioneer.jupiter.SetEnvironmentVariable;
import org.mockito.InOrder;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
import com.hazelcast.cluster.Address;
import com.hazelcast.map.IMap;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import static
org.apache.seatunnel.engine.common.config.server.QueueType.BLOCKINGQUEUE;
import static
org.apache.seatunnel.engine.core.classloader.DefaultClassLoaderService.SKIP_CHECK_JAR;
+import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
@@ -407,4 +418,200 @@ class StateTransitionCleanupTest extends
AbstractSeaTunnelServerTest {
this.runningJobStateTimestamps = runningJobStateTimestamps;
}
}
+
+ @Test
+ void testWorkerFailureDuringCancelKeepsPipelineCanceling() throws
Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ PlanWithStateMaps planWithStateMaps = createPhysicalPlan(jobId);
+
+ SubPlan subPlan =
planWithStateMaps.physicalPlan.getPipelineList().get(0);
+ subPlan.initStateFuture();
+ subPlan.updatePipelineState(PipelineStatus.CANCELING);
+
+ // A vertex the cancel loop has not reached yet (still CREATED) is
reported FAILED
+ // by CoordinatorService.makeTasksFailed() once the worker is lost.
This drives the
+ // real callback path in SubPlan.addPhysicalVertexCallBack().
+ PhysicalVertex workerLostVertex =
subPlan.getPhysicalVertexList().get(0);
+ workerLostVertex.startPhysicalVertex();
+ workerLostVertex.updateStateByExecutionService(
+ new TaskExecutionState(
+ workerLostVertex.getTaskGroupLocation(),
+ ExecutionState.FAILED,
+ "worker lost"));
+
+ await().atMost(Duration.ofSeconds(10)).until(() ->
subPlan.getFinishedTaskNum().get() == 1);
+ // Moving the pipeline to FAILING here is exactly the race that made a
+ // user-requested cancel end up FAILED (#12353).
+ Assertions.assertEquals(PipelineStatus.CANCELING,
subPlan.getPipelineState());
+ }
+
+ @Test
+ void testWorkerFailureDuringCancelResolvesToCanceled() throws Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ PlanWithStateMaps planWithStateMaps = createPhysicalPlan(jobId);
+
+ SubPlan subPlan =
planWithStateMaps.physicalPlan.getPipelineList().get(0);
+ subPlan.setJobMaster(mockJobMaster());
+
+ subPlan.initStateFuture();
+ subPlan.updatePipelineState(PipelineStatus.CANCELING);
+
+ // One task fails with the lost worker; every other task (including the
+ // coordinator vertices) is canceled.
+ List<PhysicalVertex> allVertices = new
ArrayList<>(subPlan.getPhysicalVertexList());
+ allVertices.addAll(subPlan.getCoordinatorVertexList());
+ for (PhysicalVertex vertex : allVertices) {
+ vertex.startPhysicalVertex();
+ }
+ allVertices
+ .get(0)
+ .updateStateByExecutionService(
+ new TaskExecutionState(
+ allVertices.get(0).getTaskGroupLocation(),
+ ExecutionState.FAILED,
+ "worker lost"));
+ for (int i = 1; i < allVertices.size(); i++) {
+ allVertices
+ .get(i)
+ .updateStateByExecutionService(
+ new TaskExecutionState(
+ allVertices.get(i).getTaskGroupLocation(),
+ ExecutionState.CANCELED));
+ }
+
+ await().atMost(Duration.ofSeconds(10))
+ .until(() ->
PipelineStatus.CANCELED.equals(subPlan.getPipelineState()));
+ }
+
+ @Test
+ void testAllTasksFinishedNaturallyDuringCancelResolvesToFinished() throws
Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ PlanWithStateMaps planWithStateMaps = createPhysicalPlan(jobId);
+
+ SubPlan subPlan =
planWithStateMaps.physicalPlan.getPipelineList().get(0);
+ subPlan.setJobMaster(mockJobMaster());
+
+ subPlan.initStateFuture();
+ subPlan.updatePipelineState(PipelineStatus.CANCELING);
+
+ // Every task (including the coordinator vertices) completes FINISHED
while the
+ // pipeline state has not left CANCELING yet: the pipeline must still
end FINISHED,
+ // not be reclassified as CANCELED.
+ List<PhysicalVertex> allVertices = new
ArrayList<>(subPlan.getPhysicalVertexList());
+ allVertices.addAll(subPlan.getCoordinatorVertexList());
+ for (PhysicalVertex vertex : allVertices) {
+ vertex.startPhysicalVertex();
+ vertex.updateStateByExecutionService(
+ new TaskExecutionState(vertex.getTaskGroupLocation(),
ExecutionState.FINISHED));
+ }
+
+ await().atMost(Duration.ofSeconds(10))
+ .until(() ->
PipelineStatus.FINISHED.equals(subPlan.getPipelineState()));
+ }
+
+ /**
+ * Sibling-vertex ordering (#12353): the job-level cancel loop processes
pipelines one by one
+ * and is still cancelling the checkpoint coordinator of the first
pipeline when the worker
+ * hosting every vertex of a later pipeline is removed. {@code
+ * CoordinatorService.makeTasksFailed()} completes the vertices of that
pipeline — including its
+ * coordinator vertex, also deployed on the crashed member — as {@code
FAILED} before the cancel
+ * loop has ever reached the pipeline, so the pipeline state has not
become {@code CANCELING}
+ * yet. The pipeline must still end {@code CANCELED}: the user-requested
cancel must not be
+ * overwritten to {@code FAILED}.
+ */
+ @Test
+ void testSiblingWorkerFailureWhileCancelInFlightResolvesToCanceled()
throws Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ PlanWithStateMaps planWithStateMaps = createPhysicalPlan(jobId);
+ PhysicalPlan physicalPlan = planWithStateMaps.physicalPlan;
+
+ // A second pipeline the job-level cancel loop has not reached while
it is still stuck
+ // cancelling the first pipeline's checkpoint coordinator.
+ SubPlan notYetReachedPipeline =
+ createPhysicalPlan(
+
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME)
+ .newId())
+ .physicalPlan
+ .getPipelineList()
+ .get(0);
+ physicalPlan.getPipelineList().add(notYetReachedPipeline);
+
+ // The first pipeline's checkpoint cancellation blocks the job-level
cancel loop.
+ CompletableFuture<CheckpointCoordinatorState>
firstPipelineCancelFuture =
+ new CompletableFuture<>();
+ CountDownLatch cancelLoopBlockedOnFirstPipeline = new
CountDownLatch(1);
+ physicalPlan.setJobMaster(
+ mockJobMaster(
+ invocation -> {
+ cancelLoopBlockedOnFirstPipeline.countDown();
+ return new
PassiveCompletableFuture<>(firstPipelineCancelFuture);
+ }));
+ // The not-yet-reached pipeline must not block while computing its end
state.
+ notYetReachedPipeline.setJobMaster(mockJobMaster());
+
+ physicalPlan.initStateFuture();
+ // Move every pipeline to RUNNING before starting the job state
machine, so the
+ // SCHEDULED branch of PhysicalPlan.stateProcess() skips them (no
resource
+ // application) and cancelJob() walks the CANCELING branch.
+ for (SubPlan subPlan : physicalPlan.getPipelineList()) {
+ subPlan.updatePipelineState(PipelineStatus.RUNNING);
+ }
+ physicalPlan.startJob();
+
+ Thread cancelJobThread = new Thread(physicalPlan::cancelJob);
+ cancelJobThread.start();
+ Assertions.assertTrue(
+ cancelLoopBlockedOnFirstPipeline.await(10, TimeUnit.SECONDS),
+ "the job-level cancel loop must reach the first pipeline");
+
+ // The crashed worker hosted every vertex of the not-yet-reached
pipeline (task
+ // vertices and coordinator vertex, all still RUNNING when the member
is removed).
+ List<PhysicalVertex> crashedVertices =
+ new ArrayList<>(notYetReachedPipeline.getPhysicalVertexList());
+
crashedVertices.addAll(notYetReachedPipeline.getCoordinatorVertexList());
+ Assertions.assertTrue(
+ crashedVertices.size() >= 2,
+ "the pipeline must host at least two vertices on the crashed
member");
+ for (PhysicalVertex vertex : crashedVertices) {
+ vertex.startPhysicalVertex();
+ vertex.updateStateByExecutionService(
+ new TaskExecutionState(
+ vertex.getTaskGroupLocation(),
ExecutionState.FAILED, "worker lost"));
+ }
+
+ await().atMost(Duration.ofSeconds(10))
+ .until(() ->
notYetReachedPipeline.getPipelineState().isEndState());
+ Assertions.assertEquals(PipelineStatus.CANCELED,
notYetReachedPipeline.getPipelineState());
+
+ // Let the blocked job-level cancel loop finish.
+ firstPipelineCancelFuture.complete(
+ new
CheckpointCoordinatorState(CheckpointCoordinatorStatus.FINISHED, null));
+ cancelJobThread.join(5000);
+ }
+
+ private JobMaster mockJobMaster() {
+ return mockJobMaster(
+ invocation ->
+ new PassiveCompletableFuture<>(
+ CompletableFuture.completedFuture(
+ new CheckpointCoordinatorState(
+
CheckpointCoordinatorStatus.FINISHED, null))));
+ }
+
+ private JobMaster mockJobMaster(
+ Answer<PassiveCompletableFuture<CheckpointCoordinatorState>>
cancelCheckpointAnswer) {
+ CheckpointManager checkpointManager =
Mockito.mock(CheckpointManager.class);
+ Mockito.when(checkpointManager.cancelCheckpoint(Mockito.anyInt()))
+ .thenAnswer(cancelCheckpointAnswer);
+
Mockito.when(checkpointManager.waitCheckpointCoordinatorComplete(Mockito.anyInt()))
+ .thenReturn(
+ new PassiveCompletableFuture<>(
+ CompletableFuture.completedFuture(
+ new CheckpointCoordinatorState(
+
CheckpointCoordinatorStatus.FINISHED, null))));
+ JobMaster jobMaster = Mockito.mock(JobMaster.class);
+
Mockito.when(jobMaster.getCheckpointManager()).thenReturn(checkpointManager);
+
Mockito.when(jobMaster.getExecutorService()).thenReturn(Executors.newCachedThreadPool());
+ return jobMaster;
+ }
}
