This is an automated email from the ASF dual-hosted git repository.
SbloodyS 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 86edf18211 [Fix-18604][Master] Preserve forced-success state when
recovering failed workflows (#18608)
86edf18211 is described below
commit 86edf18211c552a7efcbc91df1336850f5bd573f
Author: Molin Wang <[email protected]>
AuthorDate: Wed Sep 2 10:34:19 2026 +0800
[Fix-18604][Master] Preserve forced-success state when recovering failed
workflows (#18608)
---
.../statemachine/TaskForceSuccessStateAction.java | 9 ++
.../TaskForceSuccessStateActionTest.java | 88 ++++++++++++
...WorkflowInstanceRecoverFailureTaskTestCase.java | 44 ++++++
...e_workflow_with_forced_success_predecessor.yaml | 154 +++++++++++++++++++++
4 files changed, 295 insertions(+)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateAction.java
index 3c34c52f28..d2e8b7fe18 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateAction.java
@@ -18,6 +18,8 @@
package org.apache.dolphinscheduler.server.master.engine.task.statemachine;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import
org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent;
import lombok.extern.slf4j.Slf4j;
@@ -27,6 +29,13 @@ import org.springframework.stereotype.Component;
@Component
public class TaskForceSuccessStateAction extends TaskSuccessStateAction {
+ @Override
+ protected void persistentTaskInstanceSuccessEventToDB(final ITaskExecution
taskExecution,
+ final
TaskSuccessLifecycleEvent taskSuccessEvent) {
+ // FORCED_SUCCESS is already persisted by the API.
+ // Skip the SUCCESS update while retaining variable merging and
downstream transitions.
+ }
+
@Override
public TaskExecutionStatus matchState() {
return TaskExecutionStatus.FORCED_SUCCESS;
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateActionTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateActionTest.java
new file mode 100644
index 0000000000..c07ab9920b
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskForceSuccessStateActionTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.dolphinscheduler.server.master.engine.task.statemachine;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.server.master.engine.AbstractLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.ITaskGroupCoordinator;
+import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
+import
org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskStartLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class TaskForceSuccessStateActionTest {
+
+ private TaskForceSuccessStateAction taskForceSuccessStateAction;
+
+ @Mock
+ private ITaskGroupCoordinator taskGroupCoordinator;
+
+ @Mock
+ private TaskInstanceDao taskInstanceDao;
+
+ @Mock
+ private ITaskExecution taskExecution;
+
+ @Mock
+ private IWorkflowExecution workflowExecution;
+
+ @Mock
+ private WorkflowEventBus workflowEventBus;
+
+ private TaskInstance taskInstance;
+
+ @BeforeEach
+ void setUp() {
+ taskForceSuccessStateAction = new TaskForceSuccessStateAction();
+ taskForceSuccessStateAction.taskGroupCoordinator =
taskGroupCoordinator;
+ taskForceSuccessStateAction.taskInstanceDao = taskInstanceDao;
+
+ taskInstance = new TaskInstance();
+ taskInstance.setState(TaskExecutionStatus.FORCED_SUCCESS);
+
+ when(taskExecution.getTaskInstance()).thenReturn(taskInstance);
+ when(taskExecution.getWorkflowEventBus()).thenReturn(workflowEventBus);
+ when(workflowExecution.getWorkflowInstance()).thenReturn(new
WorkflowInstance());
+ }
+
+ @Test
+ void shouldPreserveForcedSuccessWhileContinuingWorkflow() {
+ taskForceSuccessStateAction.onStartEvent(
+ workflowExecution,
+ taskExecution,
+ TaskStartLifecycleEvent.of(taskExecution));
+
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.FORCED_SUCCESS);
+ verify(workflowEventBus).publish(any(AbstractLifecycleEvent.class));
+ }
+}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverFailureTaskTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverFailureTaskTestCase.java
index 90144c9007..99cb697c56 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverFailureTaskTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverFailureTaskTestCase.java
@@ -94,6 +94,50 @@ public class WorkflowInstanceRecoverFailureTaskTestCase
extends AbstractMasterIn
masterContainer.assertAllResourceReleased();
}
+ @Test
+ @DisplayName("Test recover failure tasks preserves forced-success task
state")
+ public void testRecoverFailureTasks_preservesForcedSuccessTaskState() {
+ final String yaml =
+
"/it/recover_failure_tasks/failure_workflow_with_forced_success_predecessor.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+
+ final Integer workflowInstanceId =
context.getWorkflowInstances().get(0).getId();
+ workflowOperator.recoverFailureTasks(workflowInstanceId);
+
+ await()
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+ final WorkflowInstance workflowInstance =
repository.queryWorkflowInstance(workflowInstanceId);
+ assertThat(workflowInstance.getState())
+ .isEqualTo(WorkflowExecutionStatus.SUCCESS);
+ assertThat(workflowInstance.getRunTimes())
+ .isEqualTo(2);
+
+ final List<TaskInstance> taskInstances =
repository.queryTaskInstance(workflowInstanceId);
+ assertThat(taskInstances)
+ .hasSize(3);
+
+ assertThat(taskInstances)
+ .filteredOn(t -> "A".equals(t.getName()))
+ .singleElement()
+ .matches(t -> t.getState() ==
TaskExecutionStatus.FORCED_SUCCESS)
+ .matches(t -> t.getFlag() == Flag.YES);
+
+ assertThat(taskInstances)
+ .filteredOn(t -> "B".equals(t.getName()))
+ .anySatisfy(t -> {
+
assertThat(t.getState()).isEqualTo(TaskExecutionStatus.FAILURE);
+ assertThat(t.getFlag()).isEqualTo(Flag.NO);
+ })
+ .anySatisfy(t -> {
+
assertThat(t.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ assertThat(t.getFlag()).isEqualTo(Flag.YES);
+ assertThat(t.getLogPath()).isNotEmpty();
+ });
+ });
+ masterContainer.assertAllResourceReleased();
+ }
+
@Test
@DisplayName("Test recover a failure workflow from another master")
public void testRecoverFailureWorkflow_from_another_master() {
diff --git
a/dolphinscheduler-master/src/test/resources/it/recover_failure_tasks/failure_workflow_with_forced_success_predecessor.yaml
b/dolphinscheduler-master/src/test/resources/it/recover_failure_tasks/failure_workflow_with_forced_success_predecessor.yaml
new file mode 100644
index 0000000000..1e29530fa5
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/recover_failure_tasks/failure_workflow_with_forced_success_predecessor.yaml
@@ -0,0 +1,154 @@
+#
+# 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
+
+workflowInstances:
+ - id: 1
+ name: workflow_with_forced_success_predecessor-20240816071251690
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ projectCode: 1
+ state: FAILURE
+ recovery: NO
+ startTime: 2024-08-16 07:12:52
+ endTime: 2024-08-16 07:12:57
+ runTimes: 1
+ host: 127.0.0.1:5678
+ commandType: START_PROCESS
+ commandParam:
'{"commandType":"START_PROCESS","startNodes":[],"commandParams":[],"timeZone":"UTC"}'
+ taskDependType: TASK_POST
+ commandStartTime: 2024-08-16 07:12:52
+ isSubWorkflow: NO
+ executorId: 1
+ historyCmd: START_PROCESS
+ workerGroup: default
+ globalParams: '[]'
+ varPool: '[]'
+ dryRun: 0
+
+taskInstances:
+ - id: 1
+ name: A
+ taskType: LogicFakeTask
+ workflowInstanceId: 1
+ workflowInstanceName:
workflow_with_forced_success_predecessor-20240816071251690
+ projectCode: 1
+ taskCode: 1
+ taskDefinitionVersion: 1
+ state: FORCED_SUCCESS
+ firstSubmitTime: 2024-08-16 07:12:52
+ submitTime: 2024-08-16 07:12:53
+ startTime: 2024-08-16 07:12:53
+ endTime: 2024-08-16 07:12:54
+ retryTimes: 0
+ host: 127.0.0.1:1234
+ maxRetryTimes: 0
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo hello"}'
+ flag: YES
+ retryInterval: 0
+ delayTime: 0
+ workerGroup: default
+ executorId: 1
+ varPool: '[]'
+ taskExecuteType: BATCH
+ - id: 2
+ name: B
+ taskType: LogicFakeTask
+ workflowInstanceId: 1
+ workflowInstanceName:
workflow_with_forced_success_predecessor-20240816071251690
+ projectCode: 1
+ taskCode: 2
+ taskDefinitionVersion: 1
+ state: FAILURE
+ firstSubmitTime: 2024-08-16 07:12:54
+ submitTime: 2024-08-16 07:12:55
+ startTime: 2024-08-16 07:12:55
+ endTime: 2024-08-16 07:12:57
+ retryTimes: 0
+ host: 127.0.0.1:1234
+ maxRetryTimes: 0
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo hello"}'
+ flag: YES
+ retryInterval: 0
+ delayTime: 0
+ workerGroup: default
+ executorId: 1
+ varPool: '[]'
+ taskExecuteType: BATCH
+
+workflows:
+ - name: workflow_with_forced_success_predecessor
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a fake workflow with a forced-success predecessor
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: A
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo hello"}'
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: B
+ code: 2
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: '{"localParams":null,"varPool":[],"shellScript":"echo hello"}'
+ 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: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 1
+ preTaskVersion: 1
+ postTaskCode: 2
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
