[Fix-18707][Master] Protect other task lifecycle updates on persistence failure

Author: ThreeFire1

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java

@@ -49,6 +49,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
+import java.util.function.Consumer;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -125,39 +126,43 @@ public void onFatalEvent(final IWorkflowExecution workflowExecution,
     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
     public void onDispatchedEvent(final IWorkflowExecution workflowExecution,
                                   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
     public void onRuntimeContextChangedEvent(final IWorkflowExecution workflowExecution,
                                              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
@@ -173,8 +178,7 @@ public void onPausedEvent(final IWorkflowExecution workflowExecution,
     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
@@ -190,9 +194,10 @@ public void onKilledEvent(final IWorkflowExecution workflowExecution,
     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());
+        });
 
     }
 
@@ -223,9 +228,10 @@ public void onFailedEvent(final IWorkflowExecution workflowExecution,
     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
@@ -247,24 +253,32 @@ protected void mergeTaskVarPoolToWorkflow(final IWorkflowExecution workflowExecu
         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();
         final List<Property> finalVarPool = VarPoolUtils.mergeVarPoolJsonString(taskInstance.getVarPool(),
                 JSONUtils.toJsonString(taskSuccessEvent.getVarPool()));
         final String serializedFinalVarPool = VarPoolUtils.serializeVarPool(finalVarPool);
 
-        // Keep the shared state retryable if persisting the success result throws.
-        final TaskInstance taskInstanceToUpdate = new TaskInstance();
-        BeanUtils.copyProperties(taskInstance, taskInstanceToUpdate);
-        taskInstanceToUpdate.setState(TaskExecutionStatus.SUCCESS);
-        taskInstanceToUpdate.setEndTime(taskSuccessEvent.getEndTime());
-        taskInstanceToUpdate.setVarPool(serializedFinalVarPool);
-        taskInstanceDao.updateById(taskInstanceToUpdate);
-
-        taskInstance.setState(TaskExecutionStatus.SUCCESS);
-        taskInstance.setEndTime(taskSuccessEvent.getEndTime());
-        taskInstance.setVarPool(serializedFinalVarPool);
+        persistTaskInstanceUpdate(taskInstance, instance -> {
+            instance.setState(TaskExecutionStatus.SUCCESS);
+            instance.setEndTime(taskSuccessEvent.getEndTime());
+            instance.setVarPool(serializedFinalVarPool);
+        });
     }
 
     /**

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java

@@ -19,7 +19,6 @@
 
 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 void onDispatchEvent(final IWorkflowExecution workflowExecution,
                 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(),