add TaskFatalLifecycleEvent and handler

Author: 苏义超

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/client/PhysicalTaskExecutorClientDelegator.java

@@ -66,9 +66,13 @@ public void dispatch(final ITaskExecutionRunnable taskExecutionRunnable) throws
         final TaskExecutionContext taskExecutionContext = taskExecutionRunnable.getTaskExecutionContext();
         final String taskName = taskExecutionContext.getTaskName();
         final String workerGroup = taskExecutionContext.getWorkerGroup();
+
+        // workerGroup not exist
         if (!clusterManager.getWorkerClusters().containsWorkerGroup(workerGroup)) {
             throw new WorkerGroupNotFoundException(workerGroup);
         }
+
+        // select an available worker from the worker group; throws NoAvailableWorkerException if none is available.
         final String physicalTaskExecutorAddress = workerLoadBalancer
                 .select(workerGroup)
                 .map(Host::of)

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/TaskLifecycleEventType.java

@@ -29,6 +29,10 @@ public enum TaskLifecycleEventType implements ILifecycleEventType {
      * Dispatch the task instance to target.
      */
     DISPATCH,
+    /**
+     * Task instance encounters catastrophic failure(such as initialization failure), it will enter a failed state.
+     */
+    FATAL,
     /**
      * The task instance is dispatched to the target executor server.
      */

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

@@ -36,6 +36,7 @@
 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;
@@ -99,6 +100,38 @@ protected void releaseTaskInstanceResourcesIfNeeded(final ITaskExecutionRunnable
         }
     }
 
+    @Override
+    public void onFatalEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+                             final ITaskExecutionRunnable taskExecutionRunnable,
+                             final TaskFatalLifecycleEvent taskFatalEvent) {
+        releaseTaskInstanceResourcesIfNeeded(taskExecutionRunnable);
+        persistentTaskInstanceFatalEventToDB(taskExecutionRunnable, taskFatalEvent);
+
+        if (taskExecutionRunnable.isTaskInstanceCanRetry()) {
+            taskExecutionRunnable.getWorkflowEventBus().publish(TaskRetryLifecycleEvent.of(taskExecutionRunnable));
+            return;
+        }
+
+        // If all successors are condition tasks, then the task will not be marked as failure.
+        // And the DAG will continue to execute.
+        final IWorkflowExecutionGraph workflowExecutionGraph = taskExecutionRunnable.getWorkflowExecutionGraph();
+        if (workflowExecutionGraph.isAllSuccessorsAreConditionTask(taskExecutionRunnable)) {
+            mergeTaskVarPoolToWorkflow(workflowExecutionRunnable, taskExecutionRunnable);
+            publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+            return;
+        }
+        taskExecutionRunnable.getWorkflowExecutionGraph().markTaskExecutionRunnableChainFailure(taskExecutionRunnable);
+        publishWorkflowInstanceTopologyLogicalTransitionEvent(taskExecutionRunnable);
+    }
+
+    private void persistentTaskInstanceFatalEventToDB(final ITaskExecutionRunnable taskExecutionRunnable,
+                                                      final TaskFatalLifecycleEvent taskFatalEvent) {
+        final TaskInstance taskInstance = taskExecutionRunnable.getTaskInstance();
+        taskInstance.setState(TaskExecutionStatus.FAILURE);
+        taskInstance.setEndTime(taskFatalEvent.getEndTime());
+        taskInstanceDao.updateById(taskInstance);
+    }
+
     @Override
     public void onDispatchedEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
                                   final ITaskExecutionRunnable taskExecutionRunnable,

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

@@ -22,6 +22,7 @@
 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.TaskFailoverLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFatalLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskKillLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskKilledLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskPauseLifecycleEvent;
@@ -91,6 +92,14 @@ void onDispatchEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
                          final ITaskExecutionRunnable taskExecutionRunnable,
                          final TaskDispatchLifecycleEvent taskDispatchEvent);
 
+    /**
+     * Perform the necessary actions when the task in a certain state receive a {@link TaskFatalLifecycleEvent}.
+     * <p> This method is called when the task encounters catastrophic failure (e.g., initialization failure).
+     */
+    void onFatalEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
+                      final ITaskExecutionRunnable taskExecutionRunnable,
+                      final TaskFatalLifecycleEvent taskFatalEvent);
+
     /**
      * Perform the necessary actions when the task in a certain state receive a {@link TaskDispatchedLifecycleEvent}.
      * <p> This method is called when the task has been dispatched to executor.