natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Take over task instance in master failover (#13798)

This commit is contained in:
Wenjun Ruan 2023-03-28 16:02:57 +08:00 committed by GitHub
parent 21ca8ddf03
commit d1b6e6f02c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
38 changed files with 360 additions and 440 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java
View File

@ -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;
}

49
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterRpcClient.java Normal file
View File

@ -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);
}
}

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerBootstrap.java
View File

@ -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,

1
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StreamTaskExecuteRunnable.java
View File

@ -349,6 +349,7 @@ public class StreamTaskExecuteRunnable implements Runnable {
.taskType(taskInstance.getTaskType()).taskParams(taskInstance.getTaskParams()).build());
Map<String, Property> propertyMap = paramParsingPreparation(taskInstance, baseParam);
TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get()
.buildWorkflowInstanceHost(masterConfig.getMasterAddress())
.buildTaskInstanceRelatedInfo(taskInstance)
.buildTaskDefinitionRelatedInfo(taskDefinition)
.buildResourceParametersInfo(resources)

97
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

@ -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<WorkflowSubmitStatue> {
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<WorkflowSubmitStatue> {
* @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<WorkflowSubmitStatue> {
@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<WorkflowSubmitStatue> {
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<WorkflowSubmitStatue> {
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<WorkflowSubmitStatue> {
}
}
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<WorkflowSubmitStatue> {
TaskNode taskNodeObject = dag.getNode(taskNode);
Optional<TaskInstance> 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<WorkflowSubmitStatue> {
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
*

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

@ -341,6 +341,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
Map<String, Property> propertyMap =
curingParamsService.paramParsingPreparation(taskInstance, baseParam, processInstance);
return TaskExecutionContextBuilder.get()
.buildWorkflowInstanceHost(masterConfig.getMasterAddress())
.buildTaskInstanceRelatedInfo(taskInstance)
.buildTaskDefinitionRelatedInfo(taskInstance.getTaskDefine())
.buildProcessInstanceRelatedInfo(taskInstance.getProcessInstance())

140
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/MasterFailoverService.java
View File

@ -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<ProcessDefinition> processDefinitions =
processDefinitionDao.queryProcessDefinitionsByCodesAndVersions(needFailoverProcessInstanceList);
Map<Long, ProcessDefinition> 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<TaskInstance> 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
* <p>
* 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<Date> 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;
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/service/WorkerFailoverService.java
View File

@ -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);

1
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutionContextTestUtils.java
View File

@ -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)

6
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java
View File

@ -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");

21
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/service/FailoverServiceTest.java
View File

@ -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

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

@ -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

83
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateResponseCommand.java
View File

@ -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
+ '}';
}
}

18
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/HostUpdateCommand.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeRequest.java
View File

@ -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;
}
}

42
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/processor/HostUpdateResponseProcessor.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/task/WorkflowHostChangeResponse.java
View File

@ -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;
}
}

10
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
View File

@ -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()));

2
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
View File

