diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java index 7bff4e9d21..f073fd55f3 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java @@ -128,12 +128,14 @@ public class TaskPriorityQueueConsumer extends BaseDaemonThread { List failedDispatchTasks = this.batchDispatch(fetchTaskNum); if (CollectionUtils.isNotEmpty(failedDispatchTasks)) { + logger.info("{} tasks dispatch failed, will retry to dispatch", failedDispatchTasks.size()); TaskMetrics.incTaskDispatchFailed(failedDispatchTasks.size()); for (TaskPriority dispatchFailedTask : failedDispatchTasks) { taskPriorityQueue.put(dispatchFailedTask); } // If the all task dispatch failed, will sleep for 1s to avoid the master cpu higher. if (fetchTaskNum == failedDispatchTasks.size()) { + logger.info("All tasks dispatch failed, will sleep a while to avoid the master cpu higher"); TimeUnit.MILLISECONDS.sleep(Constants.SLEEP_TIME_MILLIS); } } @@ -209,6 +211,8 @@ public class TaskPriorityQueueConsumer extends BaseDaemonThread { if (isTaskNeedToCheck(taskPriority)) { if (taskInstanceIsFinalState(taskPriority.getTaskId())) { // when task finish, ignore this task, there is no need to dispatch anymore + logger.info("Task {} is already finished, no need to dispatch, task instance id: {}", + taskInstance.getName(), taskInstance.getId()); return true; } } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java index 25b8037dd6..62da82156f 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/executor/NettyExecutorManager.java @@ -114,7 +114,7 @@ public class NettyExecutorManager extends AbstractExecutorManager { // is not belongs to the down worker ISSUE-10842. context.getTaskInstance().setHost(host.getAddress()); } catch (ExecuteException ex) { - logger.error(String.format("execute command : %s error", command), ex); + logger.error("Execute command {} error", command, ex); try { failNodeSet.add(host.getAddress()); Set tmpAllIps = new HashSet<>(allNodes); @@ -156,7 +156,7 @@ public class NettyExecutorManager extends AbstractExecutorManager { nettyRemotingClient.send(host, command); success = true; } catch (Exception ex) { - logger.error(String.format("send command : %s to %s error", command, host), ex); + logger.error("Send command to {} error, command: {}", host, command, ex); retryCount--; ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS); } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskRetryStateEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskRetryStateEventHandler.java index 1ac198497f..1d5f7a3883 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskRetryStateEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskRetryStateEventHandler.java @@ -22,17 +22,23 @@ import org.apache.dolphinscheduler.common.enums.StateEventType; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics; import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Map; @AutoService(StateEventHandler.class) public class TaskRetryStateEventHandler implements StateEventHandler { + private static final Logger logger = LoggerFactory.getLogger(TaskRetryStateEventHandler.class); + @Override public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable, StateEvent stateEvent) throws StateEventHandleException { TaskStateEvent taskStateEvent = (TaskStateEvent) stateEvent; + logger.info("Handle task instance retry event, taskInstanceId: {}", taskStateEvent.getTaskInstanceId()); + TaskMetrics.incTaskInstanceByState("retry"); Map waitToRetryTaskInstanceMap = workflowExecuteRunnable.getWaitToRetryTaskInstanceMap(); TaskInstance taskInstance = waitToRetryTaskInstanceMap.get(taskStateEvent.getTaskCode()); diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskStateEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskStateEventHandler.java index aab5d2d6a2..935fa9db37 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskStateEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskStateEventHandler.java @@ -54,6 +54,10 @@ public class TaskStateEventHandler implements StateEventHandler { throw new StateEventHandleError("Task state event handle error due to task state is null"); } + logger.info( + "Handle task instance state event, the current task instance state {} will be changed to {}", + task.getState(), taskStateEvent.getStatus()); + Map completeTaskMap = workflowExecuteRunnable.getCompleteTaskMap(); if (task.getState().isFinished()) { diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskTimeoutStateEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskTimeoutStateEventHandler.java index 1678594323..9efe773b19 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskTimeoutStateEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskTimeoutStateEventHandler.java @@ -50,6 +50,8 @@ public class TaskTimeoutStateEventHandler implements StateEventHandler { "Cannot find the task instance from workflow execute runnable, taskInstanceId: %s", taskStateEvent.getTaskInstanceId()))); + logger.info("Handle task instance state timout event, taskInstanceId: {}", taskStateEvent.getTaskInstanceId()); + if (TimeoutFlag.CLOSE == taskInstance.getTaskDefine().getTimeoutFlag()) { return true; } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskWaitTaskGroupStateHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskWaitTaskGroupStateHandler.java index 9a3c59a949..8dcca91d89 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskWaitTaskGroupStateHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskWaitTaskGroupStateHandler.java @@ -21,11 +21,17 @@ import org.apache.dolphinscheduler.common.enums.StateEventType; import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable; import com.google.auto.service.AutoService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; @AutoService(StateEventHandler.class) public class TaskWaitTaskGroupStateHandler implements StateEventHandler { + + private static final Logger logger = LoggerFactory.getLogger(TaskWaitTaskGroupStateHandler.class); + @Override public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable, StateEvent stateEvent) { + logger.info("Handle task instance wait task group event, taskInstanceId: {}", stateEvent.getTaskInstanceId()); return workflowExecuteRunnable.checkForceStartAndWakeUp(stateEvent); } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowBlockStateEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowBlockStateEventHandler.java index f7349fcbd1..cac30a8072 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowBlockStateEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowBlockStateEventHandler.java @@ -38,6 +38,7 @@ public class WorkflowBlockStateEventHandler implements StateEventHandler { @Override public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable, StateEvent stateEvent) throws StateEventHandleError { + logger.info("Handle workflow instance state block event"); Optional taskInstanceOptional = workflowExecuteRunnable.getTaskInstance(stateEvent.getTaskInstanceId()); if (!taskInstanceOptional.isPresent()) { diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowTimeoutStateEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowTimeoutStateEventHandler.java index b04866a76a..4a9fd99392 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowTimeoutStateEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowTimeoutStateEventHandler.java @@ -22,11 +22,17 @@ import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics; import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable; import com.google.auto.service.AutoService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; @AutoService(StateEventHandler.class) public class WorkflowTimeoutStateEventHandler implements StateEventHandler { + + private static final Logger logger = LoggerFactory.getLogger(WorkflowTimeoutStateEventHandler.class); + @Override public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable, StateEvent stateEvent) { + logger.info("Handle workflow instance timeout event"); ProcessInstanceMetrics.incProcessInstanceByState("timeout"); workflowExecuteRunnable.processTimeout(); return true; 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 6a3a12e1e4..351ebe7c17 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 @@ -136,12 +136,14 @@ public class MasterSchedulerBootstrap extends BaseDaemonThread implements AutoCl try { if (!ServerLifeCycleManager.isRunning()) { // the current server is not at running status, cannot consume command. + logger.warn("The current server {} is not at running status, cannot consumes commands.", this.masterAddress); Thread.sleep(Constants.SLEEP_TIME_MILLIS); } // todo: if the workflow event queue is much, we need to handle the back pressure boolean isOverload = OSUtils.isOverload(masterConfig.getMaxCpuLoadAvg(), masterConfig.getReservedMemory()); if (isOverload) { + logger.warn("The current server {} is overload, cannot consumes commands.", this.masterAddress); MasterServerMetrics.incMasterOverload(); Thread.sleep(Constants.SLEEP_TIME_MILLIS); continue; diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java index 2a19b1ca8d..8a4b195a5b 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java @@ -120,13 +120,13 @@ public class StateWheelExecuteThread extends BaseDaemonThread { public void addProcess4TimeoutCheck(ProcessInstance processInstance) { processInstanceTimeoutCheckList.add(processInstance.getId()); - logger.info("Success add workflow instance into timeout check list"); + logger.info("Success add workflow instance {} into timeout check list", processInstance.getId()); } public void removeProcess4TimeoutCheck(int processInstanceId) { boolean removeFlag = processInstanceTimeoutCheckList.remove(processInstanceId); if (removeFlag) { - logger.info("Success remove workflow instance from timeout check list"); + logger.info("Success remove workflow instance {} from timeout check list", processInstanceId); } } @@ -154,7 +154,7 @@ public class StateWheelExecuteThread extends BaseDaemonThread { (long) processInstance.getTimeout() * Constants.SEC_2_MINUTES_TIME_UNIT); if (timeRemain < 0) { - logger.info("Workflow instance timeout, adding timeout event"); + logger.info("Workflow instance {} timeout, adding timeout event", processInstance.getId()); addProcessTimeoutEvent(processInstance); processInstanceTimeoutCheckList.remove(processInstance.getId()); logger.info("Workflow instance timeout, added timeout event"); 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 258eefe9af..90c85cf20e 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 @@ -455,7 +455,7 @@ public class WorkflowExecuteRunnable implements Callable { } TaskInstance newTaskInstance = cloneRetryTaskInstance(taskInstance); if (newTaskInstance == null) { - logger.error("retry fail, new taskInstance is null, task code:{}, task id:{}", + logger.error("Retry task fail because new taskInstance is null, task code:{}, task id:{}", taskInstance.getTaskCode(), taskInstance.getId()); return; @@ -463,7 +463,7 @@ public class WorkflowExecuteRunnable implements Callable { waitToRetryTaskInstanceMap.put(newTaskInstance.getTaskCode(), newTaskInstance); if (!taskInstance.retryTaskIntervalOverTime()) { logger.info( - "failure task will be submitted: process id: {}, task instance code: {} state:{} retry times:{} / {}, interval:{}", + "Failure task will be submitted, process id: {}, task instance code: {}, state: {}, retry times: {} / {}, interval: {}", processInstance.getId(), newTaskInstance.getTaskCode(), newTaskInstance.getState(), newTaskInstance.getRetryTimes(), newTaskInstance.getMaxRetryTimes(), newTaskInstance.getRetryInterval()); @@ -785,7 +785,7 @@ public class WorkflowExecuteRunnable implements Callable { ProcessDag processDag = generateFlowDag(taskNodeList, startNodeNameList, recoveryNodeCodeList, processInstance.getTaskDependType()); if (processDag == null) { - logger.error("processDag is null"); + logger.error("ProcessDag is null"); return; } // generate process dag @@ -818,6 +818,8 @@ public class WorkflowExecuteRunnable implements Callable { task.getTaskCode(), task.getState()); if (validTaskMap.containsKey(task.getTaskCode())) { + logger.warn("Have same taskCode taskInstance when init task queue, need to check taskExecutionStatus, taskCode:{}", + task.getTaskCode()); int oldTaskInstanceId = validTaskMap.get(task.getTaskCode()); TaskInstance oldTaskInstance = taskInstanceMap.get(oldTaskInstanceId); if (!oldTaskInstance.getState().isFinished() && task.getState().isFinished()) { @@ -825,14 +827,13 @@ public class WorkflowExecuteRunnable implements Callable { processService.updateTaskInstance(task); continue; } - logger.warn("have same taskCode taskInstance when init task queue, taskCode:{}", - task.getTaskCode()); } validTaskMap.put(task.getTaskCode(), task.getId()); taskInstanceMap.put(task.getId(), task); if (task.isTaskComplete()) { + logger.info("TaskInstance is already complete."); completeTaskMap.put(task.getTaskCode(), task.getId()); continue; } @@ -842,6 +843,7 @@ public class WorkflowExecuteRunnable implements Callable { } if (task.taskCanRetry()) { if (task.getState().isNeedFaultTolerance()) { + logger.info("TaskInstance needs fault tolerance, will be added to standby list."); task.setFlag(Flag.NO); processService.updateTaskInstance(task); @@ -849,6 +851,7 @@ public class WorkflowExecuteRunnable implements Callable { TaskInstance tolerantTaskInstance = cloneTolerantTaskInstance(task); addTaskToStandByList(tolerantTaskInstance); } else { + logger.info("Retry taskInstance, taskState: {}", task.getState()); retryTaskInstance(task); } continue; @@ -930,15 +933,14 @@ public class WorkflowExecuteRunnable implements Callable { boolean submit = taskProcessor.action(TaskAction.SUBMIT); if (!submit) { - logger.error("process id:{} name:{} submit standby task id:{} name:{} failed!", - processInstance.getId(), - processInstance.getName(), - taskInstance.getId(), + logger.error("Submit standby task failed!, taskCode: {}, taskName: {}", + taskInstance.getTaskCode(), taskInstance.getName()); return Optional.empty(); } // in a dag, only one taskInstance is valid per taskCode, so need to set the old taskInstance invalid + LoggerUtils.setWorkflowAndTaskInstanceIDMDC(taskInstance.getProcessInstanceId(), taskInstance.getId()); if (validTaskMap.containsKey(taskInstance.getTaskCode())) { int oldTaskInstanceId = validTaskMap.get(taskInstance.getTaskCode()); if (taskInstance.getId() != oldTaskInstanceId) { @@ -965,19 +967,16 @@ public class WorkflowExecuteRunnable implements Callable { taskInstance.getProcessInstanceId(), taskInstance.getTaskGroupPriority()); if (!acquireTaskGroup) { - logger.info("submit task name :{}, but the first time to try to acquire task group failed", - taskInstance.getName()); + logger.info("Submitted task will not be dispatch right now because the first time to try to acquire" + + " task group failed, taskInstanceName: {}, taskGroupId: {}", + taskInstance.getName(), taskGroupId); return Optional.of(taskInstance); } } boolean dispatchSuccess = taskProcessor.action(TaskAction.DISPATCH); if (!dispatchSuccess) { - logger.error("process id:{} name:{} dispatch standby task id:{} name:{} failed!", - processInstance.getId(), - processInstance.getName(), - taskInstance.getId(), - taskInstance.getName()); + logger.error("Dispatch standby process {} task {} failed", processInstance.getName(), taskInstance.getName()); return Optional.empty(); } taskProcessor.action(TaskAction.RUN); @@ -1005,11 +1004,11 @@ public class WorkflowExecuteRunnable implements Callable { } return Optional.of(taskInstance); } catch (Exception e) { - logger.error("submit standby task error, taskCode: {}, taskInstanceId: {}", - taskInstance.getTaskCode(), - taskInstance.getId(), - e); + logger.error("Submit standby task {} error, taskCode: {}", taskInstance.getName(), + taskInstance.getTaskCode(), e); return Optional.empty(); + } finally { + LoggerUtils.removeWorkflowAndTaskInstanceIdMDC(); } } @@ -1072,7 +1071,8 @@ public class WorkflowExecuteRunnable implements Callable { public TaskInstance cloneRetryTaskInstance(TaskInstance taskInstance) { TaskNode taskNode = dag.getNode(Long.toString(taskInstance.getTaskCode())); if (taskNode == null) { - logger.error("taskNode is null, code:{}", taskInstance.getTaskCode()); + logger.error("Clone retry taskInstance error because taskNode is null, taskCode:{}", + taskInstance.getTaskCode()); return null; } TaskInstance newTaskInstance = newTaskInstance(processInstance, taskNode); @@ -1099,7 +1099,8 @@ public class WorkflowExecuteRunnable implements Callable { public TaskInstance cloneTolerantTaskInstance(TaskInstance taskInstance) { TaskNode taskNode = dag.getNode(Long.toString(taskInstance.getTaskCode())); if (taskNode == null) { - logger.error("taskNode is null, code:{}", taskInstance.getTaskCode()); + logger.error("Clone tolerant taskInstance error because taskNode is null, taskCode:{}", + taskInstance.getTaskCode()); return null; } TaskInstance newTaskInstance = newTaskInstance(processInstance, taskNode); @@ -1340,11 +1341,11 @@ public class WorkflowExecuteRunnable implements Callable { } if (task.getId() != null && completeTaskMap.containsKey(task.getTaskCode())) { - logger.info("task {} has already run success", task.getName()); + logger.info("Task has already run success, taskName: {}", task.getName()); continue; } if (task.getState().isKill()) { - logger.info("task {} stopped, the state is {}", task.getName(), task.getState()); + logger.info("Task is be stopped, the state is {}, taskInstanceId: {}", task.getState(), task.getId()); continue; } @@ -1396,8 +1397,8 @@ public class WorkflowExecuteRunnable implements Callable { } } } - logger.info("taskCode: {} completeDependTaskList: {}", taskCode, - Arrays.toString(completeTaskMap.keySet().toArray())); + logger.info("The dependTasks of task all success, currentTaskCode: {}, dependTaskCodes: {}", + taskCode, Arrays.toString(completeTaskMap.keySet().toArray())); return DependResult.SUCCESS; } @@ -1429,6 +1430,9 @@ public class WorkflowExecuteRunnable implements Callable { List nextTaskList = DagHelper.parseConditionTask(dependNodeName, skipTaskNodeMap, dag, getCompleteTaskInstanceMap()); if (!nextTaskList.contains(nextNodeName)) { + logger.info("DependTask is a condition task, and its next condition branch does not hava current task, " + + "dependTaskCode: {}, currentTaskCode: {}", dependNodeName, nextNodeName + ); return false; } } else { @@ -1710,10 +1714,11 @@ public class WorkflowExecuteRunnable implements Callable { */ public void addTaskToStandByList(TaskInstance taskInstance) { if (readyToSubmitTaskQueue.contains(taskInstance)) { - logger.warn("task was found in ready submit queue, task code:{}", taskInstance.getTaskCode()); + logger.warn("Task already exists in ready submit queue, no need to add again, task code:{}", + taskInstance.getTaskCode()); return; } - logger.info("add task to stand by list, task name:{}, task id:{}, task code:{}", + logger.info("Add task to stand by list, task name:{}, task id:{}, task code:{}", taskInstance.getName(), taskInstance.getId(), taskInstance.getTaskCode()); @@ -1798,8 +1803,8 @@ public class WorkflowExecuteRunnable implements Callable { TaskInstance retryTask = processService.findTaskInstanceById(task.getId()); if (retryTask != null && retryTask.getState().isForceSuccess()) { task.setState(retryTask.getState()); - logger.info("task: {} has been forced success, put it into complete task list and stop retrying", - task.getName()); + logger.info("Task {} has been forced success, put it into complete task list and stop retrying, taskInstanceId: {}", + task.getName(), task.getId()); removeTaskFromStandbyList(task); completeTaskMap.put(task.getTaskCode(), task.getId()); taskInstanceMap.put(task.getId(), task); @@ -1815,6 +1820,7 @@ public class WorkflowExecuteRunnable implements Callable { } DependResult dependResult = getDependResultForTask(task); if (DependResult.SUCCESS == dependResult) { + logger.info("The dependResult of task {} is success, so ready to submit to execute", task.getName()); Optional taskInstanceOptional = submitTaskExec(task); if (!taskInstanceOptional.isPresent()) { this.taskFailedSubmit = true; diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThreadPool.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThreadPool.java index 5d0e237027..84e784e6c9 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThreadPool.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThreadPool.java @@ -158,10 +158,17 @@ public class WorkflowExecuteThreadPool extends ThreadPoolTaskExecutor { ProcessInstance processInstance = entry.getKey(); TaskInstance taskInstance = entry.getValue(); String address = NetUtils.getAddr(masterConfig.getListenPort()); - if (processInstance.getHost().equalsIgnoreCase(address)) { - this.notifyMyself(processInstance, taskInstance); - } else { - this.notifyProcess(finishProcessInstance, processInstance, taskInstance); + try { + LoggerUtils.setWorkflowAndTaskInstanceIDMDC(processInstance.getId(), taskInstance.getId()); + if (processInstance.getHost().equalsIgnoreCase(address)) { + logger.info("Process host is local master, will notify it"); + this.notifyMyself(processInstance, taskInstance); + } else { + logger.info("Process host is remote master, will notify it"); + this.notifyProcess(finishProcessInstance, processInstance, taskInstance); + } + } finally { + LoggerUtils.removeWorkflowAndTaskInstanceIdMDC(); } } } @@ -190,8 +197,8 @@ public class WorkflowExecuteThreadPool extends ThreadPoolTaskExecutor { TaskInstance taskInstance) { String processInstanceHost = processInstance.getHost(); if (Strings.isNullOrEmpty(processInstanceHost)) { - logger.error("process {} host is empty, cannot notify task {} now", processInstance.getId(), - taskInstance.getId()); + logger.error("Process {} host is empty, cannot notify task {} now, taskId: {}", processInstance.getName(), + taskInstance.getName(), taskInstance.getId()); return; } WorkflowStateEventChangeCommand workflowStateEventChangeCommand = new WorkflowStateEventChangeCommand( 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 a67ee628ea..66a5b493a1 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 @@ -304,6 +304,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor { // verify tenant is null if (verifyTenantIsNull(tenant, taskInstance)) { + logger.info("Task state changes to {}", TaskExecutionStatus.FAILURE); taskInstance.setState(TaskExecutionStatus.FAILURE); processService.saveTaskInstance(taskInstance); return null; @@ -418,7 +419,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor { int ruleId = dataQualityParameters.getRuleId(); DqRule dqRule = processService.getDqRule(ruleId); if (dqRule == null) { - logger.error("can not get DqRule by id {}", ruleId); + logger.error("Can not get dataQuality rule by id {}", ruleId); return; } @@ -428,7 +429,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor { List ruleInputEntryList = processService.getRuleInputEntry(ruleId); if (CollectionUtils.isEmpty(ruleInputEntryList)) { - logger.error("{} rule input entry list is empty ", ruleId); + logger.error("Rule input entry list is empty, ruleId: {}", ruleId); return; } List executeSqlList = processService.getDqExecuteSql(ruleId); @@ -603,9 +604,7 @@ public abstract class BaseTaskProcessor implements ITaskProcessor { */ protected boolean verifyTenantIsNull(Tenant tenant, TaskInstance taskInstance) { if (tenant == null) { - logger.error("tenant not exists,process instance id : {},task instance id : {}", - taskInstance.getProcessInstance().getId(), - taskInstance.getId()); + logger.error("Tenant does not exists"); return true; } return false; diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java index e5413c1490..a3f8bd0b24 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java @@ -92,18 +92,18 @@ public class CommonTaskProcessor extends BaseTaskProcessor { this.initQueue(); } if (taskInstance.getState().isFinished()) { - logger.info("submit task , but task [{}] state [{}] is already finished. ", taskInstance.getName(), - taskInstance.getState()); + logger.info("Task {} has already finished, no need to submit to task queue, taskState: {}", + taskInstance.getName(), taskInstance.getState()); return true; } // task cannot be submitted because its execution state is RUNNING or DELAY. if (taskInstance.getState() == TaskExecutionStatus.RUNNING_EXECUTION || taskInstance.getState() == TaskExecutionStatus.DELAY_EXECUTION) { - logger.info("submit task, but the status of the task {} is already running or delayed.", - taskInstance.getName()); + logger.info("Task {} is already running or delayed, no need to submit to task queue, taskState: {}", + taskInstance.getName(), taskInstance.getState()); return true; } - logger.info("task ready to dispatch to worker: taskInstanceId: {}", taskInstance.getId()); + logger.info("Task {} is ready to dispatch to worker", taskInstance.getName()); TaskPriority taskPriority = new TaskPriority(processInstance.getProcessInstancePriority().getCode(), processInstance.getId(), taskInstance.getProcessInstancePriority().getCode(), @@ -112,17 +112,17 @@ public class CommonTaskProcessor extends BaseTaskProcessor { TaskExecutionContext taskExecutionContext = getTaskExecutionContext(taskInstance); if (taskExecutionContext == null) { - logger.error("task get taskExecutionContext fail: {}", taskInstance); + logger.error("Get taskExecutionContext fail, task: {}", taskInstance); return false; } taskPriority.setTaskExecutionContext(taskExecutionContext); taskUpdateQueue.put(taskPriority); - logger.info("Master submit task to priority queue success, taskInstanceId : {}", taskInstance.getId()); + logger.info("Task {} is submitted to priority queue success by master", taskInstance.getName()); return true; } catch (Exception e) { - logger.error("submit task error", e); + logger.error("Task {} is submitted to priority queue error", taskInstance.getName(), e); return false; } } diff --git a/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/RpcTest.java b/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/RpcTest.java index bd4211ea1a..e216dbc95f 100644 --- a/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/RpcTest.java +++ b/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/RpcTest.java @@ -44,17 +44,6 @@ public class RpcTest { userService = rpcClient.create(IUserService.class, host); } - @Test - public void sendTest() { - Integer result = userService.hi(3); - Assert.assertSame(4, result); - result = userService.hi(4); - Assert.assertSame(5, result); - userService.say("sync"); - userService.callBackIsFalse("async no call back"); - userService.hi(999999); - } - @After public void after() { NettyClient.getInstance().close();