add do with initializeFailTaskIds

Author: 苏义超

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java

@@ -46,13 +46,16 @@ public class WorkerGroupDispatcher extends BaseDaemonThread {
 
     private final Set<Integer> waitingDispatchTaskIds;
 
+    private final Set<Integer> initializeFailTaskIds;
+
     private final AtomicBoolean runningFlag = new AtomicBoolean(false);
 
     public WorkerGroupDispatcher(String workerGroupName, ITaskExecutorClient taskExecutorClient) {
         super("WorkerGroupTaskDispatcher-" + workerGroupName);
         this.taskExecutorClient = taskExecutorClient;
         this.workerGroupEventBus = new TaskDispatchableEventBus<>();
         this.waitingDispatchTaskIds = ConcurrentHashMap.newKeySet();
+        this.initializeFailTaskIds = ConcurrentHashMap.newKeySet();
         log.info("Initialize WorkerGroupDispatcher: {}", this.getName());
     }
 
@@ -139,4 +142,12 @@ public synchronized void close() {
     int queueSize() {
         return this.workerGroupEventBus.size();
     }
+
+    public void addInitializeFailTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        initializeFailTaskIds.add(taskExecutionRunnable.getId());
+    }
+
+    public boolean removeInitializeFailTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        return initializeFailTaskIds.remove(taskExecutionRunnable.getId());
+    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java

@@ -66,7 +66,15 @@ public void dispatchTask(final ITaskExecutionRunnable taskExecutionRunnable,
      */
     public boolean removeTask(ITaskExecutionRunnable taskExecutionRunnable) {
         final String workerGroup = taskExecutionRunnable.getTaskInstance().getWorkerGroup();
-        boolean removed = getOrCreateWorkerGroupDispatcher(workerGroup).removeTask(taskExecutionRunnable);
+
+        boolean removed = getOrCreateWorkerGroupDispatcher(workerGroup).removeInitializeFailTask(taskExecutionRunnable);
+        if (removed) {
+            log.info("Success removed Initialize Fail Task[id={}] from WorkerGroupDispatcher[name={}]",
+                    taskExecutionRunnable.getId(), workerGroup);
+            return removed;
+        }
+
+        removed = getOrCreateWorkerGroupDispatcher(workerGroup).removeTask(taskExecutionRunnable);
         if (removed) {
             log.info("Success removed Task[id={}] from WorkerGroupDispatcher[name={}]",
                     taskExecutionRunnable.getId(), workerGroup);
@@ -97,6 +105,13 @@ public void close() throws Exception {
         log.info("WorkerGroupDispatcherCoordinator closed...");
     }
 
+    public void addInitializeFailTask(ITaskExecutionRunnable taskExecutionRunnable) {
+        final String workerGroup = taskExecutionRunnable.getTaskInstance().getWorkerGroup();
+        getOrCreateWorkerGroupDispatcher(workerGroup).addInitializeFailTask(taskExecutionRunnable);
+        log.info("add Initialize Fail Task[id={}] to WorkerGroupDispatcher[name={}]", taskExecutionRunnable.getId(),
+                workerGroup);
+    }
+
     private WorkerGroupDispatcher getOrCreateWorkerGroupDispatcher(String workerGroup) {
         return workerGroupDispatcherMap.computeIfAbsent(workerGroup, wg -> {
             WorkerGroupDispatcher workerGroupDispatcher = new WorkerGroupDispatcher(wg, taskExecutorClient);

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

@@ -109,7 +109,14 @@ public void onDispatchEvent(final IWorkflowExecutionRunnable workflowExecutionRu
                     taskInstance.getDelayTime(),
                     remainTimeMills);
         }
-        taskExecutionRunnable.initializeTaskExecutionContext();
+
+        try {
+            taskExecutionRunnable.initializeTaskExecutionContext();
+        } catch (Exception ex) {
+            log.error("Current taskInstance: {} initializeTaskExecutionContext error", taskInstance.getName(), ex);
+            workerGroupDispatcherCoordinator.addInitializeFailTask(taskExecutionRunnable);
+            return;
+        }
         workerGroupDispatcherCoordinator.dispatchTask(taskExecutionRunnable, remainTimeMills);
     }