Refactor worker

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java

@@ -1 +1 @@
-/* * 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.remote.command;public enum CommandType {    /**     *  roll view log request     */    ROLL_VIEW_LOG_REQUEST,    /**     *  roll view log response     */    ROLL_VIEW_LOG_RESPONSE,    /**     * view whole log request     */    VIEW_WHOLE_LOG_REQUEST,    /**     * view whole log response     */    VIEW_WHOLE_LOG_RESPONSE,    /**     * get log bytes request     */    GET_LOG_BYTES_REQUEST,    /**     * get log bytes response     */    GET_LOG_BYTES_RESPONSE,    WORKER_REQUEST,    MASTER_RESPONSE,    /**     * execute task request     */    EXECUTE_TASK_REQUEST,    /**     * execute task ack     */    EXECUTE_TASK_ACK,    /**     * execute task response     */    EXECUTE_TASK_RESPONSE,    /**     *  ping     */    PING,    /**     *  pong     */    PONG;}
+/* * 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.remote.command;public enum CommandType {    /**     *  roll view log request     */    ROLL_VIEW_LOG_REQUEST,    /**     *  roll view log response     */    ROLL_VIEW_LOG_RESPONSE,    /**     * view whole log request     */    VIEW_WHOLE_LOG_REQUEST,    /**     * view whole log response     */    VIEW_WHOLE_LOG_RESPONSE,    /**     * get log bytes request     */    GET_LOG_BYTES_REQUEST,    /**     * get log bytes response     */    GET_LOG_BYTES_RESPONSE,    WORKER_REQUEST,    MASTER_RESPONSE,    /**     * execute task request     */    EXECUTE_TASK_REQUEST,    /**     * execute task ack     */    EXECUTE_TASK_ACK,    /**     * execute task response     */    EXECUTE_TASK_RESPONSE,    /**     * kill task     */    KILL_TASK_REQUEST,    /**     * kill task response     */    KILL_TASK_RESPONSE,    /**     *  ping     */    PING,    /**     *  pong     */    PONG;}

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/KillTaskRequestCommand.java

@@ -0,0 +1 @@
+/* * 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.remote.command;import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;import java.io.Serializable;/** *  kill task request command */public class KillTaskRequestCommand implements Serializable {    private int taskInstanceId;    private int processId;    private String host;    private String tenantCode;    private String logPath;    private String executePath;    public String getLogPath() {        return logPath;    }    public void setLogPath(String logPath) {        this.logPath = logPath;    }    public int getTaskInstanceId() {        return taskInstanceId;    }    public void setTaskInstanceId(int taskInstanceId) {        this.taskInstanceId = taskInstanceId;    }    public int getProcessId() {        return processId;    }    public void setProcessId(int processId) {        this.processId = processId;    }    public String getHost() {        return host;    }    public void setHost(String host) {        this.host = host;    }    public String getTenantCode() {        return tenantCode;    }    public void setTenantCode(String tenantCode) {        this.tenantCode = tenantCode;    }    public String getExecutePath() {        return executePath;    }    public void setExecutePath(String executePath) {        this.executePath = executePath;    }    /**     *  package request command     *     * @return command     */    public Command convert2Command(){        Command command = new Command();        command.setType(CommandType.KILL_TASK_REQUEST);        byte[] body = FastJsonSerializer.serialize(this);        command.setBody(body);        return command;    }}

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 ExecutorDispatcher(){
      * @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 void dispatch(final ExecutionContext context) throws ExecuteException {
             /**
              * 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 @@
 /**
  *  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 @@
 /**
  *  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 @@
  *  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 NettyExecutorManager(){
      * @throws ExecuteException
      */
     @Override
-    public void execute(ExecutionContext context) throws ExecuteException {
+    public Boolean execute(ExecutionContext context) throws ExecuteException {
 
         /**
          *  all nodes
@@ -118,6 +122,8 @@ public void execute(ExecutionContext context) throws ExecuteException {
                 }
             }
         }
+
+        return success;
     }
 
     /**
@@ -189,7 +195,7 @@ private Set<String> getAllNodes(ExecutionContext context){
                 break;
             case CLIENT:
                 break;
-             default:
+            default:
                 throw new IllegalArgumentException("invalid executor type : " + executorType);
 
         }

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

@@ -20,22 +20,10 @@
 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.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 void kill(){
      * 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 @@ protected TaskInstance submit(){
                 }
                 if(submitDB && !submitQueue){
                     // submit task to queue
-                    dispatch(task);
-                    submitQueue = true;
+                    submitQueue = dispatch(task);
                 }
                 if(submitDB && submitQueue){
                     return task;

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/WorkerServer.java

@@ -35,7 +35,8 @@
 import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.apache.dolphinscheduler.server.utils.ProcessUtils;
 import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
-import org.apache.dolphinscheduler.server.worker.processor.WorkerRequestProcessor;
+import org.apache.dolphinscheduler.server.worker.processor.TaskExecuteProcessor;
+import org.apache.dolphinscheduler.server.worker.processor.TaskKillProcessor;
 import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistry;
 import org.apache.dolphinscheduler.server.zk.ZKWorkerClient;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
@@ -168,7 +169,8 @@ public void run(){
         //init remoting server
         NettyServerConfig serverConfig = new NettyServerConfig();
         this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
-        this.nettyRemotingServer.registerProcessor(CommandType.EXECUTE_TASK_REQUEST, new WorkerRequestProcessor(processService));
+        this.nettyRemotingServer.registerProcessor(CommandType.EXECUTE_TASK_REQUEST, new TaskExecuteProcessor(processService));
+        this.nettyRemotingServer.registerProcessor(CommandType.KILL_TASK_REQUEST, new TaskKillProcessor());
         this.nettyRemotingServer.start();
 
         this.workerRegistry = new WorkerRegistry(zookeeperRegistryCenter, serverConfig.getListenPort());

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/CallbackChannel.java → dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/NettyRemoteChannel.java

@@ -18,40 +18,59 @@
 package org.apache.dolphinscheduler.server.worker.processor;
 
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import org.apache.dolphinscheduler.remote.command.Command;
+import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
+import org.apache.dolphinscheduler.remote.utils.Host;
 
 /**
  *  callback channel
  */
-public class CallbackChannel {
+public class NettyRemoteChannel {
 
     /**
      *  channel
      */
-    private Channel channel;
+    private final Channel channel;
 
     /**
      *  equest unique identification
      */
-    private long opaque;
+    private final long opaque;
 
-    public CallbackChannel(Channel channel, long opaque) {
+    /**
+     * master host
+     */
+    private final Host host;
+
+
+    public NettyRemoteChannel(Channel channel, long opaque) {
         this.channel = channel;
+        this.host = ChannelUtils.toAddress(channel);
         this.opaque = opaque;
     }
 
     public Channel getChannel() {
         return channel;
     }
 
-    public void setChannel(Channel channel) {
-        this.channel = channel;
-    }
-
     public long getOpaque() {
         return opaque;
     }
 
-    public void setOpaque(long opaque) {
-        this.opaque = opaque;
+    public Host getHost() {
+        return host;
+    }
+
+    public boolean isActive(){
+        return this.channel.isActive();
+    }
+
+    public ChannelFuture writeAndFlush(Command command){
+        return this.channel.writeAndFlush(command);
+    }
+
+    public void close(){
+        this.channel.close();
     }
 }