diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java index 8d8fd77f6a..6a4e1c628b 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.server.master.builder; +import static com.google.common.base.Preconditions.checkNotNull; import static org.apache.dolphinscheduler.common.constants.Constants.SEC_2_MINUTES_TIME_UNIT; import org.apache.dolphinscheduler.common.enums.TimeoutFlag; @@ -166,6 +167,11 @@ public class TaskExecutionContextBuilder { return this; } + public TaskExecutionContextBuilder buildWorkflowInstanceHost(String masterHost) { + taskExecutionContext.setWorkflowInstanceHost(masterHost); + return this; + } + /** * create * @@ -173,6 +179,7 @@ public class TaskExecutionContextBuilder { */ public TaskExecutionContext create() { + checkNotNull(taskExecutionContext.getWorkflowInstanceHost(), "The workflow instance host cannot be empty"); return taskExecutionContext; } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterRpcClient.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterRpcClient.java new file mode 100644 index 0000000000..4bc9788685 --- /dev/null +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterRpcClient.java @@ -0,0 +1,49 @@ +/* + * 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.rpc; + +import org.apache.dolphinscheduler.remote.NettyRemotingClient; +import org.apache.dolphinscheduler.remote.command.Command; +import org.apache.dolphinscheduler.remote.config.NettyClientConfig; +import org.apache.dolphinscheduler.remote.exceptions.RemotingException; +import org.apache.dolphinscheduler.remote.utils.Host; + +import lombok.NonNull; +import lombok.extern.slf4j.Slf4j; + +import org.springframework.stereotype.Component; + +@Slf4j +@Component +public class MasterRpcClient { + + private final NettyRemotingClient client; + + private static final long DEFAULT_TIME_OUT_MILLS = 10_000L; + + public MasterRpcClient() { + client = new NettyRemotingClient(new NettyClientConfig()); + log.info("Success initialized ApiServerRPCClient..."); + } + + public Command sendSyncCommand(@NonNull Host host, + @NonNull Command rpcCommand) throws RemotingException, InterruptedException { + return client.sendSync(host, rpcCommand, DEFAULT_TIME_OUT_MILLS); + } + +} diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java index 0934693e41..0a8ab85ee3 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java @@ -32,7 +32,6 @@ import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao; import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager; import org.apache.dolphinscheduler.server.master.config.MasterConfig; -import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; import org.apache.dolphinscheduler.server.master.event.WorkflowEvent; import org.apache.dolphinscheduler.server.master.event.WorkflowEventQueue; import org.apache.dolphinscheduler.server.master.event.WorkflowEventType; @@ -40,6 +39,7 @@ import org.apache.dolphinscheduler.server.master.exception.MasterException; import org.apache.dolphinscheduler.server.master.metrics.MasterServerMetrics; import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics; import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager; +import org.apache.dolphinscheduler.server.master.rpc.MasterRpcClient; import org.apache.dolphinscheduler.service.alert.ProcessAlertManager; import org.apache.dolphinscheduler.service.command.CommandService; import org.apache.dolphinscheduler.service.expand.CuringParamsService; @@ -87,7 +87,7 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl private ProcessAlertManager processAlertManager; @Autowired - private NettyExecutorManager nettyExecutorManager; + private MasterRpcClient masterRpcClient; /** * master prepare exec service @@ -189,7 +189,7 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl commandService, processService, processInstanceDao, - nettyExecutorManager, + masterRpcClient, processAlertManager, masterConfig, stateWheelExecuteThread, diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java index a869207a24..7ff33b1667 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java @@ -349,6 +349,7 @@ public class StreamTaskExecuteRunnable implements Runnable { .taskType(taskInstance.getTaskType()).taskParams(taskInstance.getTaskParams()).build()); Map propertyMap = paramParsingPreparation(taskInstance, baseParam); TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() + .buildWorkflowInstanceHost(masterConfig.getMasterAddress()) .buildTaskInstanceRelatedInfo(taskInstance) .buildTaskDefinitionRelatedInfo(taskDefinition) .buildResourceParametersInfo(resources) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java index 3951f121f5..7f79ea6967 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java @@ -67,10 +67,11 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.Direct; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.plugin.task.api.model.Property; import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; -import org.apache.dolphinscheduler.remote.command.HostUpdateCommand; +import org.apache.dolphinscheduler.remote.command.task.WorkflowHostChangeRequest; +import org.apache.dolphinscheduler.remote.command.task.WorkflowHostChangeResponse; +import org.apache.dolphinscheduler.remote.exceptions.RemotingException; import org.apache.dolphinscheduler.remote.utils.Host; import org.apache.dolphinscheduler.server.master.config.MasterConfig; -import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; import org.apache.dolphinscheduler.server.master.event.StateEvent; import org.apache.dolphinscheduler.server.master.event.StateEventHandleError; import org.apache.dolphinscheduler.server.master.event.StateEventHandleException; @@ -80,6 +81,7 @@ import org.apache.dolphinscheduler.server.master.event.StateEventHandlerManager; import org.apache.dolphinscheduler.server.master.event.TaskStateEvent; import org.apache.dolphinscheduler.server.master.event.WorkflowStateEvent; import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics; +import org.apache.dolphinscheduler.server.master.rpc.MasterRpcClient; import org.apache.dolphinscheduler.server.master.runner.task.ITaskProcessor; import org.apache.dolphinscheduler.server.master.runner.task.TaskAction; import org.apache.dolphinscheduler.server.master.runner.task.TaskProcessorFactory; @@ -99,6 +101,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.math.NumberUtils; import org.apache.commons.lang3.tuple.Pair; +import java.lang.reflect.InvocationTargetException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -144,7 +147,7 @@ public class WorkflowExecuteRunnable implements Callable { private final ProcessAlertManager processAlertManager; - private final NettyExecutorManager nettyExecutorManager; + private final MasterRpcClient masterRpcClient; private final ProcessInstance processInstance; @@ -245,7 +248,7 @@ public class WorkflowExecuteRunnable implements Callable { * @param processInstance processInstance * @param processService processService * @param processInstanceDao processInstanceDao - * @param nettyExecutorManager nettyExecutorManager + * @param masterRpcClient masterRpcClient * @param processAlertManager processAlertManager * @param masterConfig masterConfig * @param stateWheelExecuteThread stateWheelExecuteThread @@ -255,7 +258,7 @@ public class WorkflowExecuteRunnable implements Callable { @NonNull CommandService commandService, @NonNull ProcessService processService, @NonNull ProcessInstanceDao processInstanceDao, - @NonNull NettyExecutorManager nettyExecutorManager, + @NonNull MasterRpcClient masterRpcClient, @NonNull ProcessAlertManager processAlertManager, @NonNull MasterConfig masterConfig, @NonNull StateWheelExecuteThread stateWheelExecuteThread, @@ -266,7 +269,7 @@ public class WorkflowExecuteRunnable implements Callable { this.commandService = commandService; this.processInstanceDao = processInstanceDao; this.processInstance = processInstance; - this.nettyExecutorManager = nettyExecutorManager; + this.masterRpcClient = masterRpcClient; this.processAlertManager = processAlertManager; this.stateWheelExecuteThread = stateWheelExecuteThread; this.curingParamsService = curingParamsService; @@ -976,11 +979,6 @@ public class WorkflowExecuteRunnable implements Callable { ITaskProcessor taskProcessor = TaskProcessorFactory.getTaskProcessor(taskInstance.getTaskType()); taskProcessor.init(taskInstance, processInstance); - if (taskInstance.getState().isRunning() - && taskProcessor.getType().equalsIgnoreCase(Constants.COMMON_TASK_TYPE)) { - notifyProcessHostUpdate(taskInstance); - } - boolean submit = taskProcessor.action(TaskAction.SUBMIT); if (!submit) { log.error("Submit standby task failed!, taskCode: {}, taskName: {}", @@ -1064,23 +1062,6 @@ public class WorkflowExecuteRunnable implements Callable { } } - private void notifyProcessHostUpdate(TaskInstance taskInstance) { - if (StringUtils.isEmpty(taskInstance.getHost())) { - return; - } - - try { - HostUpdateCommand hostUpdateCommand = new HostUpdateCommand(); - hostUpdateCommand.setProcessHost(masterAddress); - hostUpdateCommand.setTaskInstanceId(taskInstance.getId()); - Host host = new Host(taskInstance.getHost()); - nettyExecutorManager.doExecute(host, hostUpdateCommand.convert2Command()); - } catch (Exception e) { - // Do we need to catch this exception? - log.error("notify process host update", e); - } - } - /** * find task instance in db. * in case submit more than one same name task in the same time. @@ -1370,7 +1351,25 @@ public class WorkflowExecuteRunnable implements Callable { TaskNode taskNodeObject = dag.getNode(taskNode); Optional existTaskInstanceOptional = getTaskInstance(taskNodeObject.getCode()); if (existTaskInstanceOptional.isPresent()) { - taskInstances.add(existTaskInstanceOptional.get()); + TaskInstance existTaskInstance = existTaskInstanceOptional.get(); + if (existTaskInstance.getState() == TaskExecutionStatus.RUNNING_EXECUTION + || existTaskInstance.getState() == TaskExecutionStatus.DISPATCH) { + // try to take over task instance + if (tryToTakeOverTaskInstance(existTaskInstance)) { + log.info("Success take over task {}", existTaskInstance.getName()); + continue; + } else { + // set the task instance state to fault tolerance + existTaskInstance.setFlag(Flag.NO); + existTaskInstance.setState(TaskExecutionStatus.NEED_FAULT_TOLERANCE); + validTaskMap.remove(existTaskInstance.getTaskCode()); + taskInstanceDao.updateTaskInstance(existTaskInstance); + existTaskInstance = cloneTolerantTaskInstance(existTaskInstance); + log.info("task {} cannot be take over will generate a tolerant task instance", + existTaskInstance.getName()); + } + } + taskInstances.add(existTaskInstance); continue; } TaskInstance task = createTaskInstance(processInstance, taskNodeObject); @@ -1416,6 +1415,46 @@ public class WorkflowExecuteRunnable implements Callable { updateProcessInstanceState(); } + private boolean tryToTakeOverTaskInstance(TaskInstance taskInstance) { + if (TaskProcessorFactory.isMasterTask(taskInstance.getTaskType())) { + return false; + } + try { + org.apache.dolphinscheduler.remote.command.Command command = + masterRpcClient.sendSyncCommand(Host.of(taskInstance.getHost()), + new WorkflowHostChangeRequest(taskInstance.getId(), masterAddress).convert2Command()); + if (command == null) { + log.error( + "Takeover task instance failed, the worker {} might not be alive, will try to create a new task instance", + taskInstance.getHost()); + return false; + } + WorkflowHostChangeResponse workflowHostChangeResponse = + JSONUtils.parseObject(command.getBody(), WorkflowHostChangeResponse.class); + if (workflowHostChangeResponse == null || !workflowHostChangeResponse.isSuccess()) { + log.error( + "Takeover task instance failed, receive a failed response from worker: {}, will try to create a new task instance", + taskInstance.getHost()); + return false; + } + + ITaskProcessor taskProcessor = TaskProcessorFactory.getTaskProcessor(taskInstance.getTaskType()); + taskProcessor.init(taskInstance, processInstance); + + taskInstanceMap.put(taskInstance.getId(), taskInstance); + stateWheelExecuteThread.addTask4TimeoutCheck(processInstance, taskInstance); + stateWheelExecuteThread.addTask4RetryCheck(processInstance, taskInstance); + activeTaskProcessorMaps.put(taskInstance.getTaskCode(), taskProcessor); + return true; + } catch (RemotingException | InterruptedException | InstantiationException | IllegalAccessException + | InvocationTargetException e) { + log.error( + "Takeover task instance failed, the worker {} might not be alive, will try to create a new task instance", + taskInstance.getHost(), e); + return false; + } + } + /** * determine whether the dependencies of the task node are complete * diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java index 31bfe4c772..56683c2b22 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java @@ -341,6 +341,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor { Map propertyMap = curingParamsService.paramParsingPreparation(taskInstance, baseParam, processInstance); return TaskExecutionContextBuilder.get() + .buildWorkflowInstanceHost(masterConfig.getMasterAddress()) .buildTaskInstanceRelatedInfo(taskInstance) .buildTaskDefinitionRelatedInfo(taskInstance.getTaskDefine()) .buildProcessInstanceRelatedInfo(taskInstance.getProcessInstance()) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/MasterFailoverService.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/MasterFailoverService.java index 915b8f6ec9..a95cb9debb 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/MasterFailoverService.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/MasterFailoverService.java @@ -20,45 +20,26 @@ package org.apache.dolphinscheduler.server.master.service; import org.apache.dolphinscheduler.common.constants.Constants; import org.apache.dolphinscheduler.common.enums.NodeType; import org.apache.dolphinscheduler.common.model.Server; -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.repository.ProcessDefinitionDao; -import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao; -import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; -import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; import org.apache.dolphinscheduler.registry.api.RegistryClient; -import org.apache.dolphinscheduler.remote.command.TaskKillRequestCommand; -import org.apache.dolphinscheduler.remote.utils.Host; -import org.apache.dolphinscheduler.server.master.builder.TaskExecutionContextBuilder; import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager; import org.apache.dolphinscheduler.server.master.config.MasterConfig; -import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException; -import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics; -import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics; -import org.apache.dolphinscheduler.server.master.runner.task.TaskProcessorFactory; -import org.apache.dolphinscheduler.service.log.LogClient; import org.apache.dolphinscheduler.service.process.ProcessService; -import org.apache.dolphinscheduler.service.utils.ProcessUtils; import org.apache.commons.collections4.CollectionUtils; -import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.time.StopWatch; import java.util.Date; import java.util.List; -import java.util.Map; import java.util.Optional; import java.util.concurrent.TimeUnit; -import java.util.function.Function; import java.util.stream.Collectors; import lombok.NonNull; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import io.micrometer.core.annotation.Counted; @@ -73,32 +54,17 @@ public class MasterFailoverService { private final ProcessService processService; private final String localAddress; - private final NettyExecutorManager nettyExecutorManager; - private final ProcessInstanceExecCacheManager processInstanceExecCacheManager; - private final LogClient logClient; - - private final TaskInstanceDao taskInstanceDao; - - @Autowired - private ProcessDefinitionDao processDefinitionDao; - public MasterFailoverService(@NonNull RegistryClient registryClient, @NonNull MasterConfig masterConfig, @NonNull ProcessService processService, - @NonNull NettyExecutorManager nettyExecutorManager, - @NonNull ProcessInstanceExecCacheManager processInstanceExecCacheManager, - @NonNull LogClient logClient, - @NonNull TaskInstanceDao taskInstanceDao) { + @NonNull ProcessInstanceExecCacheManager processInstanceExecCacheManager) { this.registryClient = registryClient; this.masterConfig = masterConfig; this.processService = processService; - this.nettyExecutorManager = nettyExecutorManager; this.localAddress = masterConfig.getMasterAddress(); this.processInstanceExecCacheManager = processInstanceExecCacheManager; - this.logClient = logClient; - this.taskInstanceDao = taskInstanceDao; } /** @@ -159,49 +125,19 @@ public class MasterFailoverService { needFailoverProcessInstanceList.size(), needFailoverProcessInstanceList.stream().map(ProcessInstance::getId).collect(Collectors.toList())); - List processDefinitions = - processDefinitionDao.queryProcessDefinitionsByCodesAndVersions(needFailoverProcessInstanceList); - Map codeDefinitionMap = processDefinitions - .stream() - .collect(Collectors.toMap(ProcessDefinition::getCode, Function.identity())); - for (ProcessInstance processInstance : needFailoverProcessInstanceList) { - try { - LogUtils.setWorkflowInstanceIdMDC(processInstance.getId()); + try ( + LogUtils.MDCAutoClosableContext mdcAutoClosableContext = + LogUtils.setWorkflowInstanceIdMDC(processInstance.getId())) { log.info("WorkflowInstance failover starting"); if (!checkProcessInstanceNeedFailover(masterStartupTimeOptional, processInstance)) { - log.info("WorkflowInstance doesn't need to failover"); continue; } - ProcessDefinition processDefinition = codeDefinitionMap.get(processInstance.getProcessDefinitionCode()); - processInstance.setProcessDefinition(processDefinition); - int processInstanceId = processInstance.getId(); - List taskInstanceList = - taskInstanceDao.findValidTaskListByProcessId(processInstanceId, processInstance.getTestFlag()); - for (TaskInstance taskInstance : taskInstanceList) { - try { - LogUtils.setTaskInstanceIdMDC(taskInstance.getId()); - log.info("TaskInstance failover starting"); - if (!checkTaskInstanceNeedFailover(taskInstance)) { - log.info("The taskInstance doesn't need to failover"); - continue; - } - failoverTaskInstance(processInstance, taskInstance); - log.info("TaskInstance failover finished"); - } finally { - LogUtils.removeTaskInstanceIdMDC(); - } - } ProcessInstanceMetrics.incProcessInstanceByStateAndProcessDefinitionCode("failover", processInstance.getProcessDefinitionCode().toString()); - // updateProcessInstance host is null to mark this processInstance has been failover - // and insert a failover command - processInstance.setHost(Constants.NULL); processService.processNeedFailoverProcessInstances(processInstance); log.info("WorkflowInstance failover finished"); - } finally { - LogUtils.removeWorkflowInstanceIdMDC(); } } @@ -225,75 +161,13 @@ public class MasterFailoverService { return Optional.ofNullable(serverStartupTime); } - /** - * failover task instance - *

- * 1. kill yarn/k8s job if run on worker and there are yarn/k8s jobs in tasks. - * 2. change task state from running to need failover. - * 3. try to notify local master - * - * @param processInstance - * @param taskInstance - */ - private void failoverTaskInstance(@NonNull ProcessInstance processInstance, @NonNull TaskInstance taskInstance) { - TaskMetrics.incTaskInstanceByState("failover"); - boolean isMasterTask = TaskProcessorFactory.isMasterTask(taskInstance.getTaskType()); - - taskInstance.setProcessInstance(processInstance); - - if (!isMasterTask) { - log.info("The failover taskInstance is not master task"); - TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() - .buildTaskInstanceRelatedInfo(taskInstance) - .buildProcessInstanceRelatedInfo(processInstance) - .buildProcessDefinitionRelatedInfo(processInstance.getProcessDefinition()) - .create(); - - if (masterConfig.isKillApplicationWhenTaskFailover()) { - // only kill yarn/k8s job if exists , the local thread has exited - log.info("TaskInstance failover begin kill the task related yarn or k8s job"); - ProcessUtils.killApplication(logClient, taskExecutionContext); - } - // kill worker task, When the master failover and worker failover happened in the same time, - // the task may not be failover if we don't set NEED_FAULT_TOLERANCE. - // This can be improved if we can load all task when cache a workflowInstance in memory - sendKillCommandToWorker(taskInstance); - } else { - log.info("The failover taskInstance is a master task"); - } - - taskInstance.setState(TaskExecutionStatus.NEED_FAULT_TOLERANCE); - taskInstanceDao.upsertTaskInstance(taskInstance); - } - - private void sendKillCommandToWorker(@NonNull TaskInstance taskInstance) { - if (StringUtils.isEmpty(taskInstance.getHost())) { - return; - } - try { - TaskKillRequestCommand killCommand = new TaskKillRequestCommand(taskInstance.getId()); - Host workerHost = Host.of(taskInstance.getHost()); - nettyExecutorManager.doExecute(workerHost, killCommand.convert2Command()); - log.info("Failover task success, has killed the task in worker: {}", taskInstance.getHost()); - } catch (ExecuteException e) { - log.error("Kill task failed", e); - } - } - - private boolean checkTaskInstanceNeedFailover(@NonNull TaskInstance taskInstance) { - if (taskInstance.getState() != null && taskInstance.getState().isFinished()) { - // The task is already finished, so we don't need to failover this task instance - return false; - } - return true; - } - private boolean checkProcessInstanceNeedFailover(Optional beFailoveredMasterStartupTimeOptional, @NonNull ProcessInstance processInstance) { // The process has already been failover, since when we do master failover we will hold a lock, so we can // guarantee // the host will not be set concurrent. if (Constants.NULL.equals(processInstance.getHost())) { + log.info("The workflowInstance's host is NULL, no need to failover"); return false; } if (!beFailoveredMasterStartupTimeOptional.isPresent()) { @@ -304,16 +178,20 @@ public class MasterFailoverService { if (processInstance.getStartTime().after(beFailoveredMasterStartupTime)) { // The processInstance is newly created + log.info("The workflowInstance is newly created, no need to failover"); return false; } if (processInstance.getRestartTime() != null && processInstance.getRestartTime().after(beFailoveredMasterStartupTime)) { // the processInstance is already be failovered. + log.info( + "The workflowInstance's restartTime is after the dead master startup time, no need to failover"); return false; } if (processInstanceExecCacheManager.contains(processInstance.getId())) { // the processInstance is a running process instance in the current master + log.info("The workflowInstance is running in the current master, no need to failover"); return false; } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java index 315c02ec09..0c9e4d3102 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java @@ -170,6 +170,7 @@ public class WorkerFailoverService { if (!isMasterTask) { log.info("The failover taskInstance is not master task"); TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() + .buildWorkflowInstanceHost(masterConfig.getMasterAddress()) .buildTaskInstanceRelatedInfo(taskInstance) .buildProcessInstanceRelatedInfo(processInstance) .buildProcessDefinitionRelatedInfo(processInstance.getProcessDefinition()) @@ -181,7 +182,7 @@ public class WorkerFailoverService { ProcessUtils.killApplication(logClient, taskExecutionContext); } } else { - log.info("The failover taskInstance is a master task"); + log.info("The failover taskInstance is a master task, no need to failover in worker failover"); } taskInstance.setState(TaskExecutionStatus.NEED_FAULT_TOLERANCE); diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutionContextTestUtils.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutionContextTestUtils.java index ed34fa0603..b95356acc6 100644 --- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutionContextTestUtils.java +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutionContextTestUtils.java @@ -46,6 +46,7 @@ public class ExecutionContextTestUtils { processInstance.setCommandType(CommandType.COMPLEMENT_DATA); taskInstance.setProcessInstance(processInstance); TaskExecutionContext context = TaskExecutionContextBuilder.get() + .buildWorkflowInstanceHost("127.0.0.1:5678") .buildTaskInstanceRelatedInfo(taskInstance) .buildProcessInstanceRelatedInfo(processInstance) .buildProcessDefinitionRelatedInfo(processDefinition) diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java index 13b6c48004..bb86c11dde 100644 --- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java @@ -36,7 +36,7 @@ import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao; import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao; import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao; import org.apache.dolphinscheduler.server.master.config.MasterConfig; -import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; +import org.apache.dolphinscheduler.server.master.rpc.MasterRpcClient; import org.apache.dolphinscheduler.service.alert.ProcessAlertManager; import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; import org.apache.dolphinscheduler.service.command.CommandService; @@ -116,11 +116,11 @@ public class WorkflowExecuteRunnableTest { stateWheelExecuteThread = Mockito.mock(StateWheelExecuteThread.class); curingGlobalParamsService = Mockito.mock(CuringParamsService.class); - NettyExecutorManager nettyExecutorManager = Mockito.mock(NettyExecutorManager.class); + MasterRpcClient masterRpcClient = Mockito.mock(MasterRpcClient.class); ProcessAlertManager processAlertManager = Mockito.mock(ProcessAlertManager.class); workflowExecuteThread = Mockito.spy( new WorkflowExecuteRunnable(processInstance, commandService, processService, processInstanceDao, - nettyExecutorManager, + masterRpcClient, processAlertManager, config, stateWheelExecuteThread, curingGlobalParamsService, taskInstanceDao, taskDefinitionLogDao)); Field dag = WorkflowExecuteRunnable.class.getDeclaredField("dag"); diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java index 8289da93b9..030221ef6e 100644 --- a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java @@ -29,13 +29,11 @@ import org.apache.dolphinscheduler.common.model.Server; import org.apache.dolphinscheduler.common.utils.NetUtils; import org.apache.dolphinscheduler.dao.entity.ProcessInstance; import org.apache.dolphinscheduler.dao.entity.TaskInstance; -import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao; import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.registry.api.RegistryClient; import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager; import org.apache.dolphinscheduler.server.master.config.MasterConfig; -import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; import org.apache.dolphinscheduler.server.master.event.StateEvent; import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable; import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThreadPool; @@ -43,7 +41,6 @@ import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; import org.apache.dolphinscheduler.service.log.LogClient; import org.apache.dolphinscheduler.service.process.ProcessService; -import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Arrays; import java.util.Date; @@ -58,7 +55,6 @@ import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; import org.springframework.context.ApplicationContext; -import org.springframework.util.ReflectionUtils; import com.google.common.collect.Lists; @@ -86,18 +82,12 @@ public class FailoverServiceTest { @Mock private ProcessInstanceExecCacheManager cacheManager; - @Mock - private NettyExecutorManager nettyExecutorManager; - @Mock private ProcessInstanceExecCacheManager processInstanceExecCacheManager; @Mock private LogClient logClient; - @Mock - private ProcessDefinitionDao processDefinitionDao; - private static int masterPort = 5678; private static int workerPort = 1234; @@ -117,11 +107,8 @@ public class FailoverServiceTest { testMasterHost = NetUtils.getAddr(masterConfig.getListenPort()); given(masterConfig.getMasterAddress()).willReturn(testMasterHost); MasterFailoverService masterFailoverService = - new MasterFailoverService(registryClient, masterConfig, processService, nettyExecutorManager, - processInstanceExecCacheManager, logClient, taskInstanceDao); - Field processDefinitionDaoField = masterFailoverService.getClass().getDeclaredField("processDefinitionDao"); - processDefinitionDaoField.setAccessible(true); - ReflectionUtils.setField(processDefinitionDaoField, masterFailoverService, processDefinitionDao); + new MasterFailoverService(registryClient, masterConfig, processService, + processInstanceExecCacheManager); WorkerFailoverService workerFailoverService = new WorkerFailoverService(registryClient, masterConfig, processService, @@ -196,13 +183,11 @@ public class FailoverServiceTest { masterTaskInstance.setState(TaskExecutionStatus.SUCCESS); failoverService.failoverServerWhenDown(testMasterHost, NodeType.MASTER); Assertions.assertNotEquals(masterTaskInstance.getState(), TaskExecutionStatus.NEED_FAULT_TOLERANCE); - Assertions.assertEquals(Constants.NULL, processInstance.getHost()); processInstance.setHost(testMasterHost); masterTaskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION); failoverService.failoverServerWhenDown(testMasterHost, NodeType.MASTER); - Assertions.assertEquals(masterTaskInstance.getState(), TaskExecutionStatus.NEED_FAULT_TOLERANCE); - Assertions.assertEquals(Constants.NULL, processInstance.getHost()); + Assertions.assertEquals(masterTaskInstance.getState(), TaskExecutionStatus.RUNNING_EXECUTION); } @Test diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java index 88f0f117f3..997de50638 100644 --- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java +++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java @@ -99,15 +99,8 @@ public enum CommandType { ALERT_SEND_RESPONSE, - /** - * process host update - */ - PROCESS_HOST_UPDATE_REQUEST, - - /** - * process host update response - */ - PROCESS_HOST_UPDATE_RESPONSE, + WORKFLOW_HOST_CHANGE_REQUEST, + WORKFLOW_HOST_CHANGE_RESPONSE, /** * state event request diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateResponseCommand.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateResponseCommand.java deleted file mode 100644 index b44856c83d..0000000000 --- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateResponseCommand.java +++ /dev/null @@ -1,83 +0,0 @@ -/* - * 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.common.utils.JSONUtils; - -import java.io.Serializable; - -public class HostUpdateResponseCommand implements Serializable { - - private int taskInstanceId; - - private String processHost; - - private int status; - - public HostUpdateResponseCommand(int taskInstanceId, String processHost, int code) { - this.taskInstanceId = taskInstanceId; - this.processHost = processHost; - this.status = code; - } - - public int getTaskInstanceId() { - return this.taskInstanceId; - } - - public void setTaskInstanceId(int taskInstanceId) { - this.taskInstanceId = taskInstanceId; - } - - public String getProcessHost() { - return this.processHost; - } - - public void setProcessHost(String processHost) { - this.processHost = processHost; - } - - public int getStatus() { - return status; - } - - public void setStatus(int status) { - this.status = status; - } - - /** - * package request command - * - * @return command - */ - public Command convert2Command() { - Command command = new Command(); - command.setType(CommandType.PROCESS_HOST_UPDATE_REQUEST); - byte[] body = JSONUtils.toJsonByteArray(this); - command.setBody(body); - return command; - } - - @Override - public String toString() { - return "HostUpdateResponseCommand{" - + "taskInstanceId=" + taskInstanceId - + "host=" + processHost - + '}'; - } - -} diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateCommand.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeRequest.java similarity index 73% rename from dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateCommand.java rename to dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeRequest.java index b55453b206..1a480e247f 100644 --- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateCommand.java +++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeRequest.java @@ -15,23 +15,26 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.remote.command; +package org.apache.dolphinscheduler.remote.command.task; import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.remote.command.Command; +import org.apache.dolphinscheduler.remote.command.CommandType; import java.io.Serializable; +import lombok.AllArgsConstructor; import lombok.Data; +import lombok.NoArgsConstructor; -/** - * process host update - */ @Data -public class HostUpdateCommand implements Serializable { +@AllArgsConstructor +@NoArgsConstructor +public class WorkflowHostChangeRequest implements Serializable { private int taskInstanceId; - private String processHost; + private String workflowHost; /** * package request command @@ -40,10 +43,9 @@ public class HostUpdateCommand implements Serializable { */ public Command convert2Command() { Command command = new Command(); - command.setType(CommandType.PROCESS_HOST_UPDATE_REQUEST); + command.setType(CommandType.WORKFLOW_HOST_CHANGE_REQUEST); byte[] body = JSONUtils.toJsonByteArray(this); command.setBody(body); return command; } - } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/processor/HostUpdateResponseProcessor.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeResponse.java similarity index 50% rename from dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/processor/HostUpdateResponseProcessor.java rename to dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeResponse.java index c5a2af34da..5ce58f2aa2 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/processor/HostUpdateResponseProcessor.java +++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeResponse.java @@ -15,28 +15,42 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.server.master.processor; +package org.apache.dolphinscheduler.remote.command.task; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.remote.command.Command; import org.apache.dolphinscheduler.remote.command.CommandType; -import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; -import lombok.extern.slf4j.Slf4j; +import java.io.Serializable; -import com.google.common.base.Preconditions; -import io.netty.channel.Channel; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; -@Slf4j -public class HostUpdateResponseProcessor implements NettyRequestProcessor { +@Data +@AllArgsConstructor +@NoArgsConstructor +public class WorkflowHostChangeResponse implements Serializable { - @Override - public void process(Channel channel, Command command) { - Preconditions.checkArgument(CommandType.PROCESS_HOST_UPDATE_RESPONSE == command.getType(), - String.format("invalid command type : %s", command.getType())); + boolean success; - HostUpdateResponseProcessor responseCommand = - JSONUtils.parseObject(command.getBody(), HostUpdateResponseProcessor.class); - log.info("received process host response command : {}", responseCommand); + public static WorkflowHostChangeResponse success() { + WorkflowHostChangeResponse response = new WorkflowHostChangeResponse(); + response.setSuccess(true); + return response; + } + + public static WorkflowHostChangeResponse failed() { + WorkflowHostChangeResponse response = new WorkflowHostChangeResponse(); + response.setSuccess(false); + return response; + } + + public Command convert2Command(long opaque) { + Command command = new Command(opaque); + command.setType(CommandType.WORKFLOW_HOST_CHANGE_RESPONSE); + byte[] body = JSONUtils.toJsonByteArray(this); + command.setBody(body); + return command; } } diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java index b38bff49d4..0c7b4a9b36 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java @@ -1589,17 +1589,15 @@ public class ProcessServiceImpl implements ProcessService { @Override @Transactional public void processNeedFailoverProcessInstances(ProcessInstance processInstance) { - // 1 update processInstance host is null + // updateProcessInstance host is null to mark this processInstance has been failover + // and insert a failover command processInstance.setHost(Constants.NULL); processInstanceMapper.updateById(processInstance); - ProcessDefinition processDefinition = findProcessDefinition(processInstance.getProcessDefinitionCode(), - processInstance.getProcessDefinitionVersion()); - // 2 insert into recover command Command cmd = new Command(); - cmd.setProcessDefinitionCode(processDefinition.getCode()); - cmd.setProcessDefinitionVersion(processDefinition.getVersion()); + cmd.setProcessDefinitionCode(processInstance.getProcessDefinitionCode()); + cmd.setProcessDefinitionVersion(processInstance.getProcessDefinitionVersion()); cmd.setProcessInstanceId(processInstance.getId()); cmd.setCommandParam( String.format("{\"%s\":%d}", CMD_PARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId())); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java index efc0289bf8..80a84db327 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java @@ -69,6 +69,8 @@ public class TaskExecutionContext implements Serializable { */ private String taskType; + private String workflowInstanceHost; + /** * host */ diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContextCacheManager.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContextCacheManager.java index 5ee1f35d17..dbe39d7042 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContextCacheManager.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContextCacheManager.java @@ -30,7 +30,7 @@ public class TaskExecutionContextCacheManager { /** * taskInstance cache */ - private static Map taskRequestContextCache = new ConcurrentHashMap<>(); + private static final Map taskRequestContextCache = new ConcurrentHashMap<>(); /** * get taskInstance by taskInstance id diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/log/TaskInstanceLogHeader.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/log/TaskInstanceLogHeader.java index 12e0250168..4bfb995bb5 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/log/TaskInstanceLogHeader.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/log/TaskInstanceLogHeader.java @@ -14,27 +14,50 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.dolphinscheduler.server.worker.log; +import java.util.List; + +import lombok.extern.slf4j.Slf4j; + +import com.google.common.collect.Lists; + +@Slf4j public class TaskInstanceLogHeader { - public static final String INITIALIZE_TASK_CONTEXT_HEADER = "" + - "***********************************************************************************************\n" + - "********************************* Initialize task context ***********************************\n" + - "***********************************************************************************************\n"; + private static final List INITIALIZE_TASK_CONTEXT_HEADER = Lists.newArrayList( + "***********************************************************************************************", + "********************************* Initialize task context ***********************************", + "***********************************************************************************************"); + private static final List LOAD_TASK_INSTANCE_PLUGIN_HEADER = Lists.newArrayList( + "***********************************************************************************************", + "********************************* Load task instance plugin *********************************", + "***********************************************************************************************"); - public static final String LOAD_TASK_INSTANCE_PLUGIN_HEADER = "" + - "***********************************************************************************************\n" + - "********************************* Load task instance plugin *********************************\n" + - "***********************************************************************************************\n"; + public static void printInitializeTaskContextHeader() { + INITIALIZE_TASK_CONTEXT_HEADER.forEach(log::info); + } - public static final String EXECUTE_TASK_HEADER = "" + - "***********************************************************************************************\n" + - "********************************* Execute task instance *************************************\n" + - "***********************************************************************************************\n"; + private static final List EXECUTE_TASK_HEADER = Lists.newArrayList( + "***********************************************************************************************", + "********************************* Execute task instance *************************************", + "***********************************************************************************************"); - public static final String FINALIZE_TASK_HEADER = "" + - "***********************************************************************************************\n" + - "********************************* Finalize task instance ************************************\n" + - "***********************************************************************************************\n"; + private static final List FINALIZE_TASK_HEADER = Lists.newArrayList( + "***********************************************************************************************", + "********************************* Finalize task instance ************************************", + "***********************************************************************************************"); + + public static void printLoadTaskInstancePluginHeader() { + LOAD_TASK_INSTANCE_PLUGIN_HEADER.forEach(log::info); + } + + public static void printExecuteTaskHeader() { + EXECUTE_TASK_HEADER.forEach(log::info); + } + + public static void printFinalizeTaskHeader() { + FINALIZE_TASK_HEADER.forEach(log::info); + } } diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageSender.java index 519c5b2b3e..fc9a9ee81f 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageSender.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageSender.java @@ -34,7 +34,7 @@ public interface MessageSender { /** * Build the message from task context and message received address. */ - T buildMessage(TaskExecutionContext taskExecutionContext, String messageReceiverAddress); + T buildMessage(TaskExecutionContext taskExecutionContext); /** * The message type can be sent by this sender. diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteResultMessageSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteResultMessageSender.java index 3cdabad3de..dd410e7c60 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteResultMessageSender.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteResultMessageSender.java @@ -42,11 +42,11 @@ public class TaskExecuteResultMessageSender implements MessageSender workerRpcClient.send(Host.of(message.getMessageReceiverAddress()), message.convert2Command()); } - public TaskRejectCommand buildMessage(TaskExecutionContext taskExecutionContext, String masterAddress) { + @Override + public TaskRejectCommand buildMessage(TaskExecutionContext taskExecutionContext) { TaskRejectCommand taskRejectMessage = new TaskRejectCommand(workerConfig.getWorkerAddress(), - masterAddress, + taskExecutionContext.getWorkflowInstanceHost(), System.currentTimeMillis()); taskRejectMessage.setTaskInstanceId(taskExecutionContext.getTaskInstanceId()); taskRejectMessage.setProcessInstanceId(taskExecutionContext.getProcessInstanceId()); diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskUpdatePidMessageSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskUpdatePidMessageSender.java index e50c00df7a..8109ef5f64 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskUpdatePidMessageSender.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskUpdatePidMessageSender.java @@ -45,11 +45,10 @@ public class TaskUpdatePidMessageSender implements MessageSender 0) { log.info("Current taskInstance is choose delay execution, delay time: {}s", remainTime); taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION); - workerMessageSender.sendMessage(taskExecutionContext, workflowMasterAddress, - CommandType.TASK_EXECUTE_RESULT); + workerMessageSender.sendMessage(taskExecutionContext, CommandType.TASK_EXECUTE_RESULT); } WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder .createWorkerDelayTaskExecuteRunnableFactory( taskExecutionContext, workerConfig, - workflowMasterAddress, workerMessageSender, workerRpcClient, taskPluginManager, @@ -131,8 +129,7 @@ public class TaskDispatchProcessor implements NettyRequestProcessor { log.warn( "submit task to wait queue error, queue is full, current queue size is {}, will send a task reject message to master", workerManager.getWaitSubmitQueueSize()); - workerMessageSender.sendMessageWithRetry(taskExecutionContext, workflowMasterAddress, - CommandType.TASK_REJECT); + workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_REJECT); } else { log.info("Submit task to wait queue success, current queue size is {}", workerManager.getWaitSubmitQueueSize()); diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkflowHostChangeProcessor.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkflowHostChangeProcessor.java new file mode 100644 index 0000000000..33da734151 --- /dev/null +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkflowHostChangeProcessor.java @@ -0,0 +1,91 @@ +/* + * 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.worker.processor; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheManager; +import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; +import org.apache.dolphinscheduler.remote.command.Command; +import org.apache.dolphinscheduler.remote.command.CommandType; +import org.apache.dolphinscheduler.remote.command.task.WorkflowHostChangeRequest; +import org.apache.dolphinscheduler.remote.command.task.WorkflowHostChangeResponse; +import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; +import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import com.google.common.base.Preconditions; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelFutureListener; + +/** + * update process host + * this used when master failover + */ +@Component +public class WorkflowHostChangeProcessor implements NettyRequestProcessor { + + private final Logger logger = LoggerFactory.getLogger(WorkflowHostChangeProcessor.class); + + @Autowired + private MessageRetryRunner messageRetryRunner; + + @Override + public void process(Channel channel, Command command) { + Preconditions.checkArgument(CommandType.WORKFLOW_HOST_CHANGE_REQUEST == command.getType(), + String.format("invalid command type : %s", command.getType())); + WorkflowHostChangeRequest workflowHostChangeRequest = + JSONUtils.parseObject(command.getBody(), WorkflowHostChangeRequest.class); + if (workflowHostChangeRequest == null) { + logger.error("host update command is null"); + return; + } + logger.info("Received workflow host change command : {}", workflowHostChangeRequest); + try ( + final LogUtils.MDCAutoClosableContext mdcAutoClosableContext = + LogUtils.setTaskInstanceIdMDC(workflowHostChangeRequest.getTaskInstanceId())) { + WorkflowHostChangeResponse workflowHostChangeResponse; + TaskExecutionContext taskExecutionContext = + TaskExecutionContextCacheManager.getByTaskInstanceId(workflowHostChangeRequest.getTaskInstanceId()); + if (taskExecutionContext != null) { + taskExecutionContext.setWorkflowInstanceHost(workflowHostChangeRequest.getWorkflowHost()); + messageRetryRunner.updateMessageHost(workflowHostChangeRequest.getTaskInstanceId(), + workflowHostChangeRequest.getWorkflowHost()); + workflowHostChangeResponse = WorkflowHostChangeResponse.success(); + logger.info("Success update workflow host, taskInstanceId : {}, workflowHost: {}", + workflowHostChangeRequest.getTaskInstanceId(), workflowHostChangeRequest.getWorkflowHost()); + } else { + workflowHostChangeResponse = WorkflowHostChangeResponse.failed(); + logger.error("Cannot find the taskExecutionContext, taskInstanceId : {}", + workflowHostChangeRequest.getTaskInstanceId()); + } + channel.writeAndFlush(workflowHostChangeResponse.convert2Command(command.getOpaque())).addListener( + (ChannelFutureListener) channelFuture -> { + if (!channelFuture.isSuccess()) { + logger.error("send host update response failed"); + } + }); + } + } + +} diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerMessageSender.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerMessageSender.java index d816420493..f61bb332e7 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerMessageSender.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerMessageSender.java @@ -56,13 +56,12 @@ public class WorkerMessageSender { // todo: use message rather than context public void sendMessageWithRetry(@NonNull TaskExecutionContext taskExecutionContext, - @NonNull String messageReceiverAddress, @NonNull CommandType messageType) { MessageSender messageSender = messageSenderMap.get(messageType); if (messageSender == null) { throw new IllegalArgumentException("The messageType is invalidated, messageType: " + messageType); } - BaseCommand baseCommand = messageSender.buildMessage(taskExecutionContext, messageReceiverAddress); + BaseCommand baseCommand = messageSender.buildMessage(taskExecutionContext); try { messageRetryRunner.addRetryMessage(taskExecutionContext.getTaskInstanceId(), messageType, baseCommand); messageSender.sendMessage(baseCommand); @@ -71,14 +70,12 @@ public class WorkerMessageSender { } } - public void sendMessage(@NonNull TaskExecutionContext taskExecutionContext, - @NonNull String messageReceiverAddress, - @NonNull CommandType messageType) { + public void sendMessage(@NonNull TaskExecutionContext taskExecutionContext, @NonNull CommandType messageType) { MessageSender messageSender = messageSenderMap.get(messageType); if (messageSender == null) { throw new IllegalArgumentException("The messageType is invalidated, messageType: " + messageType); } - BaseCommand baseCommand = messageSender.buildMessage(taskExecutionContext, messageReceiverAddress); + BaseCommand baseCommand = messageSender.buildMessage(taskExecutionContext); try { messageSender.sendMessage(baseCommand); } catch (RemotingException e) { diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerRpcServer.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerRpcServer.java index 6af593460b..f9d64e6551 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerRpcServer.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerRpcServer.java @@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.remote.command.CommandType; import org.apache.dolphinscheduler.remote.config.NettyServerConfig; import org.apache.dolphinscheduler.remote.processor.LoggerRequestProcessor; import org.apache.dolphinscheduler.server.worker.config.WorkerConfig; -import org.apache.dolphinscheduler.server.worker.processor.HostUpdateProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskDispatchProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskExecuteResultAckProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskExecuteRunningAckProcessor; @@ -30,6 +29,7 @@ import org.apache.dolphinscheduler.server.worker.processor.TaskKillProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskRejectAckProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskSavePointProcessor; import org.apache.dolphinscheduler.server.worker.processor.TaskUpdatePidAckProcessor; +import org.apache.dolphinscheduler.server.worker.processor.WorkflowHostChangeProcessor; import java.io.Closeable; @@ -66,7 +66,7 @@ public class WorkerRpcServer implements Closeable { private TaskExecuteResultAckProcessor taskExecuteResultAckProcessor; @Autowired - private HostUpdateProcessor hostUpdateProcessor; + private WorkflowHostChangeProcessor workflowHostChangeProcessor; @Autowired private LoggerRequestProcessor loggerRequestProcessor; @@ -89,7 +89,8 @@ public class WorkerRpcServer implements Closeable { taskUpdatePidAckProcessor); this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESULT_ACK, taskExecuteResultAckProcessor); this.nettyRemotingServer.registerProcessor(CommandType.TASK_REJECT_ACK, taskRejectAckProcessor); - this.nettyRemotingServer.registerProcessor(CommandType.PROCESS_HOST_UPDATE_REQUEST, hostUpdateProcessor); + this.nettyRemotingServer.registerProcessor(CommandType.WORKFLOW_HOST_CHANGE_REQUEST, + workflowHostChangeProcessor); this.nettyRemotingServer.registerProcessor(CommandType.TASK_SAVEPOINT_REQUEST, taskSavePointProcessor); // log server this.nettyRemotingServer.registerProcessor(CommandType.GET_APP_ID_REQUEST, loggerRequestProcessor); diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java index 92c9baf718..1f5c021717 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java @@ -34,14 +34,12 @@ public class DefaultWorkerDelayTaskExecuteRunnable extends WorkerDelayTaskExecut public DefaultWorkerDelayTaskExecuteRunnable(@NonNull TaskExecutionContext taskExecutionContext, @NonNull WorkerConfig workerConfig, - @NonNull String workflowMaster, @NonNull WorkerMessageSender workerMessageSender, @NonNull WorkerRpcClient workerRpcClient, @NonNull TaskPluginManager taskPluginManager, @Nullable StorageOperate storageOperate) { super(taskExecutionContext, workerConfig, - workflowMaster, workerMessageSender, workerRpcClient, taskPluginManager, diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java index 151800742c..58f406a725 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java @@ -34,14 +34,12 @@ public class DefaultWorkerDelayTaskExecuteRunnableFactory protected DefaultWorkerDelayTaskExecuteRunnableFactory(@NonNull TaskExecutionContext taskExecutionContext, @NonNull WorkerConfig workerConfig, - @NonNull String workflowMasterAddress, @NonNull WorkerMessageSender workerMessageSender, @NonNull WorkerRpcClient workerRpcClient, @NonNull TaskPluginManager taskPluginManager, @Nullable StorageOperate storageOperate) { super(taskExecutionContext, workerConfig, - workflowMasterAddress, workerMessageSender, workerRpcClient, taskPluginManager, @@ -53,7 +51,6 @@ public class DefaultWorkerDelayTaskExecuteRunnableFactory return new DefaultWorkerDelayTaskExecuteRunnable( taskExecutionContext, workerConfig, - workflowMasterAddress, workerMessageSender, workerRpcClient, taskPluginManager, diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskCallbackImpl.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskCallbackImpl.java index 506325e485..b286902772 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskCallbackImpl.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskCallbackImpl.java @@ -33,11 +33,11 @@ public class TaskCallbackImpl implements TaskCallBack { private final WorkerMessageSender workerMessageSender; - private final String masterAddress; + private final TaskExecutionContext taskExecutionContext; - public TaskCallbackImpl(WorkerMessageSender workerMessageSender, String masterAddress) { + public TaskCallbackImpl(WorkerMessageSender workerMessageSender, TaskExecutionContext taskExecutionContext) { this.workerMessageSender = workerMessageSender; - this.masterAddress = masterAddress; + this.taskExecutionContext = taskExecutionContext; } @Override @@ -52,7 +52,7 @@ public class TaskCallbackImpl implements TaskCallBack { log.info("send remote application info {}", applicationInfo); taskExecutionContext.setAppIds(applicationInfo.getAppIds()); - workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_EXECUTE_RUNNING); + workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_EXECUTE_RUNNING); } @Override @@ -64,7 +64,7 @@ public class TaskCallbackImpl implements TaskCallBack { return; } - workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_UPDATE_PID); + workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_UPDATE_PID); } } diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java index 7fdfc405ef..bed6ab4d90 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java @@ -36,14 +36,12 @@ public abstract class WorkerDelayTaskExecuteRunnable extends WorkerTaskExecuteRu protected WorkerDelayTaskExecuteRunnable(@NonNull TaskExecutionContext taskExecutionContext, @NonNull WorkerConfig workerConfig, - @NonNull String masterAddress, @NonNull WorkerMessageSender workerMessageSender, @NonNull WorkerRpcClient workerRpcClient, @NonNull TaskPluginManager taskPluginManager, @Nullable StorageOperate storageOperate) { super(taskExecutionContext, workerConfig, - masterAddress, workerMessageSender, workerRpcClient, taskPluginManager, diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java index 9144a2f2e0..c16dff395b 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java @@ -34,7 +34,6 @@ public abstract class WorkerDelayTaskExecuteRunnableFactory createWorkerDelayTaskExecuteRunnableFactory(@NonNull TaskExecutionContext taskExecutionContext, @NonNull WorkerConfig workerConfig, - @NonNull String workflowMasterAddress, @NonNull WorkerMessageSender workerMessageSender, @NonNull WorkerRpcClient workerRpcClient, @NonNull TaskPluginManager taskPluginManager, @Nullable StorageOperate storageOperate) { return new DefaultWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext, workerConfig, - workflowMasterAddress, workerMessageSender, workerRpcClient, taskPluginManager, diff --git a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessorTest.java b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessorTest.java index 26c0dc2806..9276b7b3c4 100644 --- a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessorTest.java +++ b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessorTest.java @@ -73,7 +73,7 @@ public class TaskDispatchProcessorTest { Mockito.verify(workerManagerThread, Mockito.atMostOnce()).offer(Mockito.any()); Mockito.verify(workerMessageSender, Mockito.never()).sendMessageWithRetry(taskExecutionContext, - "localhost:5678", CommandType.TASK_REJECT); + CommandType.TASK_REJECT); } public Command createDispatchCommand(TaskExecutionContext taskExecutionContext) { diff --git a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java index 6ebdf6f3ec..74b8d122d4 100644 --- a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java +++ b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java @@ -57,7 +57,6 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest { WorkerTaskExecuteRunnable workerTaskExecuteRunnable = new DefaultWorkerDelayTaskExecuteRunnable( taskExecutionContext, workerConfig, - masterAddress, workerMessageSender, alertClientService, taskPluginManager, @@ -82,7 +81,6 @@ public class DefaultWorkerDelayTaskExecuteRunnableTest { WorkerTaskExecuteRunnable workerTaskExecuteRunnable = new DefaultWorkerDelayTaskExecuteRunnable( taskExecutionContext, workerConfig, - masterAddress, workerMessageSender, alertClientService, taskPluginManager,