throw TaskExecutionContextCreateException if initializeTaskExecutionContext fail

Author: 苏义超

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java

@@ -131,9 +131,8 @@ private void doFireSingleWorkflowEventBus(final IWorkflowExecutionRunnable workf
                     ThreadUtils.sleep(5_000);
                     return;
                 }
-                if (ExceptionUtils.isIllegalArgumentException(ex)
-                        && lifecycleEvent instanceof AbstractTaskLifecycleEvent) {
-                    // If exception is IllegalArgumentException and the event is task-related
+                if (ExceptionUtils.isTaskExecutionContextCreateException(ex)) {
+                    // If task initializeTaskExecutionContext before dispatch is failed
                     // construct and publish a dedicated TaskFatalLifecycleEvent
                     // so that the event will be handled by TaskFatalLifecycleEventHandler
                     AbstractTaskLifecycleEvent taskLifecycleEvent = (AbstractTaskLifecycleEvent) lifecycleEvent;

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

@@ -36,6 +36,7 @@
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
+import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
 
 import java.util.concurrent.TimeUnit;
 
@@ -109,7 +110,14 @@ public void onDispatchEvent(final IWorkflowExecutionRunnable workflowExecutionRu
                     taskInstance.getDelayTime(),
                     remainTimeMills);
         }
-        taskExecutionRunnable.initializeTaskExecutionContext();
+
+        try {
+            taskExecutionRunnable.initializeTaskExecutionContext();
+        } catch (Exception ex) {
+            log.error("Failed to initialize task execution context, taskName: {}", taskInstance.getName(), ex);
+            throw new TaskExecutionContextCreateException(ex.getMessage());
+
+        }
         workerGroupDispatcherCoordinator.dispatchTask(taskExecutionRunnable, remainTimeMills);
     }
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/exception/TaskExecutionContextCreateException.java

@@ -17,7 +17,7 @@
 
 package org.apache.dolphinscheduler.server.master.exception;
 
-public class TaskExecutionContextCreateException extends MasterException {
+public class TaskExecutionContextCreateException extends RuntimeException {
 
     public TaskExecutionContextCreateException(String message) {
         super(message);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/ExceptionUtils.java

@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.server.master.utils;
 
+import org.apache.dolphinscheduler.server.master.exception.TaskExecutionContextCreateException;
+
 import org.springframework.dao.DataAccessResourceFailureException;
 
 public class ExceptionUtils {
@@ -25,7 +27,7 @@ public static boolean isDatabaseConnectedFailedException(Throwable e) {
         return e instanceof DataAccessResourceFailureException;
     }
 
-    public static boolean isIllegalArgumentException(Throwable e) {
-        return e instanceof IllegalArgumentException;
+    public static boolean isTaskExecutionContextCreateException(Throwable e) {
+        return e instanceof TaskExecutionContextCreateException;
     }
 }