@ -69,6 +69,8 @@ public class TaskExecutionContext implements Serializable {
*/
private String taskType;
private String workflowInstanceHost;
/**
* host
*/

2
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContextCacheManager.java
View File

@ -30,7 +30,7 @@ public class TaskExecutionContextCacheManager {
/**
* taskInstance cache
*/
private static Map<Integer, TaskExecutionContext> taskRequestContextCache = new ConcurrentHashMap<>();
private static final Map<Integer, TaskExecutionContext> taskRequestContextCache = new ConcurrentHashMap<>();
/**
* get taskInstance by taskInstance id

55
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/log/TaskInstanceLogHeader.java
View File

@ -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<String> INITIALIZE_TASK_CONTEXT_HEADER = Lists.newArrayList(
"***********************************************************************************************",
"********************************* Initialize task context ***********************************",
"***********************************************************************************************");
private static final List<String> 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<String> 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<String> 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);
}
}

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageSender.java
View File

@ -34,7 +34,7 @@ public interface MessageSender<T extends BaseCommand> {
/**
* 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.

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteResultMessageSender.java
View File

@ -42,11 +42,11 @@ public class TaskExecuteResultMessageSender implements MessageSender<TaskExecute
workerRpcClient.send(Host.of(message.getMessageReceiverAddress()), message.convert2Command());
}
public TaskExecuteResultCommand buildMessage(TaskExecutionContext taskExecutionContext,
String messageReceiverAddress) {
@Override
public TaskExecuteResultCommand buildMessage(TaskExecutionContext taskExecutionContext) {
TaskExecuteResultCommand taskExecuteResultMessage =
new TaskExecuteResultCommand(workerConfig.getWorkerAddress(),
messageReceiverAddress,
taskExecutionContext.getWorkflowInstanceHost(),
System.currentTimeMillis());
taskExecuteResultMessage.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());
taskExecuteResultMessage.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());

6
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskExecuteRunningMessageSender.java
View File

@ -44,11 +44,11 @@ public class TaskExecuteRunningMessageSender implements MessageSender<TaskExecut
workerRpcClient.send(Host.of(message.getMessageReceiverAddress()), message.convert2Command());
}
public TaskExecuteRunningCommand buildMessage(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull String messageReceiverAddress) {
@Override
public TaskExecuteRunningCommand buildMessage(@NonNull TaskExecutionContext taskExecutionContext) {
TaskExecuteRunningCommand taskExecuteRunningMessage =
new TaskExecuteRunningCommand(workerConfig.getWorkerAddress(),
messageReceiverAddress,
taskExecutionContext.getWorkflowInstanceHost(),
System.currentTimeMillis());
taskExecuteRunningMessage.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
taskExecuteRunningMessage.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskRejectMessageSender.java
View File

@ -42,9 +42,10 @@ public class TaskRejectMessageSender implements MessageSender<TaskRejectCommand>
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());

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskUpdatePidMessageSender.java
View File

@ -45,11 +45,10 @@ public class TaskUpdatePidMessageSender implements MessageSender<TaskUpdatePidCo
}
@Override
public TaskUpdatePidCommand buildMessage(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull String messageReceiverAddress) {
public TaskUpdatePidCommand buildMessage(@NonNull TaskExecutionContext taskExecutionContext) {
TaskUpdatePidCommand taskUpdatePidCommand =
new TaskUpdatePidCommand(workerConfig.getWorkerAddress(),
messageReceiverAddress,
taskExecutionContext.getWorkflowInstanceHost(),
System.currentTimeMillis());
taskUpdatePidCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
taskUpdatePidCommand.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());

58
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/HostUpdateProcessor.java
View File

@ -1,58 +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.server.worker.processor;
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.command.HostUpdateCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.google.common.base.Preconditions;
import io.netty.channel.Channel;
/**
* update process host
* this used when master failover
*/
@Component
@Slf4j
public class HostUpdateProcessor implements NettyRequestProcessor {
@Autowired
private MessageRetryRunner messageRetryRunner;
@Override
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.PROCESS_HOST_UPDATE_REQUEST == command.getType(),
String.format("invalid command type : %s", command.getType()));
HostUpdateCommand updateCommand = JSONUtils.parseObject(command.getBody(), HostUpdateCommand.class);
if (updateCommand == null) {
log.error("host update command is null");
return;
}
log.info("received host update command : {}", updateCommand);
messageRetryRunner.updateMessageHost(updateCommand.getTaskInstanceId(), updateCommand.getProcessHost());
}
}

7
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessor.java
View File

@ -111,15 +111,13 @@ public class TaskDispatchProcessor implements NettyRequestProcessor {
if (remainTime > 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());

91
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkflowHostChangeProcessor.java Normal file
View File

@ -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");
}
});
}
}
}

9
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerMessageSender.java
View File

@ -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) {

7
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerRpcServer.java
View File

@ -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);

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnable.java
View File

@ -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,

3
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableFactory.java
View File

@ -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,

10
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskCallbackImpl.java
View File

@ -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);
}
}

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnable.java
View File

@ -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,

3
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerDelayTaskExecuteRunnableFactory.java
View File

