ExecutorManager interface add generic type (#2012)

* 1, master persistent task
2. extract  master and worker communication model

* 1, master persistent task
2. extract  master and worker communication model

* 1, master persistent task
2. extract  master and worker communication model

* add license

* modify javadoc error

* TaskExecutionContext create modify

* buildAckCommand taskInstanceId not set modify

* java doc error modify

* add comment

* ExecutorManager interface add generic type

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutorDispatcher.java

@@ -50,7 +50,7 @@ public class ExecutorDispatcher implements InitializingBean {
     @Autowired
     private RoundRobinHostManager hostManager;
 
-    private final ConcurrentHashMap<ExecutorType, ExecutorManager> executorManagers;
+    private final ConcurrentHashMap<ExecutorType, ExecutorManager<Boolean>> executorManagers;
 
     public ExecutorDispatcher(){
         this.executorManagers = new ConcurrentHashMap<>();
@@ -61,11 +61,11 @@ public class ExecutorDispatcher implements InitializingBean {
      * @param context context
      * @throws ExecuteException
      */
-    public void dispatch(final ExecutionContext context) throws ExecuteException {
+    public Boolean dispatch(final ExecutionContext context) throws ExecuteException {
         /**
          * get executor manager
          */
-        ExecutorManager executorManager = this.executorManagers.get(context.getExecutorType());
+        ExecutorManager<Boolean> executorManager = this.executorManagers.get(context.getExecutorType());
         if(executorManager == null){
             throw new ExecuteException("no ExecutorManager for type : " + context.getExecutorType());
         }
@@ -83,7 +83,7 @@ public class ExecutorDispatcher implements InitializingBean {
             /**
              * task execute
              */
-            executorManager.execute(context);
+            return executorManager.execute(context);
         } finally {
             executorManager.afterExecute(context);
         }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/AbstractExecutorManager.java

@@ -23,7 +23,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteExce
 /**
  *  abstract executor manager
  */
-public abstract class AbstractExecutorManager implements ExecutorManager{
+public abstract class AbstractExecutorManager<T> implements ExecutorManager<T>{
 
     /**
      * before execute , add time monitor ， timeout

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/ExecutorManager.java

@@ -23,7 +23,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteExce
 /**
  *  executor manager
  */
-public interface ExecutorManager {
+public interface ExecutorManager<T> {
 
     /**
      * before execute
@@ -37,7 +37,7 @@ public interface ExecutorManager {
      * @param context context
      * @throws ExecuteException
      */
-    void execute(ExecutionContext context) throws ExecuteException;
+    T execute(ExecutionContext context) throws ExecuteException;
 
     /**
      * after execute

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java

@@ -46,7 +46,7 @@ import java.util.Set;
  *  netty executor manager
  */
 @Service
-public class NettyExecutorManager extends AbstractExecutorManager{
+public class NettyExecutorManager extends AbstractExecutorManager<Boolean>{
 
     private final Logger logger = LoggerFactory.getLogger(NettyExecutorManager.class);
 
@@ -64,6 +64,10 @@ public class NettyExecutorManager extends AbstractExecutorManager{
     public NettyExecutorManager(){
         final NettyClientConfig clientConfig = new NettyClientConfig();
         this.nettyRemotingClient = new NettyRemotingClient(clientConfig);
+        /**
+         * register EXECUTE_TASK_RESPONSE command type TaskResponseProcessor
+         * register EXECUTE_TASK_ACK command type TaskAckProcessor
+         */
         this.nettyRemotingClient.registerProcessor(CommandType.EXECUTE_TASK_RESPONSE, new TaskResponseProcessor());
         this.nettyRemotingClient.registerProcessor(CommandType.EXECUTE_TASK_ACK, new TaskAckProcessor());
     }
@@ -74,7 +78,7 @@ public class NettyExecutorManager extends AbstractExecutorManager{
      * @throws ExecuteException
      */
     @Override
-    public void execute(ExecutionContext context) throws ExecuteException {
+    public Boolean execute(ExecutionContext context) throws ExecuteException {
 
         /**
          *  all nodes
@@ -118,6 +122,8 @@ public class NettyExecutorManager extends AbstractExecutorManager{
                 }
             }
         }
+
+        return success;
     }
 
     /**

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

@@ -20,22 +20,10 @@ import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.utils.FileUtils;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.dao.AlertDao;
-import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.Tenant;
-import org.apache.dolphinscheduler.dao.utils.BeanContext;
-import org.apache.dolphinscheduler.remote.NettyRemotingClient;
-import org.apache.dolphinscheduler.remote.command.Command;
-import org.apache.dolphinscheduler.remote.command.ExecuteTaskAckCommand;
-import org.apache.dolphinscheduler.remote.command.ExecuteTaskRequestCommand;
 import org.apache.dolphinscheduler.remote.entity.TaskExecutionContext;
-import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
-import org.apache.dolphinscheduler.remote.exceptions.RemotingException;
-import org.apache.dolphinscheduler.remote.future.InvokeCallback;
-import org.apache.dolphinscheduler.remote.future.ResponseFuture;
-import org.apache.dolphinscheduler.remote.utils.Host;
-import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;
 import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.dispatch.ExecutorDispatcher;
@@ -48,7 +36,6 @@ import org.apache.dolphinscheduler.service.queue.ITaskQueue;
 import org.apache.dolphinscheduler.service.queue.TaskQueueFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 
 import java.util.concurrent.Callable;
 
@@ -138,14 +125,16 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
      * dispatch task to worker
      * @param taskInstance
      */
-    public void dispatch(TaskInstance taskInstance){
+    private Boolean dispatch(TaskInstance taskInstance){
         TaskExecutionContext context = getTaskExecutionContext(taskInstance);
         ExecutionContext executionContext = new ExecutionContext(context, ExecutorType.WORKER);
         try {
-            dispatcher.dispatch(executionContext);
+            return dispatcher.dispatch(executionContext);
         } catch (ExecuteException e) {
             logger.error("execute exception", e);
+            return false;
         }
+
     }
 
     /**
@@ -234,8 +223,7 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
                 }
                 if(submitDB && !submitQueue){
                     // submit task to queue
-                    dispatch(task);
+                    submitQueue = dispatch(task);
-                    submitQueue = true;
                 }
                 if(submitDB && submitQueue){
                     return task;