[Fix-7146][server]Fix the task processor thread is not safe. (#7488)

* Fix the task processor thread is not safe.

* fix code style

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java

@@ -64,12 +64,6 @@ public class MasterSchedulerService extends Thread {
     @Autowired
     private ProcessService processService;
 
-    /**
-     * task processor factory
-     */
-    @Autowired
-    private TaskProcessorFactory taskProcessorFactory;
-
     /**
      * master config
      */
@@ -176,8 +170,7 @@ public class MasterSchedulerService extends Thread {
                     , nettyExecutorManager
                     , processAlertManager
                     , masterConfig
-                    , stateWheelExecuteThread
+                    , stateWheelExecuteThread);
-                    , taskProcessorFactory);
 
             this.processInstanceExecCacheManager.cache(processInstance.getId(), workflowExecuteThread);
             if (processInstance.getTimeout() > 0) {

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java

@@ -133,11 +133,6 @@ public class WorkflowExecuteThread {
      */
     private ProcessDefinition processDefinition;
 
-    /**
-     * task processor
-     */
-    private TaskProcessorFactory taskProcessorFactory;
-
     /**
      * the object of DAG
      */
@@ -227,22 +222,19 @@ public class WorkflowExecuteThread {
      * @param processAlertManager processAlertManager
      * @param masterConfig masterConfig
      * @param stateWheelExecuteThread stateWheelExecuteThread
-     * @param taskProcessorFactory taskProcessorFactory
      */
     public WorkflowExecuteThread(ProcessInstance processInstance
             , ProcessService processService
             , NettyExecutorManager nettyExecutorManager
             , ProcessAlertManager processAlertManager
             , MasterConfig masterConfig
-            , StateWheelExecuteThread stateWheelExecuteThread
+            , StateWheelExecuteThread stateWheelExecuteThread) {
-            , TaskProcessorFactory taskProcessorFactory) {
         this.processService = processService;
         this.processInstance = processInstance;
         this.masterConfig = masterConfig;
         this.nettyExecutorManager = nettyExecutorManager;
         this.processAlertManager = processAlertManager;
         this.stateWheelExecuteThread = stateWheelExecuteThread;
-        this.taskProcessorFactory = taskProcessorFactory;
     }
 
     /**
@@ -805,7 +797,7 @@ public class WorkflowExecuteThread {
      */
     private TaskInstance submitTaskExec(TaskInstance taskInstance) {
         try {
-            ITaskProcessor taskProcessor = taskProcessorFactory.getTaskProcessor(taskInstance.getTaskType());
+            ITaskProcessor taskProcessor = TaskProcessorFactory.getTaskProcessor(taskInstance.getTaskType());
             if (taskInstance.getState() == ExecutionStatus.RUNNING_EXECUTION
                     && taskProcessor.getType().equalsIgnoreCase(Constants.COMMON_TASK_TYPE)) {
                 notifyProcessHostUpdate(taskInstance);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java

@@ -40,6 +40,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.Tenant;
 import org.apache.dolphinscheduler.dao.entity.UdfFunc;
 import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
 import org.apache.dolphinscheduler.spi.enums.ResourceType;
@@ -61,7 +62,6 @@ import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 
 import com.google.common.base.Enums;
 import com.google.common.base.Strings;
@@ -80,8 +80,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
 
     protected ProcessInstance processInstance;
 
-    @Autowired
+    protected ProcessService processService = SpringApplicationContext.getBean(ProcessService.class);;
-    protected ProcessService processService;
 
     /**
      * pause task, common tasks donot need this.

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessFactory.java

@@ -0,0 +1,36 @@
+/*
+ * 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.runner.task;
+
+import org.apache.dolphinscheduler.common.Constants;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(ITaskProcessFactory.class)
+public class CommonTaskProcessFactory implements ITaskProcessFactory {
+    @Override
+    public String type() {
+        return Constants.COMMON_TASK_TYPE;
+
+    }
+
+    @Override
+    public ITaskProcessor create() {
+        return new CommonTaskProcessor();
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java

@@ -38,20 +38,14 @@ import org.apache.commons.lang.StringUtils;
 
 import java.util.Date;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
 /**
  * common task processor
  */
-@Service
 public class CommonTaskProcessor extends BaseTaskProcessor {
 
-    @Autowired
     private TaskPriorityQueue taskUpdateQueue;
 
-    @Autowired
+    private NettyExecutorManager nettyExecutorManager = SpringApplicationContext.getBean(NettyExecutorManager.class);
-    NettyExecutorManager nettyExecutorManager;
 
     @Override
     public boolean submit(TaskInstance task, ProcessInstance processInstance, int maxRetryTimes, int commitInterval, boolean isTaskLogger) {

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/ConditionTaskProcessFactory.java

@@ -0,0 +1,35 @@
+/*
+ * 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.runner.task;
+
+import org.apache.dolphinscheduler.common.enums.TaskType;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(ITaskProcessFactory.class)
+public class ConditionTaskProcessFactory implements ITaskProcessFactory {
+    @Override
+    public String type() {
+        return TaskType.CONDITIONS.getDesc();
+    }
+
+    @Override
+    public ITaskProcessor create() {
+        return new ConditionTaskProcessor();
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/ConditionTaskProcessor.java

@@ -32,6 +32,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.utils.LogUtils;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -39,13 +40,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
 /**
  * condition task processor
  */
-@Service
 public class ConditionTaskProcessor extends BaseTaskProcessor {
 
     /**
@@ -65,8 +62,7 @@ public class ConditionTaskProcessor extends BaseTaskProcessor {
      */
     private Map<Long, ExecutionStatus> completeTaskList = new ConcurrentHashMap<>();
 
-    @Autowired
+    private MasterConfig masterConfig = SpringApplicationContext.getBean(MasterConfig.class);;
-    private MasterConfig masterConfig;
 
     private TaskDefinition taskDefinition;
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/DependentTaskProcessFactory.java

@@ -0,0 +1,36 @@
+/*
+ * 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.runner.task;
+
+import org.apache.dolphinscheduler.common.enums.TaskType;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(ITaskProcessFactory.class)
+public class DependentTaskProcessFactory implements ITaskProcessFactory {
+
+    @Override
+    public String type() {
+        return TaskType.DEPENDENT.getDesc();
+    }
+
+    @Override
+    public ITaskProcessor create() {
+        return new DependentTaskProcessor();
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/DependentTaskProcessor.java

@@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.utils.DependentExecute;
 import org.apache.dolphinscheduler.server.utils.LogUtils;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -40,15 +41,11 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
 import com.fasterxml.jackson.annotation.JsonFormat;
 
 /**
  * dependent task processor
  */
-@Service
 public class DependentTaskProcessor extends BaseTaskProcessor {
 
     private DependentParameters dependentParameters;
@@ -75,8 +72,7 @@ public class DependentTaskProcessor extends BaseTaskProcessor {
     ProcessInstance processInstance;
     TaskDefinition taskDefinition;
 
-    @Autowired
+    private MasterConfig masterConfig = SpringApplicationContext.getBean(MasterConfig.class);;
-    private MasterConfig masterConfig;
 
     boolean allDependentItemFinished;
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/ITaskProcessFactory.java

@@ -0,0 +1,25 @@
+/*
+ * 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.runner.task;
+
+public interface ITaskProcessFactory {
+
+    String type();
+
+    ITaskProcessor create();
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SubTaskProcessFactory.java

@@ -0,0 +1,35 @@
+/*
+ * 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.runner.task;
+
+import org.apache.dolphinscheduler.common.enums.TaskType;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(ITaskProcessFactory.class)
+public class SubTaskProcessFactory implements ITaskProcessFactory {
+    @Override
+    public String type() {
+        return TaskType.SUB_PROCESS.getDesc();
+    }
+
+    @Override
+    public ITaskProcessor create() {
+        return new SubTaskProcessor();
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SubTaskProcessor.java

@@ -25,18 +25,15 @@ import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.remote.command.StateEventChangeCommand;
 import org.apache.dolphinscheduler.remote.processor.StateEventCallbackService;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.util.Date;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
 /**
- *
+ * subtask processor
  */
-@Service
 public class SubTaskProcessor extends BaseTaskProcessor {
 
     private ProcessInstance processInstance;
@@ -49,8 +46,7 @@ public class SubTaskProcessor extends BaseTaskProcessor {
      */
     private final Lock runLock = new ReentrantLock();
 
-    @Autowired
+    private StateEventCallbackService stateEventCallbackService = SpringApplicationContext.getBean(StateEventCallbackService.class);;
-    private StateEventCallbackService stateEventCallbackService;
 
     @Override
     public boolean submit(TaskInstance task, ProcessInstance processInstance, int masterTaskCommitRetryTimes, int masterTaskCommitInterval, boolean isTaskLogger) {

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SwitchTaskProcessFactory.java

@@ -0,0 +1,36 @@
+/*
+ * 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.runner.task;
+
+import org.apache.dolphinscheduler.common.enums.TaskType;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(ITaskProcessFactory.class)
+public class SwitchTaskProcessFactory implements ITaskProcessFactory {
+
+    @Override
+    public String type() {
+        return TaskType.SWITCH.getDesc();
+    }
+
+    @Override
+    public ITaskProcessor create() {
+        return new SwitchTaskProcessor();
+    }
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SwitchTaskProcessor.java

@@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.utils.LogUtils;
 import org.apache.dolphinscheduler.server.utils.SwitchTaskUtils;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang.StringUtils;
@@ -46,7 +47,9 @@ import java.util.stream.Collectors;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-@Service
+/**
+ * switch task processor
+ */
 public class SwitchTaskProcessor extends BaseTaskProcessor {
 
     protected final String rgex = "['\"]*\\$\\{(.*?)\\}['\"]*";
@@ -56,8 +59,7 @@ public class SwitchTaskProcessor extends BaseTaskProcessor {
     private ProcessInstance processInstance;
     TaskDefinition taskDefinition;
 
-    @Autowired
+    private MasterConfig masterConfig = SpringApplicationContext.getBean(MasterConfig.class);;
-    private MasterConfig masterConfig;
 
     /**
      * switch result

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/TaskProcessorFactory.java

@@ -21,39 +21,35 @@ import static org.apache.dolphinscheduler.common.Constants.COMMON_TASK_TYPE;
 
 import org.apache.commons.lang3.StringUtils;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.Function;
+import java.util.ServiceLoader;
-import java.util.stream.Collectors;
+import java.util.concurrent.ConcurrentHashMap;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
 
 /**
  * the factory to create task processor
  */
-@Service
 public class TaskProcessorFactory {
 
-    private static final String DEFAULT_PROCESSOR = COMMON_TASK_TYPE;
+    public static final Map<String, ITaskProcessFactory> PROCESS_FACTORY_MAP = new ConcurrentHashMap<>();
 
-    private Map<String, ITaskProcessor> taskProcessorMap;
+    private static final String DEFAULT_PROCESSOR = COMMON_TASK_TYPE;
 
-    @Autowired
+    static {
-    public TaskProcessorFactory(List<ITaskProcessor> taskProcessors) {
+        for (ITaskProcessFactory iTaskProcessor : ServiceLoader.load(ITaskProcessFactory.class)) {
-        taskProcessorMap = taskProcessors.stream().collect(Collectors.toMap(ITaskProcessor::getType, Function.identity(), (v1, v2) -> v2));
+            PROCESS_FACTORY_MAP.put(iTaskProcessor.type(), iTaskProcessor);
+        }
     }
 
-    public ITaskProcessor getTaskProcessor(String key) {
+    public static ITaskProcessor getTaskProcessor(String type) {
-        if (StringUtils.isEmpty(key)) {
+        if (StringUtils.isEmpty(type)) {
-            key = DEFAULT_PROCESSOR;
+            type = DEFAULT_PROCESSOR;
         }
-        ITaskProcessor taskProcessor = taskProcessorMap.get(key);
+        ITaskProcessFactory taskProcessFactory = PROCESS_FACTORY_MAP.get(type);
-        if (Objects.isNull(taskProcessor)) {
+        if (Objects.isNull(taskProcessFactory)) {
-            taskProcessor = taskProcessorMap.get(DEFAULT_PROCESSOR);
+            taskProcessFactory = PROCESS_FACTORY_MAP.get(DEFAULT_PROCESSOR);
         }
 
-        return taskProcessor;
+        return taskProcessFactory.create();
     }
 }

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/WorkflowExecuteThreadTest.java

@@ -111,7 +111,7 @@ public class WorkflowExecuteThreadTest {
         Mockito.when(processInstance.getProcessDefinition()).thenReturn(processDefinition);
 
         stateWheelExecuteThread = mock(StateWheelExecuteThread.class);
-        workflowExecuteThread = PowerMockito.spy(new WorkflowExecuteThread(processInstance, processService, null, null, config, stateWheelExecuteThread, taskProcessorFactory));
+        workflowExecuteThread = PowerMockito.spy(new WorkflowExecuteThread(processInstance, processService, null, null, config, stateWheelExecuteThread));
         // prepareProcess init dag
         Field dag = WorkflowExecuteThread.class.getDeclaredField("dag");
         dag.setAccessible(true);

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/task/TaskProcessorFactoryTest.java

@@ -26,14 +26,13 @@ import org.junit.Test;
 @Ignore
 public class TaskProcessorFactoryTest {
 
-    private TaskProcessorFactory taskProcessorFactory;
     @Test
     public void testFactory() {
 
         TaskInstance taskInstance = new TaskInstance();
         taskInstance.setTaskType("shell");
 
-        ITaskProcessor iTaskProcessor = taskProcessorFactory.getTaskProcessor(taskInstance.getTaskType());
+        ITaskProcessor iTaskProcessor = TaskProcessorFactory.getTaskProcessor(taskInstance.getTaskType());
 
         Assert.assertNotNull(iTaskProcessor);
     }