@ -34,7 +34,6 @@ public abstract class WorkerDelayTaskExecuteRunnableFactory<T extends WorkerDela
protected final @NonNull TaskExecutionContext taskExecutionContext;
protected final @NonNull WorkerConfig workerConfig;
protected final @NonNull String workflowMasterAddress;
protected final @NonNull WorkerMessageSender workerMessageSender;
protected final @NonNull WorkerRpcClient workerRpcClient;
protected final @NonNull TaskPluginManager taskPluginManager;
@ -43,14 +42,12 @@ public abstract class WorkerDelayTaskExecuteRunnableFactory<T extends WorkerDela
protected WorkerDelayTaskExecuteRunnableFactory(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull String workflowMasterAddress,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull WorkerRpcClient workerRpcClient,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.workflowMasterAddress = workflowMasterAddress;
this.workerMessageSender = workerMessageSender;
this.workerRpcClient = workerRpcClient;
this.taskPluginManager = taskPluginManager;

23
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java
View File

@ -68,7 +68,6 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
protected final TaskExecutionContext taskExecutionContext;
protected final WorkerConfig workerConfig;
protected final String masterAddress;
protected final WorkerMessageSender workerMessageSender;
protected final TaskPluginManager taskPluginManager;
protected final @Nullable StorageOperate storageOperate;
@ -79,14 +78,12 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
protected WorkerTaskExecuteRunnable(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull String masterAddress,
@NonNull WorkerMessageSender workerMessageSender,
@NonNull WorkerRpcClient workerRpcClient,
@NonNull TaskPluginManager taskPluginManager,
@Nullable StorageOperate storageOperate) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.masterAddress = masterAddress;
this.workerMessageSender = workerMessageSender;
this.workerRpcClient = workerRpcClient;
this.taskPluginManager = taskPluginManager;
@ -114,7 +111,7 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.FAILURE);
taskExecutionContext.setEndTime(System.currentTimeMillis());
workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_EXECUTE_RESULT);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_EXECUTE_RESULT);
log.info(
"Get a exception when execute the task, will send the task execute result to master, the current task execute result is {}",
TaskExecutionStatus.FAILURE);
@ -142,32 +139,30 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
taskExecutionContext.getProcessInstanceId(), taskExecutionContext.getTaskInstanceId());
final LogUtils.MDCAutoClosableContext mdcAutoClosableContext1 =
LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath())) {
log.info("\n{}", TaskInstanceLogHeader.INITIALIZE_TASK_CONTEXT_HEADER);
TaskInstanceLogHeader.printInitializeTaskContextHeader();
initializeTask();
if (DRY_RUN_FLAG_YES == taskExecutionContext.getDryRun()) {
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.SUCCESS);
taskExecutionContext.setEndTime(System.currentTimeMillis());
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress,
CommandType.TASK_EXECUTE_RESULT);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_EXECUTE_RESULT);
log.info(
"The current execute mode is dry run, will stop the subsequent process and set the taskInstance status to success");
return;
}
log.info("\n{}", TaskInstanceLogHeader.LOAD_TASK_INSTANCE_PLUGIN_HEADER);
TaskInstanceLogHeader.printLoadTaskInstancePluginHeader();
beforeExecute();
TaskCallBack taskCallBack = TaskCallbackImpl.builder()
.workerMessageSender(workerMessageSender)
.masterAddress(masterAddress)
.taskExecutionContext(taskExecutionContext)
.build();
log.info("\n{}", TaskInstanceLogHeader.EXECUTE_TASK_HEADER);
TaskInstanceLogHeader.printExecuteTaskHeader();
executeTask(taskCallBack);
log.info("\n{}", TaskInstanceLogHeader.FINALIZE_TASK_HEADER);
TaskInstanceLogHeader.printFinalizeTaskHeader();
afterExecute();
closeLogAppender();
} catch (Throwable ex) {
@ -194,7 +189,7 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
protected void beforeExecute() {
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.RUNNING_EXECUTION);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_EXECUTE_RUNNING);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_EXECUTE_RUNNING);
log.info("Set task status to {}", TaskExecutionStatus.RUNNING_EXECUTION);
TaskExecutionCheckerUtils.checkTenantExist(workerConfig, taskExecutionContext);
@ -258,7 +253,7 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable {
taskExecutionContext.setVarPool(JSONUtils.toJsonString(task.getParameters().getVarPool()));
// upload out files and modify the "OUT FILE" property in VarPool
TaskFilesTransferUtils.uploadOutputFiles(taskExecutionContext, storageOperate);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_EXECUTE_RESULT);
workerMessageSender.sendMessageWithRetry(taskExecutionContext, CommandType.TASK_EXECUTE_RESULT);
log.info("Send task execute result to master, the current task status: {}",
taskExecutionContext.getCurrentExecutionStatus());

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnableFactoryBuilder.java
View File

@ -34,14 +34,12 @@ public class WorkerTaskExecuteRunnableFactoryBuilder {
public static 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,

2
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskDispatchProcessorTest.java
View File

@ -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) {

2
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerDelayTaskExecuteRunnableTest.java
View File

@ -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,