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-12377-5666e8d594ef09560a5d9e65ace4cc3748bdac2a in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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; + } }
