This is an automated email from the ASF dual-hosted git repository.
wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 5a93624ace [Fix-17354][Master] Killed state of timeout failure
strategy will cause workflow can't finish or stop (#17392)
5a93624ace is described below
commit 5a93624aceda7fde573908b41cf28c30f9f98acd
Author: Wenjun Ruan <[email protected]>
AuthorDate: Tue Aug 5 20:01:06 2025 +0800
[Fix-17354][Master] Killed state of timeout failure strategy will cause
workflow can't finish or stop (#17392)
---
.../engine/graph/IWorkflowExecutionGraph.java | 4 +-
.../engine/graph/WorkflowExecutionGraph.java | 8 +-
.../WorkflowReadyPauseStateAction.java | 8 +-
.../statemachine/WorkflowReadyStopStateAction.java | 2 +-
.../statemachine/WorkflowRunningStateAction.java | 20 ++++-
.../integration/WorkflowTestCaseContext.java | 10 +++
.../integration/cases/WorkflowStartTestCase.java | 31 +++++++
.../it/start/workflow_with_timeout_kill_task.yaml | 95 ++++++++++++++++++++++
8 files changed, 165 insertions(+), 13 deletions(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java
index d9266b721c..fdbbb581da 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/IWorkflowExecutionGraph.java
@@ -182,12 +182,12 @@ public interface IWorkflowExecutionGraph {
/**
* Whether there exist the TaskExecutionRunnable chain in the graph is
finish with paused.
*/
- boolean isExistPauseTaskExecutionRunnableChain();
+ boolean isExistPausedTaskExecutionRunnableChain();
/**
* Whether there exist the TaskExecutionRunnable chain in the graph is
finish with kill.
*/
- boolean isExistKillTaskExecutionRunnableChain();
+ boolean isExistKilledTaskExecutionRunnableChain();
/**
* Check whether the given task is the end of the task chain.
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java
index a9ce38cc7b..0fa4d15cac 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowExecutionGraph.java
@@ -204,8 +204,8 @@ public class WorkflowExecutionGraph implements
IWorkflowExecutionGraph {
return false;
}
return !isExistFailureTaskExecutionRunnableChain()
- && !isExistPauseTaskExecutionRunnableChain()
- && !isExistKillTaskExecutionRunnableChain();
+ && !isExistPausedTaskExecutionRunnableChain()
+ && !isExistKilledTaskExecutionRunnableChain();
}
@Override
@@ -214,12 +214,12 @@ public class WorkflowExecutionGraph implements
IWorkflowExecutionGraph {
}
@Override
- public boolean isExistPauseTaskExecutionRunnableChain() {
+ public boolean isExistPausedTaskExecutionRunnableChain() {
return CollectionUtils.isNotEmpty(pausedTaskChains);
}
@Override
- public boolean isExistKillTaskExecutionRunnableChain() {
+ public boolean isExistKilledTaskExecutionRunnableChain() {
return CollectionUtils.isNotEmpty(killedTaskChains);
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyPauseStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyPauseStateAction.java
index 9f002d1acb..8d677a3190 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyPauseStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyPauseStateAction.java
@@ -120,13 +120,13 @@ public class WorkflowReadyPauseStateAction extends
AbstractWorkflowStateAction {
}
final WorkflowEventBus workflowEventBus =
workflowExecutionRunnable.getWorkflowEventBus();
- if (workflowExecutionGraph.isExistFailureTaskExecutionRunnableChain())
{
-
workflowEventBus.publish(WorkflowFailedLifecycleEvent.of(workflowExecutionRunnable));
+ if (workflowExecutionGraph.isExistPausedTaskExecutionRunnableChain()) {
+
workflowEventBus.publish(WorkflowPausedLifecycleEvent.of(workflowExecutionRunnable));
return;
}
- if (workflowExecutionGraph.isExistPauseTaskExecutionRunnableChain()) {
-
workflowEventBus.publish(WorkflowPausedLifecycleEvent.of(workflowExecutionRunnable));
+ if (workflowExecutionGraph.isExistFailureTaskExecutionRunnableChain())
{
+
workflowEventBus.publish(WorkflowFailedLifecycleEvent.of(workflowExecutionRunnable));
return;
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java
index 7f40a4d714..b651d2223d 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowReadyStopStateAction.java
@@ -120,7 +120,7 @@ public class WorkflowReadyStopStateAction extends
AbstractWorkflowStateAction {
}
final WorkflowEventBus workflowEventBus =
workflowExecutionRunnable.getWorkflowEventBus();
- if (workflowExecutionGraph.isExistKillTaskExecutionRunnableChain()) {
+ if (workflowExecutionGraph.isExistKilledTaskExecutionRunnableChain()) {
workflowEventBus.publish(WorkflowStoppedLifecycleEvent.of(workflowExecutionRunnable));
return;
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java
index 934a727afe..469736cea5 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java
@@ -84,7 +84,13 @@ public class WorkflowRunningStateAction extends
AbstractWorkflowStateAction {
public void onStoppedEvent(final IWorkflowExecutionRunnable
workflowExecutionRunnable,
final WorkflowStoppedLifecycleEvent
workflowStoppedEvent) {
throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
- logWarningIfCannotDoAction(workflowExecutionRunnable,
workflowStoppedEvent);
+ // [Fix-17354]
+ if
(!workflowExecutionRunnable.getWorkflowExecutionGraph().isExistKilledTaskExecutionRunnableChain())
{
+ throw new IllegalStateException(
+ "The workflow: " + workflowExecutionRunnable.getName()
+ + " does not exist tasks chain which is killed");
+ }
+ super.workflowFinish(workflowExecutionRunnable,
WorkflowExecutionStatus.STOP);
}
@Override
@@ -142,6 +148,16 @@ public class WorkflowRunningStateAction extends
AbstractWorkflowStateAction {
return;
}
+ // [Fix-17354]
+ // If there exist tasks which has set timeout failed, then will
publish a kill event to kill the task.
+ // So there might exist task which is killed, and the workflow
instance state is running.
+ // This is a special case, the workflow instance can transform from
running to stop state.
+ // Is there better way to handle this case?
+ if (workflowExecutionGraph.isExistKilledTaskExecutionRunnableChain()) {
+
workflowEventBus.publish(WorkflowStoppedLifecycleEvent.of(workflowExecutionRunnable));
+ return;
+ }
+
if (workflowExecutionGraph.isAllTaskExecutionRunnableChainSuccess()) {
workflowEventBus.publish(WorkflowSucceedLifecycleEvent.of(workflowExecutionRunnable));
return;
@@ -149,6 +165,6 @@ public class WorkflowRunningStateAction extends
AbstractWorkflowStateAction {
throw new IllegalStateException("The workflow: " +
workflowExecutionRunnable.getName() +
" state is " + workflowExecutionRunnable.getState()
- + " can only finish with success/failed but exist task which
state is not success and failure");
+ + " can only finish with task success/failed/killed but exist
task which state is not success、failure、killed");
}
}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowTestCaseContext.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowTestCaseContext.java
index 0441427d67..35cee61c93 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowTestCaseContext.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/WorkflowTestCaseContext.java
@@ -61,4 +61,14 @@ public class WorkflowTestCaseContext {
}
return workflows.get(0);
}
+
+ public WorkflowDefinition getWorkflow(String name) {
+ if (CollectionUtils.isEmpty(workflows)) {
+ throw new IllegalStateException("workflows is empty");
+ }
+ return workflows.stream()
+ .filter(workflow -> workflow.getName().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("Workflow with
name " + name + " not found"));
+ }
}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
index 4c56976dfd..530b6b7d6f 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
@@ -1004,4 +1004,35 @@ public class WorkflowStartTestCase extends
AbstractMasterIntegrationTestCase {
});
masterContainer.assertAllResourceReleased();
}
+
+ @Test
+ @DisplayName("Test start a workflow which contains a dep task with timeout
kill strategy")
+ public void testStartWorkflow_withTimeoutKillTask() {
+ final String yaml = "/it/start/workflow_with_timeout_kill_task.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition workflow =
context.getWorkflow("workflow_with_timeout_kill_task");
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO =
WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(workflow)
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build();
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ await()
+ .atMost(Duration.ofSeconds(90))
+ .untilAsserted(() -> {
+ Assertions
+
.assertThat(repository.queryWorkflowInstance(workflow))
+ .satisfiesExactly(workflowInstance ->
assertThat(workflowInstance.getState())
+ .isEqualTo(WorkflowExecutionStatus.STOP));
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflow))
+ .hasSize(1)
+ .satisfiesExactly(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("dep_task_with_timeout_killed");
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.KILL);
+ });
+ });
+ masterContainer.assertAllResourceReleased();
+ }
}
diff --git
a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_timeout_kill_task.yaml
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_timeout_kill_task.yaml
new file mode 100644
index 0000000000..a4439c1e68
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_timeout_kill_task.yaml
@@ -0,0 +1,95 @@
+#
+# 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.
+#
+
+project:
+ name: MasterIntegrationTest
+ code: 1
+ description: This is a fake project
+ userId: 1
+ userName: admin
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+
+workflows:
+ - name: workflow_with_one_fake_task_success
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a fake workflow with single task
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+ - name: workflow_with_timeout_kill_task
+ code: 2
+ version: 1
+ projectCode: 1
+ description: This is a fake workflow with single timeout task
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: B
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":[],"shellScript":"if [
\"${system.project.name}\" = \"MasterIntegrationTest\" ]; then\n exit 0
\nelse\n exit 1\nfi","resourceList":[]}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: dep_task_with_timeout_killed
+ code: 2
+ version: 1
+ projectCode: 1
+ userId: 1
+ timeoutFlag: 'OPEN'
+ timeoutNotifyStrategy: 'FAILED'
+ timeout: 1
+ taskType: DEPENDENT
+ taskParams:
'{"localParams":[],"resourceList":[],"dependence":{"checkInterval":10,"failurePolicy":"DEPENDENT_FAILURE_FAILURE","relation":"AND","dependTaskList":[{"relation":"AND","dependItemList":[{"dependentType":"DEPENDENT_ON_WORKFLOW","projectCode":1,"definitionCode":1,"depTaskCode":0,"cycle":"day","dateValue":"last1Days"}]}]}}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+
+taskRelations:
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 0
+ preTaskVersion: 0
+ postTaskCode: 1
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 2
+ workflowDefinitionVersion: 1
+ preTaskCode: 0
+ preTaskVersion: 0
+ postTaskCode: 2
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+
