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 03da5340ba [Fix-18707][Master] Keep task success retryable after
persistence failure (#18708)
03da5340ba is described below
commit 03da5340ba29e55528e86efc1c489fdf53542737
Author: ThreeFire1 <[email protected]>
AuthorDate: Fri Oct 16 15:23:12 2026 +0800
[Fix-18707][Master] Keep task success retryable after persistence failure
(#18708)
---
.../task/statemachine/AbstractTaskStateAction.java | 76 ++-
.../statemachine/TaskSubmittedStateAction.java | 7 +-
.../TaskLifecyclePersistenceRecoveryTest.java | 502 ++++++++++++++++++
.../TaskSuccessPersistenceRecoveryTest.java | 564 +++++++++++++++++++++
.../TaskLifecyclePersistenceRecoveryTestCase.java | 134 +++++
.../TaskSuccessPersistenceRecoveryTestCase.java | 93 ++++
6 files changed, 1344 insertions(+), 32 deletions(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
index d053676e88..b62fe66c74 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
@@ -49,9 +49,11 @@ import
org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event
import org.apache.commons.lang3.StringUtils;
import java.util.List;
+import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import com.google.common.collect.Lists;
@@ -124,9 +126,10 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
private void persistentTaskInstanceFatalEventToDB(final ITaskExecution
taskExecution,
final
TaskFatalLifecycleEvent taskFatalEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.FAILURE);
- taskInstance.setEndTime(taskFatalEvent.getEndTime());
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(TaskExecutionStatus.FAILURE);
+ instance.setEndTime(taskFatalEvent.getEndTime());
+ });
}
@Override
@@ -134,9 +137,10 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
final ITaskExecution taskExecution,
final TaskDispatchedLifecycleEvent
taskDispatchedEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(DISPATCH);
- taskInstance.setHost(taskDispatchedEvent.getExecutorHost());
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(DISPATCH);
+ instance.setHost(taskDispatchedEvent.getExecutorHost());
+ });
}
@Override
@@ -144,19 +148,21 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
final ITaskExecution
taskExecution,
final
TaskRuntimeContextChangedEvent taskRuntimeContextChangedEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- if
(StringUtils.isNotEmpty(taskRuntimeContextChangedEvent.getRuntimeContext())) {
-
taskInstance.setAppLink(taskRuntimeContextChangedEvent.getRuntimeContext());
- }
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ if
(StringUtils.isNotEmpty(taskRuntimeContextChangedEvent.getRuntimeContext())) {
+
instance.setAppLink(taskRuntimeContextChangedEvent.getRuntimeContext());
+ }
+ });
}
protected void persistentTaskInstanceStartedEventToDB(final ITaskExecution
taskExecution,
final
TaskRunningLifecycleEvent taskRunningEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION);
- taskInstance.setStartTime(taskRunningEvent.getStartTime());
- taskInstance.setLogPath(taskRunningEvent.getLogPath());
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(TaskExecutionStatus.RUNNING_EXECUTION);
+ instance.setStartTime(taskRunningEvent.getStartTime());
+ instance.setLogPath(taskRunningEvent.getLogPath());
+ });
}
@Override
@@ -172,8 +178,7 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
private void persistentTaskInstancePausedEventToDB(final ITaskExecution
taskExecution,
final
TaskPausedLifecycleEvent taskPausedEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.PAUSE);
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance ->
instance.setState(TaskExecutionStatus.PAUSE));
}
@Override
@@ -189,9 +194,10 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
private void persistentTaskInstanceKilledEventToDB(final ITaskExecution
taskExecution,
final
TaskKilledLifecycleEvent taskKilledEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.KILL);
- taskInstance.setEndTime(taskKilledEvent.getEndTime());
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(TaskExecutionStatus.KILL);
+ instance.setEndTime(taskKilledEvent.getEndTime());
+ });
}
@@ -222,9 +228,10 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
private void persistentTaskInstanceFailedEventToDB(final ITaskExecution
taskExecution,
final
TaskFailedLifecycleEvent taskFailedEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.FAILURE);
- taskInstance.setEndTime(taskFailedEvent.getEndTime());
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(TaskExecutionStatus.FAILURE);
+ instance.setEndTime(taskFailedEvent.getEndTime());
+ });
}
@Override
@@ -246,15 +253,32 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
workflowInstance.setVarPool(VarPoolUtils.serializeVarPool(finalVarPool));
}
+ /**
+ * Persist an event's fields before applying them to the shared instance,
so a failed update remains retryable.
+ * The updater must only assign the event's fields and must be safe to
apply to both instances. Compute merged
+ * values before calling this method; do not publish events or mutate
nested objects in the updater.
+ */
+ protected void persistTaskInstanceUpdate(final TaskInstance taskInstance,
+ final Consumer<TaskInstance>
updater) {
+ final TaskInstance taskInstanceToUpdate = new TaskInstance();
+ BeanUtils.copyProperties(taskInstance, taskInstanceToUpdate);
+ updater.accept(taskInstanceToUpdate);
+ taskInstanceDao.updateById(taskInstanceToUpdate);
+ updater.accept(taskInstance);
+ }
+
protected void persistentTaskInstanceSuccessEventToDB(final ITaskExecution
taskExecution,
final
TaskSuccessLifecycleEvent taskSuccessEvent) {
final TaskInstance taskInstance = taskExecution.getTaskInstance();
- taskInstance.setState(TaskExecutionStatus.SUCCESS);
- taskInstance.setEndTime(taskSuccessEvent.getEndTime());
final List<Property> finalVarPool =
VarPoolUtils.mergeVarPoolJsonString(taskInstance.getVarPool(),
JSONUtils.toJsonString(taskSuccessEvent.getVarPool()));
- taskInstance.setVarPool(VarPoolUtils.serializeVarPool(finalVarPool));
- taskInstanceDao.updateById(taskInstance);
+ final String serializedFinalVarPool =
VarPoolUtils.serializeVarPool(finalVarPool);
+
+ persistTaskInstanceUpdate(taskInstance, instance -> {
+ instance.setState(TaskExecutionStatus.SUCCESS);
+ instance.setEndTime(taskSuccessEvent.getEndTime());
+ instance.setVarPool(serializedFinalVarPool);
+ });
}
/**
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
index 05b98f93c7..ec055cc5e9 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
@@ -19,7 +19,6 @@ package
org.apache.dolphinscheduler.server.master.engine.task.statemachine;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import
org.apache.dolphinscheduler.server.master.engine.task.dispatcher.WorkerGroupDispatcherCoordinator;
import
org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
@@ -56,9 +55,6 @@ public class TaskSubmittedStateAction extends
AbstractTaskStateAction {
@Autowired
private WorkerGroupDispatcherCoordinator workerGroupDispatcherCoordinator;
- @Autowired
- private TaskInstanceDao taskInstanceDao;
-
@Override
public void onStartEvent(final IWorkflowExecution workflowExecution,
final ITaskExecution taskExecution,
@@ -104,8 +100,7 @@ public class TaskSubmittedStateAction extends
AbstractTaskStateAction {
taskInstance.getFirstSubmitTime(),
taskInstance.getDelayTime() * 60L) * 1_000;
if (remainTimeMills > 0) {
- taskInstance.setState(TaskExecutionStatus.DELAY_EXECUTION);
- taskInstanceDao.updateById(taskInstance);
+ persistTaskInstanceUpdate(taskInstance, instance ->
instance.setState(TaskExecutionStatus.DELAY_EXECUTION));
log.info("Current taskInstance: {} is choose delay execution,
delay time: {}/min, remainTime: {}/ms",
taskInstance.getName(),
taskInstance.getDelayTime(),
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskLifecyclePersistenceRecoveryTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskLifecyclePersistenceRecoveryTest.java
new file mode 100644
index 0000000000..2ba68ae3c9
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskLifecyclePersistenceRecoveryTest.java
@@ -0,0 +1,502 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
+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.ILifecycleEventHandler;
+import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
+import org.apache.dolphinscheduler.server.master.engine.ITaskGroupCoordinator;
+import org.apache.dolphinscheduler.server.master.engine.TaskGroupCoordinator;
+import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
+import
org.apache.dolphinscheduler.server.master.engine.WorkflowEventBusFireWorker;
+import
org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraph;
+import
org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient;
+import
org.apache.dolphinscheduler.server.master.engine.task.dispatcher.WorkerGroupDispatcherCoordinator;
+import
org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskDispatchLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskDispatchedLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFailedLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFatalLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskKilledLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskPausedLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskRetryLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskRunningLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskRuntimeContextChangedEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.AbstractTaskLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskDispatchLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskDispatchedLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskFailedLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskFatalLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskKilledLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskPausedLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskRunningLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskRuntimeContextChangedLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.WorkflowLifecycleEventType;
+import
org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
+import
org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.springframework.beans.BeanUtils;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.test.util.ReflectionTestUtils;
+
+class TaskLifecyclePersistenceRecoveryTest {
+
+ private enum EventType {
+ DISPATCHED, RUNNING, PAUSED, KILLED, FAILED, FATAL, CONTEXT, DELAY
+ }
+
+ @ParameterizedTest
+ @EnumSource(EventType.class)
+ void shouldKeepSharedFieldsUnchangedUntilPersistenceRecovers(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ final List<Object> original = fields(fixture.shared);
+ fixture.fail.set(true);
+
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+
+ assertThat(fields(fixture.shared)).isEqualTo(original);
+ assertThat(fields(fixture.persisted)).isEqualTo(original);
+ verifyNoInteractions(fixture.client, fixture.graph,
fixture.dispatcher);
+ verify(fixture.bus,
never()).publish(any(AbstractLifecycleEvent.class));
+
+ fixture.fail.set(false);
+ fixture.deliver.run();
+
+ assertThat(fixture.attempts).hasValue(2);
+ fixture.assertPersistedResult();
+ fixture.assertAck(type == EventType.FATAL || type == EventType.DELAY ?
0 : 1);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"DISPATCHED", "PAUSED",
"KILLED"})
+ void
shouldRetryTheDatabaseUpdateInsteadOfAcknowledgingAnUnpersistedState(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ fixture.fail.set(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+
+ fixture.fail.set(false);
+ fixture.deliver.run();
+
+ assertThat(fixture.attempts).hasValue(2);
+ fixture.assertPersistedResult();
+ fixture.assertAck(1);
+ }
+
+ @ParameterizedTest
+ @EnumSource(EventType.class)
+ void shouldNotAcknowledgeOrAdvanceWhilePersistenceKeepsFailing(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ fixture.fail.set(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+ assertThat(fixture.attempts).hasValue(2);
+ verifyNoInteractions(fixture.client, fixture.graph,
fixture.dispatcher);
+ verify(fixture.bus,
never()).publish(any(AbstractLifecycleEvent.class));
+ }
+
+ @ParameterizedTest
+ @EnumSource(EventType.class)
+ void
shouldPersistNormallyWithoutReplacingTheSharedInstanceOrUnrelatedFields(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+
when(fixture.dao.updateById(any(TaskInstance.class))).thenAnswer(invocation -> {
+ final TaskInstance candidate = invocation.getArgument(0);
+ assertThat(candidate).isNotSameAs(fixture.shared);
+ BeanUtils.copyProperties(candidate, fixture.persisted);
+ // A field outside this event's update must not be overwritten by
copying the whole candidate back.
+ fixture.shared.setName("updated-name");
+ return true;
+ });
+ fixture.deliver.run();
+ fixture.assertPersistedResult();
+ assertThat(fixture.task.getTaskInstance()).isSameAs(fixture.shared);
+ assertThat(fixture.shared.getName()).isEqualTo("updated-name");
+ }
+
+ @ParameterizedTest
+ @EnumSource(EventType.class)
+ void shouldPreserveExistingBehaviorWhenUpdateReturnsFalse(final EventType
type) {
+ final Fixture fixture = new Fixture(type);
+
when(fixture.dao.updateById(any(TaskInstance.class))).thenReturn(false);
+ fixture.deliver.run();
+
assertThat(fixture.shared.getState()).isEqualTo(fixture.expectedState());
+ fixture.assertAck(type == EventType.FATAL || type == EventType.DELAY ?
0 : 1);
+ verify(fixture.dao, never()).queryById(any());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"DISPATCHED", "PAUSED",
"KILLED"})
+ void
shouldAcknowledgeAgainWithoutRepeatingPersistenceAfterAckFailure(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ doThrow(new IllegalStateException("synthetic ACK
failure")).doNothing().when(fixture.client)
+ .ackTaskExecutorLifecycleEvent(eq(fixture.task), any());
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(IllegalStateException.class);
+ fixture.assertPersistedResult();
+ fixture.deliver.run();
+ assertThat(fixture.attempts).hasValue(1);
+ fixture.assertAck(2);
+ verify(fixture.bus, times(type == EventType.DISPATCHED ? 0 :
1)).publish(any(AbstractLifecycleEvent.class));
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"FAILED", "FATAL"})
+ void shouldScheduleTaskRetryOnlyAfterFailureIsPersisted(final EventType
type) {
+ final Fixture fixture = new Fixture(type);
+ fixture.shared.setMaxRetryTimes(1);
+ when(fixture.task.isTaskInstanceCanRetry()).thenReturn(true);
+ fixture.fail.set(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+ verify(fixture.bus,
never()).publish(any(AbstractLifecycleEvent.class));
+ fixture.fail.set(false);
+ fixture.deliver.run();
+ verify(fixture.bus).publish(any(TaskRetryLifecycleEvent.class));
+ verifyNoInteractions(fixture.graph);
+ fixture.assertPersistedResult();
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"FAILED", "FATAL"})
+ void shouldKeepConditionSuccessorsEligibleAfterPersistenceRecovers(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+
when(fixture.graph.isAllSuccessorsAreConditionTask(fixture.task)).thenReturn(true);
+ fixture.fail.set(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+ fixture.fail.set(false);
+ fixture.deliver.run();
+ verify(fixture.graph,
never()).markTaskExecutionChainFailure(fixture.task);
+ verify(fixture.bus).publish(any(AbstractLifecycleEvent.class));
+ fixture.assertPersistedResult();
+ }
+
+ @Test
+ void shouldKeepExistingAppLinkWhenRuntimeContextIsEmpty() {
+ final Fixture fixture = new Fixture(EventType.CONTEXT);
+ fixture.bind(new
TaskRuntimeContextChangedLifecycleEventHandler(fixture.client),
+ new TaskRuntimeContextChangedEvent(fixture.task, ""));
+ fixture.deliver.run();
+ assertThat(fixture.shared.getAppLink()).isEqualTo("old-context");
+ assertThat(fixture.persisted.getAppLink()).isEqualTo("old-context");
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"PAUSED", "DISPATCHED"})
+ void shouldRecoverThroughTheRealWorkflowEventBus(final EventType type) {
+ final Fixture fixture = new Fixture(type);
+ final IWorkflowExecuteContext context =
mock(IWorkflowExecuteContext.class);
+ final WorkflowInstance workflowInstance =
fixture.workflow.getWorkflowInstance();
+ when(context.getWorkflowInstance()).thenReturn(workflowInstance);
+ when(fixture.workflow.getWorkflowExecuteContext()).thenReturn(context);
+ when(fixture.workflow.getId()).thenReturn(1);
+ final WorkflowEventBusFireWorker worker = new
WorkflowEventBusFireWorker();
+ worker.registerWorkflowEventBus(fixture.workflow);
+ worker.registerEventHandler(fixture.handler);
+ worker.registerEventHandler(new
ILifecycleEventHandler<AbstractLifecycleEvent>() {
+
+ @Override
+ public void handle(final IWorkflowExecution workflow, final
AbstractLifecycleEvent event) {
+ // Drain the downstream event; the task event's retry and
persistence are under test.
+ }
+
+ @Override
+ public ILifecycleEventType matchEventType() {
+ return
WorkflowLifecycleEventType.TOPOLOGY_LOGICAL_TRANSACTION_WITH_TASK_FINISH;
+ }
+ });
+ fixture.bus.publish(fixture.event);
+ fixture.fail.set(true);
+ worker.fireAllRegisteredEvent();
+ assertThat(fixture.attempts).hasValue(1);
+ fixture.assertAck(0);
+ assertThat(fixture.bus.isEmpty()).isFalse();
+ fixture.fail.set(false);
+ worker.fireAllRegisteredEvent();
+ assertThat(fixture.attempts).hasValue(2);
+ fixture.assertPersistedResult();
+ fixture.assertAck(1);
+ assertThat(fixture.bus.isEmpty()).isTrue();
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"PAUSED", "KILLED"})
+ void shouldRetryResourceReleaseBeforePersistingTheTerminalState(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ final TaskGroupQueueDao queueDao =
fixture.useRealTaskGroupCoordinator();
+ when(queueDao.queryByTaskInstanceId(1))
+ .thenThrow(new DataAccessResourceFailureException("synthetic
task group query failure"))
+ .thenReturn(Collections.emptyList());
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+
assertThat(fixture.shared.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(fixture.attempts).hasValue(0);
+ fixture.assertAck(0);
+ fixture.deliver.run();
+ verify(queueDao, times(2)).queryByTaskInstanceId(1);
+ fixture.assertPersistedResult();
+ fixture.assertAck(1);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"PAUSED", "KILLED"})
+ void
shouldRetryAQueueDeletionFailureBeforePersistingTheTerminalState(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ final TaskGroupQueueDao queueDao =
fixture.useRealTaskGroupCoordinator();
+ final TaskGroupQueue queue = new TaskGroupQueue();
+ queue.setId(10);
+
when(queueDao.queryByTaskInstanceId(1)).thenReturn(Collections.singletonList(queue));
+ when(queueDao.deleteById(queue))
+ .thenThrow(new DataAccessResourceFailureException("synthetic
task group deletion failure"))
+ .thenReturn(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+
assertThat(fixture.shared.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(fixture.attempts).hasValue(0);
+ fixture.assertAck(0);
+ fixture.deliver.run();
+ verify(queueDao, times(2)).deleteById(queue);
+ fixture.assertPersistedResult();
+ fixture.assertAck(1);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = EventType.class, names = {"PAUSED", "KILLED"})
+ void
shouldNotReleaseTheSameQueueEntryAgainWhenTheTaskUpdateIsRetried(final
EventType type) {
+ final Fixture fixture = new Fixture(type);
+ final TaskGroupQueueDao queueDao =
fixture.useRealTaskGroupCoordinator();
+ final TaskGroupQueue queue = new TaskGroupQueue();
+ queue.setId(10);
+ final AtomicBoolean allocated = new AtomicBoolean(true);
+ when(queueDao.queryByTaskInstanceId(1)).thenAnswer(invocation ->
allocated.get()
+ ? Collections.singletonList(queue)
+ : Collections.emptyList());
+ when(queueDao.deleteById(queue)).thenAnswer(invocation ->
allocated.getAndSet(false));
+ fixture.fail.set(true);
+
assertThatThrownBy(fixture.deliver::run).isInstanceOf(DataAccessResourceFailureException.class);
+ assertThat(allocated).isFalse();
+ fixture.fail.set(false);
+ fixture.deliver.run();
+ verify(queueDao, times(2)).queryByTaskInstanceId(1);
+ verify(queueDao).deleteById(queue);
+ fixture.assertPersistedResult();
+ fixture.assertAck(1);
+ }
+
+ private static List<Object> fields(final TaskInstance instance) {
+ return Arrays.asList(instance.getState(), instance.getHost(),
instance.getStartTime(),
+ instance.getEndTime(), instance.getLogPath(),
instance.getAppLink(), instance.getVarPool());
+ }
+
+ private static final class Fixture {
+
+ private static final Date EVENT_TIME = new Date(1_000L);
+ private final TaskInstanceDao dao = mock(TaskInstanceDao.class);
+ private final ITaskGroupCoordinator taskGroup =
mock(ITaskGroupCoordinator.class);
+ private final TaskExecutorClient client =
mock(TaskExecutorClient.class);
+ private final WorkerGroupDispatcherCoordinator dispatcher =
mock(WorkerGroupDispatcherCoordinator.class);
+ private final ITaskExecution task = mock(ITaskExecution.class);
+ private final IWorkflowExecution workflow =
mock(IWorkflowExecution.class);
+ private final IWorkflowExecutionGraph graph =
mock(IWorkflowExecutionGraph.class);
+ private final WorkflowEventBus bus = spy(new WorkflowEventBus());
+ private final TaskInstance shared = new TaskInstance();
+ private final TaskInstance persisted = new TaskInstance();
+ private final AtomicBoolean fail = new AtomicBoolean();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final EventType type;
+ private final TaskStateActionFactory factory;
+ private Runnable deliver;
+ private AbstractTaskLifecycleEventHandler<?> handler;
+ private AbstractTaskLifecycleEvent event;
+
+ private Fixture(final EventType type) {
+ this.type = type;
+ shared.setId(1);
+ shared.setName("task-persistence-recovery");
+ shared.setState(type == EventType.DISPATCHED || type ==
EventType.DELAY
+ ? TaskExecutionStatus.SUBMITTED_SUCCESS
+ : type == EventType.RUNNING ? TaskExecutionStatus.DISPATCH
: TaskExecutionStatus.RUNNING_EXECUTION);
+ shared.setHost("old-worker:1234");
+ shared.setLogPath("old-log");
+ shared.setAppLink("old-context");
+ shared.setVarPool("[]");
+ shared.setFirstSubmitTime(new Date());
+ shared.setDelayTime(10);
+ BeanUtils.copyProperties(shared, persisted);
+ final WorkflowInstance workflowInstance = new WorkflowInstance();
+ workflowInstance.setId(1);
+ workflowInstance.setVarPool("[]");
+ when(task.getTaskInstance()).thenReturn(shared);
+ when(task.getId()).thenReturn(1);
+ when(task.getWorkflowEventBus()).thenReturn(bus);
+ when(task.getWorkflowExecutionGraph()).thenReturn(graph);
+ when(workflow.getWorkflowInstance()).thenReturn(workflowInstance);
+ when(workflow.getWorkflowEventBus()).thenReturn(bus);
+
when(dao.updateById(any(TaskInstance.class))).thenAnswer(invocation -> {
+ attempts.incrementAndGet();
+ if (fail.get()) {
+ throw new DataAccessResourceFailureException("synthetic
task update failure");
+ }
+ BeanUtils.copyProperties(invocation.getArgument(0), persisted);
+ return true;
+ });
+ final List<ITaskStateAction> actions = new ArrayList<>();
+ final List<AbstractTaskStateAction> realActions =
Arrays.asList(new TaskSubmittedStateAction(),
+ new TaskDispatchStateAction(), new
TaskRunningStateAction(), new TaskPauseStateAction(),
+ new TaskKillStateAction(), new TaskFailureStateAction(),
new TaskDelayExecutionStateAction());
+ for (TaskExecutionStatus state : TaskExecutionStatus.values()) {
+ final AbstractTaskStateAction real =
realActions.stream().filter(a -> a.matchState() == state)
+ .findFirst().orElse(null);
+ if (real != null) {
+ real.taskInstanceDao = dao;
+ real.taskGroupCoordinator = taskGroup;
+ if (real instanceof TaskSubmittedStateAction) {
+ // Also initialize the shadowing field on the
unmodified implementation for the red test.
+ ReflectionTestUtils.setField(real, "taskInstanceDao",
dao);
+ ReflectionTestUtils.setField(real,
"workerGroupDispatcherCoordinator", dispatcher);
+ }
+ actions.add(real);
+ } else {
+ final ITaskStateAction action =
mock(ITaskStateAction.class);
+ when(action.matchState()).thenReturn(state);
+ actions.add(action);
+ }
+ }
+ factory = new TaskStateActionFactory(actions);
+ switch (type) {
+ case DISPATCHED:
+ bind(new TaskDispatchedLifecycleEventHandler(client),
+ new TaskDispatchedLifecycleEvent(task,
"new-worker:1234"));
+ break;
+ case RUNNING:
+ bind(new TaskRunningLifecycleEventHandler(client),
+ new TaskRunningLifecycleEvent(task, "new-log",
EVENT_TIME));
+ break;
+ case PAUSED:
+ bind(new TaskPausedLifecycleEventHandler(client),
TaskPausedLifecycleEvent.of(task));
+ break;
+ case KILLED:
+ bind(new TaskKilledLifecycleEventHandler(client), new
TaskKilledLifecycleEvent(task, EVENT_TIME));
+ break;
+ case FAILED:
+ bind(new TaskFailedLifecycleEventHandler(client), new
TaskFailedLifecycleEvent(task, EVENT_TIME));
+ break;
+ case FATAL:
+ bind(new TaskFatalLifecycleEventHandler(), new
TaskFatalLifecycleEvent(task, EVENT_TIME));
+ break;
+ case CONTEXT:
+ bind(new
TaskRuntimeContextChangedLifecycleEventHandler(client),
+ new TaskRuntimeContextChangedEvent(task,
"new-context"));
+ break;
+ case DELAY:
+ bind(new TaskDispatchLifecycleEventHandler(),
TaskDispatchLifecycleEvent.of(task));
+ break;
+ default:
+ throw new IllegalArgumentException(type.name());
+ }
+ }
+
+ private <T extends AbstractTaskLifecycleEvent> void bind(final
AbstractTaskLifecycleEventHandler<T> handler,
+ final T
event) {
+ this.handler = handler;
+ this.event = event;
+ ReflectionTestUtils.setField(handler, "taskStateActionFactory",
factory);
+ deliver = () -> handler.handle(workflow, event);
+ }
+
+ private TaskGroupQueueDao useRealTaskGroupCoordinator() {
+ shared.setTaskGroupId(1);
+ final TaskGroupQueueDao queueDao = mock(TaskGroupQueueDao.class);
+ final TaskGroupCoordinator coordinator = new
TaskGroupCoordinator();
+ ReflectionTestUtils.setField(coordinator, "taskGroupQueueDao",
queueDao);
+ ((AbstractTaskStateAction) factory
+
.getTaskStateAction(TaskExecutionStatus.RUNNING_EXECUTION)).taskGroupCoordinator
= coordinator;
+ return queueDao;
+ }
+
+ private TaskExecutionStatus expectedState() {
+ switch (type) {
+ case DISPATCHED:
+ return TaskExecutionStatus.DISPATCH;
+ case PAUSED:
+ return TaskExecutionStatus.PAUSE;
+ case KILLED:
+ return TaskExecutionStatus.KILL;
+ case FAILED:
+ case FATAL:
+ return TaskExecutionStatus.FAILURE;
+ case DELAY:
+ return TaskExecutionStatus.DELAY_EXECUTION;
+ default:
+ return TaskExecutionStatus.RUNNING_EXECUTION;
+ }
+ }
+
+ private void assertPersistedResult() {
+ assertThat(shared.getState()).isEqualTo(expectedState());
+ assertThat(fields(persisted)).isEqualTo(fields(shared));
+ if (type == EventType.DISPATCHED) {
+ assertThat(shared.getHost()).isEqualTo("new-worker:1234");
+ } else if (type == EventType.RUNNING) {
+ assertThat(shared.getStartTime()).isEqualTo(EVENT_TIME);
+ assertThat(shared.getLogPath()).isEqualTo("new-log");
+ } else if (type == EventType.CONTEXT) {
+ assertThat(shared.getAppLink()).isEqualTo("new-context");
+ } else if (type == EventType.KILLED || type == EventType.FAILED ||
type == EventType.FATAL) {
+ assertThat(shared.getEndTime()).isEqualTo(EVENT_TIME);
+ } else if (type == EventType.PAUSED) {
+ assertThat(shared.getEndTime()).isNull();
+ verify(graph).markTaskExecutionChainPause(task);
+ } else if (type == EventType.DELAY) {
+ verify(dispatcher).dispatchTask(eq(task), anyLong());
+ }
+ }
+
+ private void assertAck(final int count) {
+ verify(client,
times(count)).ackTaskExecutorLifecycleEvent(eq(task),
+
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+ }
+}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessPersistenceRecoveryTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessPersistenceRecoveryTest.java
new file mode 100644
index 0000000000..dc368d9a63
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessPersistenceRecoveryTest.java
@@ -0,0 +1,564 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+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.DataType;
+import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
+import org.apache.dolphinscheduler.server.master.engine.AbstractLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventHandler;
+import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
+import org.apache.dolphinscheduler.server.master.engine.ITaskGroupCoordinator;
+import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBus;
+import
org.apache.dolphinscheduler.server.master.engine.WorkflowEventBusFireWorker;
+import
org.apache.dolphinscheduler.server.master.engine.task.client.PhysicalTaskExecutorClientDelegator;
+import
org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient;
+import
org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler.TaskSuccessLifecycleEventHandler;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.WorkflowLifecycleEventType;
+import
org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent;
+import
org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
+import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager;
+import
org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+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;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import com.google.common.collect.ImmutableMap;
+
+@ExtendWith(MockitoExtension.class)
+class TaskSuccessPersistenceRecoveryTest {
+
+ @Mock
+ private TaskInstanceDao taskInstanceDao;
+
+ @Mock
+ private ITaskGroupCoordinator taskGroupCoordinator;
+
+ @Mock
+ private TaskExecutorClient taskExecutorClient;
+
+ @Mock
+ private WorkflowAlertManager workflowAlertManager;
+
+ @Mock
+ private ITaskExecution taskExecution;
+
+ @Mock
+ private IWorkflowExecution workflowExecution;
+
+ private WorkflowEventBus workflowEventBus;
+
+ private final AtomicBoolean failPersistence = new AtomicBoolean();
+
+ private final AtomicInteger persistenceAttempts = new AtomicInteger();
+
+ private final PersistentTaskSnapshot persistentTaskSnapshot = new
PersistentTaskSnapshot();
+
+ private TaskInstance taskInstance;
+
+ private TaskSuccessLifecycleEvent successEvent;
+
+ private TestableTaskSuccessLifecycleEventHandler handler;
+
+ private TaskStateActionFactory taskStateActionFactory;
+
+ @BeforeEach
+ void setUp() {
+ taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setName("terminal-state-recovery");
+ taskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION);
+ taskInstance.setVarPool("[]");
+ persistentTaskSnapshot.copyFrom(taskInstance);
+
+ final WorkflowInstance workflowInstance = new WorkflowInstance();
+ workflowInstance.setId(1);
+ workflowInstance.setName("terminal-state-recovery-workflow");
+ workflowInstance.setVarPool("[]");
+
+ workflowEventBus = spy(new WorkflowEventBus());
+
+ lenient().when(taskExecution.getId()).thenReturn(1);
+
lenient().when(taskExecution.getName()).thenReturn(taskInstance.getName());
+ when(taskExecution.getTaskInstance()).thenReturn(taskInstance);
+
lenient().when(taskExecution.getWorkflowEventBus()).thenReturn(workflowEventBus);
+
lenient().when(workflowExecution.getWorkflowInstance()).thenReturn(workflowInstance);
+
lenient().when(workflowExecution.getWorkflowEventBus()).thenReturn(workflowEventBus);
+
+
lenient().when(taskInstanceDao.updateById(any(TaskInstance.class))).thenAnswer(invocation
-> {
+ persistenceAttempts.incrementAndGet();
+ if (failPersistence.get()) {
+ throw new DataAccessResourceFailureException("synthetic
database write failure");
+ }
+ persistentTaskSnapshot.copyFrom(invocation.getArgument(0));
+ return true;
+ });
+
+ final TaskRunningStateAction runningStateAction = new
TaskRunningStateAction();
+ runningStateAction.taskGroupCoordinator = taskGroupCoordinator;
+ runningStateAction.taskInstanceDao = taskInstanceDao;
+
+ final TaskSuccessStateAction successStateAction = new
TaskSuccessStateAction();
+ successStateAction.taskGroupCoordinator = taskGroupCoordinator;
+ successStateAction.taskInstanceDao = taskInstanceDao;
+
+ taskStateActionFactory = new TaskStateActionFactory(
+ createStateActions(runningStateAction, successStateAction));
+
+ handler = new
TestableTaskSuccessLifecycleEventHandler(taskExecutorClient,
workflowAlertManager);
+ handler.setTaskStateActionFactory(taskStateActionFactory);
+
+ successEvent = TaskSuccessLifecycleEvent.builder()
+ .taskExecution(taskExecution)
+ .endTime(new Date(1_791_964_800_123L))
+ .varPool(Collections.emptyList())
+ .needAlert(false)
+ .build();
+ }
+
+ @Test
+ void shouldPersistAndAcknowledgeSuccessWithoutFailure() {
+ handler.handle(workflowExecution, successEvent);
+
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(persistentTaskSnapshot.endTime).isEqualTo(successEvent.getEndTime());
+ assertThat(persistenceAttempts).hasValue(1);
+ verify(workflowEventBus).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(1);
+ }
+
+ @Test
+ void shouldKeepInMemoryStateRetryableWhenFirstPersistenceFails() {
+ failPersistence.set(true);
+
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(taskInstance.getEndTime()).isNull();
+ assertThat(taskInstance.getVarPool()).isEqualTo("[]");
+ assertThat(persistenceAttempts).hasValue(1);
+ verifyAck(0);
+ }
+
+ @Test
+ void
shouldPersistAndAcknowledgeWhenSameSuccessEventIsReportedAfterRecovery() {
+ failPersistence.set(true);
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+
+ failPersistence.set(false);
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(2);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(persistentTaskSnapshot.endTime).isEqualTo(successEvent.getEndTime());
+ verify(workflowEventBus).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(1);
+ }
+
+ @Test
+ void
shouldNotAcknowledgeRepeatedSuccessReportsWhilePersistenceKeepsFailing() {
+ failPersistence.set(true);
+
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+
+ assertThat(persistenceAttempts).hasValue(2);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ verifyAck(0);
+ }
+
+ @Test
+ void shouldAcknowledgeDuplicateSuccessAfterItWasPersisted() {
+ handler.handle(workflowExecution, successEvent);
+ clearInvocations(taskExecutorClient, workflowEventBus);
+
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(1);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+ verify(workflowEventBus,
never()).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(1);
+ }
+
+ @Test
+ void
shouldRetryRecognizedConnectionFailureThroughWorkflowEventBusAndPersistAfterRecovery()
{
+ final IWorkflowExecuteContext workflowExecuteContext =
mock(IWorkflowExecuteContext.class);
+ final WorkflowInstance workflowInstance =
workflowExecution.getWorkflowInstance();
+
when(workflowExecuteContext.getWorkflowInstance()).thenReturn(workflowInstance);
+
when(workflowExecution.getWorkflowExecuteContext()).thenReturn(workflowExecuteContext);
+ when(workflowExecution.getId()).thenReturn(1);
+
when(workflowExecution.getName()).thenReturn("terminal-state-recovery-workflow");
+
+ final WorkflowEventBusFireWorker fireWorker = new
WorkflowEventBusFireWorker();
+ fireWorker.registerWorkflowEventBus(workflowExecution);
+ fireWorker.registerEventHandler(handler);
+ fireWorker.registerEventHandler(new
NoOpWorkflowTopologyEventHandler());
+ workflowEventBus.publish(successEvent);
+
+ failPersistence.set(true);
+ fireWorker.fireAllRegisteredEvent();
+
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(persistenceAttempts).hasValue(1);
+ assertThat(workflowEventBus.isEmpty()).isFalse();
+ verifyAck(0);
+
+ failPersistence.set(false);
+ fireWorker.fireAllRegisteredEvent();
+
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(persistentTaskSnapshot.endTime).isEqualTo(successEvent.getEndTime());
+ assertThat(persistenceAttempts).hasValue(2);
+ assertThat(workflowEventBus.isEmpty()).isTrue();
+ verifyAck(1);
+ }
+
+ @Test
+ void
shouldPreserveAndMergeOutputVariablesAcrossPersistenceFailureAndRetransmission()
{
+ final String originalTaskVarPool = serializeProperties(
+ property("existing", "old"),
+ property("result", "before"));
+ final String originalWorkflowVarPool =
serializeProperties(property("workflow", "kept"));
+ taskInstance.setVarPool(originalTaskVarPool);
+ persistentTaskSnapshot.copyFrom(taskInstance);
+
workflowExecution.getWorkflowInstance().setVarPool(originalWorkflowVarPool);
+ successEvent = successEventWithVariables(
+ property("result", "after"),
+ property("output", "new"));
+
+ failPersistence.set(true);
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(taskInstance.getEndTime()).isNull();
+ assertThat(taskInstance.getVarPool()).isEqualTo(originalTaskVarPool);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+
assertThat(persistentTaskSnapshot.varPool).isEqualTo(originalTaskVarPool);
+
assertThat(workflowExecution.getWorkflowInstance().getVarPool()).isEqualTo(originalWorkflowVarPool);
+ verify(workflowEventBus,
never()).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(0);
+
+ failPersistence.set(false);
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(2);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getEndTime()).isEqualTo(successEvent.getEndTime());
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of(
+ "existing", "old",
+ "result", "after",
+ "output", "new"));
+ assertVarPool(persistentTaskSnapshot.varPool, ImmutableMap.of(
+ "existing", "old",
+ "result", "after",
+ "output", "new"));
+ assertVarPool(workflowExecution.getWorkflowInstance().getVarPool(),
ImmutableMap.of(
+ "workflow", "kept",
+ "existing", "old",
+ "result", "after",
+ "output", "new"));
+
verify(workflowEventBus).publish(any(WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent.class));
+ verifyAck(1);
+
+ clearInvocations(taskExecutorClient, workflowEventBus);
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(2);
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of(
+ "existing", "old",
+ "result", "after",
+ "output", "new"));
+ assertVarPool(workflowExecution.getWorkflowInstance().getVarPool(),
ImmutableMap.of(
+ "workflow", "kept",
+ "existing", "old",
+ "result", "after",
+ "output", "new"));
+ verify(workflowEventBus,
never()).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(1);
+ }
+
+ @Test
+ void shouldPersistThenPublishTopologyThenAcknowledge() {
+ taskInstance.setVarPool(serializeProperties(property("existing",
"old")));
+ persistentTaskSnapshot.copyFrom(taskInstance);
+ successEvent = successEventWithVariables(property("output", "new"));
+ final List<String> actions = new ArrayList<>();
+
+
when(taskInstanceDao.updateById(any(TaskInstance.class))).thenAnswer(invocation
-> {
+ persistenceAttempts.incrementAndGet();
+ final TaskInstance taskInstanceToUpdate =
invocation.getArgument(0);
+ assertThat(taskInstanceToUpdate).isNotSameAs(taskInstance);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(taskInstance.getEndTime()).isNull();
+ assertVarPool(taskInstance.getVarPool(),
ImmutableMap.of("existing", "old"));
+
assertThat(taskInstanceToUpdate.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstanceToUpdate.getEndTime()).isEqualTo(successEvent.getEndTime());
+ assertVarPool(taskInstanceToUpdate.getVarPool(),
ImmutableMap.of("existing", "old", "output", "new"));
+ actions.add("persist-start");
+ persistentTaskSnapshot.copyFrom(taskInstanceToUpdate);
+ actions.add("persisted");
+ return true;
+ });
+ doAnswer(invocation -> {
+ assertThat((Object) invocation.getArgument(0))
+
.isInstanceOf(WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent.class);
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ actions.add("topology");
+ return invocation.callRealMethod();
+ }).when(workflowEventBus).publish(any(AbstractLifecycleEvent.class));
+ doAnswer(invocation -> {
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getEndTime()).isEqualTo(successEvent.getEndTime());
+ assertVarPool(taskInstance.getVarPool(),
ImmutableMap.of("existing", "old", "output", "new"));
+ actions.add("ack");
+ return null;
+ }).when(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(actions).containsExactly("persist-start", "persisted",
"topology", "ack");
+ }
+
+ @Test
+ void
shouldKeepPersistedSuccessWhenRealAckClientFailsAndAcknowledgeRetransmission() {
+ taskInstance.setTaskType("SHELL");
+ taskInstance.setHost("worker.test:1234");
+ taskInstance.setVarPool(serializeProperties(property("existing",
"old")));
+ persistentTaskSnapshot.copyFrom(taskInstance);
+ successEvent = successEventWithVariables(property("output", "new"));
+
+ final PhysicalTaskExecutorClientDelegator delegator =
mock(PhysicalTaskExecutorClientDelegator.class);
+ final TaskExecutorClient realTaskExecutorClient = new
TaskExecutorClient();
+ ReflectionTestUtils.setField(realTaskExecutorClient,
"physicalTaskExecutorClientDelegator", delegator);
+ doThrow(new RuntimeException("synthetic ACK send failure"))
+ .doNothing()
+ .when(delegator)
+ .ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
+
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ handler = new
TestableTaskSuccessLifecycleEventHandler(realTaskExecutorClient,
workflowAlertManager);
+ handler.setTaskStateActionFactory(taskStateActionFactory);
+
+ handler.handle(workflowExecution, successEvent);
+
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of("existing",
"old", "output", "new"));
+ assertThat(persistenceAttempts).hasValue(1);
+
verify(workflowEventBus).publish(any(WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent.class));
+
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(1);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of("existing",
"old", "output", "new"));
+ verify(workflowEventBus, times(1))
+
.publish(any(WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent.class));
+ verify(delegator, times(2)).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ @Test
+ void shouldConvergeWhenFirstUpdateCommitsButThrowsToCaller() {
+ taskInstance.setVarPool(serializeProperties(property("existing",
"old")));
+ persistentTaskSnapshot.copyFrom(taskInstance);
+ successEvent = successEventWithVariables(property("output", "new"));
+ final AtomicInteger localAttempts = new AtomicInteger();
+
when(taskInstanceDao.updateById(any(TaskInstance.class))).thenAnswer(invocation
-> {
+ persistenceAttempts.incrementAndGet();
+ localAttempts.incrementAndGet();
+ persistentTaskSnapshot.copyFrom(invocation.getArgument(0));
+ if (localAttempts.get() == 1) {
+ throw new DataAccessResourceFailureException("synthetic
uncertain commit result");
+ }
+ return true;
+ });
+
+ assertThatThrownBy(() -> handler.handle(workflowExecution,
successEvent))
+ .isInstanceOf(DataAccessResourceFailureException.class);
+
+
assertThat(persistentTaskSnapshot.state).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(taskInstance.getEndTime()).isNull();
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of("existing",
"old"));
+ verify(workflowEventBus,
never()).publish(any(AbstractLifecycleEvent.class));
+ verifyAck(0);
+
+ handler.handle(workflowExecution, successEvent);
+
+ assertThat(persistenceAttempts).hasValue(2);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getEndTime()).isEqualTo(successEvent.getEndTime());
+ assertVarPool(taskInstance.getVarPool(), ImmutableMap.of("existing",
"old", "output", "new"));
+ assertVarPool(persistentTaskSnapshot.varPool,
ImmutableMap.of("existing", "old", "output", "new"));
+
verify(workflowEventBus).publish(any(WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent.class));
+ verifyAck(1);
+ }
+
+ private TaskSuccessLifecycleEvent successEventWithVariables(final
Property... properties) {
+ return TaskSuccessLifecycleEvent.builder()
+ .taskExecution(taskExecution)
+ .endTime(new Date(1_791_964_800_123L))
+ .varPool(Arrays.asList(properties))
+ .needAlert(false)
+ .build();
+ }
+
+ private static Property property(final String name, final String value) {
+ return new Property(name, Direct.OUT, DataType.VARCHAR, value);
+ }
+
+ private static String serializeProperties(final Property... properties) {
+ return VarPoolUtils.serializeVarPool(Arrays.asList(properties));
+ }
+
+ private static void assertVarPool(final String actualJson, final
Map<String, String> expectedValues) {
+ final Map<String, Property> propertiesByName =
VarPoolUtils.deserializeVarPool(actualJson)
+ .stream()
+ .collect(Collectors.toMap(Property::getProp,
Function.identity()));
+ assertThat(propertiesByName).containsOnlyKeys(expectedValues.keySet());
+ expectedValues.forEach((name, value) -> {
+ final Property property = propertiesByName.get(name);
+ assertThat(property.getValue()).as("value of %s",
name).isEqualTo(value);
+ assertThat(property.getDirect()).as("direct of %s",
name).isEqualTo(Direct.OUT);
+ assertThat(property.getType()).as("type of %s",
name).isEqualTo(DataType.VARCHAR);
+ });
+ }
+
+ private List<ITaskStateAction> createStateActions(final
TaskRunningStateAction runningStateAction,
+ final
TaskSuccessStateAction successStateAction) {
+ return Arrays.stream(TaskExecutionStatus.values())
+ .map(state -> {
+ if (state == TaskExecutionStatus.RUNNING_EXECUTION) {
+ return runningStateAction;
+ }
+ if (state == TaskExecutionStatus.SUCCESS) {
+ return successStateAction;
+ }
+ final ITaskStateAction stateAction =
mock(ITaskStateAction.class);
+ when(stateAction.matchState()).thenReturn(state);
+ return stateAction;
+ })
+ .collect(Collectors.toList());
+ }
+
+ private void verifyAck(final int expectedInvocations) {
+ if (expectedInvocations == 0) {
+ verify(taskExecutorClient, never()).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ return;
+ }
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ private static final class PersistentTaskSnapshot {
+
+ private TaskExecutionStatus state;
+
+ private Date endTime;
+
+ private String varPool;
+
+ private void copyFrom(final TaskInstance source) {
+ state = source.getState();
+ endTime = source.getEndTime() == null ? null : new
Date(source.getEndTime().getTime());
+ varPool = source.getVarPool();
+ }
+
+ }
+
+ private static final class NoOpWorkflowTopologyEventHandler
+ implements
+ ILifecycleEventHandler<AbstractLifecycleEvent> {
+
+ @Override
+ public void handle(final IWorkflowExecution workflowExecution, final
AbstractLifecycleEvent event) {
+ // The task success transition is the behavior under test;
draining its downstream event keeps the bus
+ // finite.
+ }
+
+ @Override
+ public ILifecycleEventType matchEventType() {
+ return
WorkflowLifecycleEventType.TOPOLOGY_LOGICAL_TRANSACTION_WITH_TASK_FINISH;
+ }
+ }
+
+ private static final class TestableTaskSuccessLifecycleEventHandler
extends TaskSuccessLifecycleEventHandler {
+
+ private TestableTaskSuccessLifecycleEventHandler(final
TaskExecutorClient taskExecutorClient,
+ final
WorkflowAlertManager workflowAlertManager) {
+ super(taskExecutorClient, workflowAlertManager);
+ }
+
+ private void setTaskStateActionFactory(final TaskStateActionFactory
taskStateActionFactory) {
+ this.taskStateActionFactory = taskStateActionFactory;
+ }
+ }
+}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskLifecyclePersistenceRecoveryTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskLifecyclePersistenceRecoveryTestCase.java
new file mode 100644
index 0000000000..3e71d5f36b
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskLifecyclePersistenceRecoveryTestCase.java
@@ -0,0 +1,134 @@
+/*
+ * 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.integration.cases;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
+import
org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import
org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
+import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
+import
org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.springframework.boot.test.mock.mockito.SpyBean;
+import org.springframework.dao.DataAccessResourceFailureException;
+
+public class TaskLifecyclePersistenceRecoveryTestCase extends
AbstractMasterIntegrationTestCase {
+
+ @SpyBean
+ private TaskInstanceDao taskInstanceDao;
+
+ @ParameterizedTest
+ @EnumSource(value = TaskExecutionStatus.class, names = {"DISPATCH",
"RUNNING_EXECUTION"})
+ void shouldCompleteWorkflowAfterNonTerminalUpdateRecovers(final
TaskExecutionStatus state) {
+ final FailureProbe probe = failFirstUpdate(state);
+ final Integer workflowId =
start("/it/start/workflow_with_two_serial_fake_task_success.yaml");
+ awaitWorkflow(workflowId, WorkflowExecutionStatus.SUCCESS,
TaskExecutionStatus.SUCCESS, 2);
+ assertThat(probe.attempts).hasValue(2);
+ assertThat(probe.beforeFailure.get().getState()).isNotEqualTo(state);
+ }
+
+ @Test
+ void shouldFinishStoppingWorkflowAfterKillUpdateRecovers() {
+ final FailureProbe probe = failFirstUpdate(TaskExecutionStatus.KILL);
+ final Integer workflowId =
start("/it/stop/workflow_with_one_fake_task_failed.yaml");
+
await().pollInterval(Duration.ofMillis(100)).atMost(Duration.ofSeconds(30))
+ .untilAsserted(() ->
assertThat(repository.queryTaskInstance(workflowId)).satisfiesExactly(
+ task ->
assertThat(task.getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION)));
+
assertThat(workflowOperator.stopWorkflowInstance(workflowId).isSuccess()).isTrue();
+ awaitWorkflow(workflowId, WorkflowExecutionStatus.STOP,
TaskExecutionStatus.KILL, 1);
+ assertThat(probe.attempts).hasValue(2);
+
assertThat(probe.beforeFailure.get().getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ }
+
+ @Test
+ void shouldFinishFailedWorkflowAfterFailureUpdateRecovers() {
+ final FailureProbe probe =
failFirstUpdate(TaskExecutionStatus.FAILURE);
+ final Integer workflowId =
start("/it/start/workflow_with_one_fake_task_failed.yaml");
+ awaitWorkflow(workflowId, WorkflowExecutionStatus.FAILURE,
TaskExecutionStatus.FAILURE, 1);
+ assertThat(probe.attempts).hasValue(2);
+
assertThat(probe.beforeFailure.get().getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ }
+
+ @Test
+ void shouldFinishTasksSharingAGroupAfterSuccessUpdateRecovers() {
+ final FailureProbe probe =
failFirstUpdate(TaskExecutionStatus.SUCCESS);
+ final Integer workflowId =
start("/it/start/workflow_with_fake_tasks_using_task_group.yaml");
+ // The task group coordinator waits one minute before its first
scheduling round.
+ awaitWorkflow(workflowId, WorkflowExecutionStatus.SUCCESS,
TaskExecutionStatus.SUCCESS, 2,
+ Duration.ofMinutes(2));
+ assertThat(probe.attempts).hasValue(2);
+
assertThat(probe.beforeFailure.get().getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ }
+
+ private Integer start(final String resource) {
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(resource);
+ return
workflowOperator.manualTriggerWorkflow(WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(context.getOneWorkflow())
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build());
+ }
+
+ private void awaitWorkflow(final Integer workflowId, final
WorkflowExecutionStatus workflowState,
+ final TaskExecutionStatus taskState, final int
taskCount) {
+ awaitWorkflow(workflowId, workflowState, taskState, taskCount,
Duration.ofSeconds(45));
+ }
+
+ private void awaitWorkflow(final Integer workflowId, final
WorkflowExecutionStatus workflowState,
+ final TaskExecutionStatus taskState, final int
taskCount, final Duration timeout) {
+ await().atMost(timeout).untilAsserted(() -> {
+
assertThat(repository.queryWorkflowInstance(workflowId).getState()).isEqualTo(workflowState);
+
assertThat(repository.queryTaskInstance(workflowId)).hasSize(taskCount)
+ .allSatisfy(task ->
assertThat(task.getState()).isEqualTo(taskState));
+ masterContainer.assertAllResourceReleased();
+ });
+ }
+
+ private FailureProbe failFirstUpdate(final TaskExecutionStatus state) {
+ final FailureProbe probe = new FailureProbe();
+ doAnswer(invocation -> {
+ final TaskInstance candidate = invocation.getArgument(0);
+ if ("A".equals(candidate.getName()) && candidate.getState() ==
state
+ && probe.attempts.incrementAndGet() == 1) {
+
probe.beforeFailure.set(taskInstanceDao.queryById(candidate.getId()));
+ throw new DataAccessResourceFailureException("synthetic first
lifecycle update failure");
+ }
+ return invocation.callRealMethod();
+ }).when(taskInstanceDao).updateById(any(TaskInstance.class));
+ return probe;
+ }
+
+ private static final class FailureProbe {
+
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final AtomicReference<TaskInstance> beforeFailure = new
AtomicReference<>();
+ }
+}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskSuccessPersistenceRecoveryTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskSuccessPersistenceRecoveryTestCase.java
new file mode 100644
index 0000000000..fb514fb913
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/TaskSuccessPersistenceRecoveryTestCase.java
@@ -0,0 +1,93 @@
+/*
+ * 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.integration.cases;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
+import
org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import
org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
+import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
+import
org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.mock.mockito.SpyBean;
+import org.springframework.dao.DataAccessResourceFailureException;
+
+public class TaskSuccessPersistenceRecoveryTestCase extends
AbstractMasterIntegrationTestCase {
+
+ @SpyBean
+ private TaskInstanceDao taskInstanceDao;
+
+ @Test
+ public void testSuccessPersistenceFailureRetriesAndCompletesWorkflow() {
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(
+ "/it/start/workflow_with_two_serial_fake_task_success.yaml");
+ final AtomicInteger successUpdateAttempts = new AtomicInteger();
+ final AtomicReference<TaskInstance> persistedTaskBeforeFailure = new
AtomicReference<>();
+
+ // Fail only A's first success update. All other calls use the real
DAO and the isolated H2 database.
+ doAnswer(invocation -> {
+ final TaskInstance taskInstance = invocation.getArgument(0);
+ if ("A".equals(taskInstance.getName()) && taskInstance.getState()
== TaskExecutionStatus.SUCCESS
+ && successUpdateAttempts.incrementAndGet() == 1) {
+
persistedTaskBeforeFailure.set(taskInstanceDao.queryById(taskInstance.getId()));
+ throw new DataAccessResourceFailureException("synthetic first
success update failure");
+ }
+ return invocation.callRealMethod();
+ }).when(taskInstanceDao).updateById(any(TaskInstance.class));
+
+ final Integer workflowInstanceId =
workflowOperator.manualTriggerWorkflow(
+ WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(context.getOneWorkflow())
+ .runWorkflowCommandParam(new RunWorkflowCommandParam())
+ .build());
+
+ await()
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(() -> {
+
assertThat(repository.queryWorkflowInstance(workflowInstanceId).getState())
+ .isEqualTo(WorkflowExecutionStatus.SUCCESS);
+
assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .hasSize(2)
+ .extracting(TaskInstance::getName)
+ .containsExactly("A", "B");
+
assertThat(repository.queryTaskInstance(workflowInstanceId))
+ .allSatisfy(taskInstance -> {
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+
assertThat(taskInstance.getEndTime()).isNotNull();
+ });
+ });
+
+ assertThat(successUpdateAttempts).hasValue(2);
+ assertThat(persistedTaskBeforeFailure.get()).isNotNull();
+
assertThat(persistedTaskBeforeFailure.get().getState()).isEqualTo(TaskExecutionStatus.RUNNING_EXECUTION);
+ assertThat(persistedTaskBeforeFailure.get().getEndTime()).isNull();
+ masterContainer.assertAllResourceReleased();
+ }
+}
