Skip to content

DolphinScheduler
DolphinScheduler

apache / DolphinScheduler
上一次同步 1 年多

通知 706
Star 9572
Fork 3514
DolphinScheduler
DolphinScheduler

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

未验证 提交 28872d87 编写于 作者: W wind 提交者： GitHub
浏览文件
操作

[DS-6891][MasterServer] reduce db operation when process instance running (#6904) 

* remove taskInstanceCacheManager

* remove taskInstanceCacheManager

* [DS-6891][MasterServer] reduce db operation when process instance running

* [DS-6891][MasterServer] reduce db operation when process instance running

* fix test

* fix Transactional method call

* checkstyle
Co-authored-by: Ncaishunfeng <534328519@qq.com>
上级 f2d242c7
展开全部 隐藏空白更改
内联 并排
Showing with 504 addition and 744 deletion
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/TaskInstanceCacheManager.java 已删除 100644 → 0
浏览文件 @ f2d242c7
/*
* 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.cache;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
/**
* task instance state manager
*/
public interface TaskInstanceCacheManager {
/**
* get taskInstance by taskInstance id
*
* @param taskInstanceId taskInstanceId
* @return taskInstance
*/
TaskInstance getByTaskInstanceId(Integer taskInstanceId);
/**
* cache taskInstance
*
* @param taskExecutionContext taskExecutionContext
*/
void cacheTaskInstance(TaskExecutionContext taskExecutionContext);
/**
* cache taskInstance
*
* @param taskAckCommand taskAckCommand
*/
void cacheTaskInstance(TaskExecuteAckCommand taskAckCommand);
/**
* cache taskInstance
*
* @param taskExecuteResponseCommand taskExecuteResponseCommand
*/
void cacheTaskInstance(TaskExecuteResponseCommand taskExecuteResponseCommand);
/**
* remove taskInstance by taskInstanceId
* @param taskInstanceId taskInstanceId
*/
void removeByTaskInstanceId(Integer taskInstanceId);
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/impl/TaskInstanceCacheManagerImpl.java 已删除 100644 → 0
浏览文件 @ f2d242c7
/*
* 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.cache.impl;
import static org.apache.dolphinscheduler.common.Constants.CACHE_REFRESH_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* taskInstance state manager
*/
@Component
public class TaskInstanceCacheManagerImpl implements TaskInstanceCacheManager {
/**
* taskInstance cache
*/
private Map<Integer,TaskInstance> taskInstanceCache = new ConcurrentHashMap<>();
/**
* process service
*/
@Autowired
private ProcessService processService;
/**
* taskInstance cache refresh timer
*/
private Timer refreshTaskInstanceTimer = null;
@PostConstruct
public void init() {
//issue#5539 add thread to fetch task state from database in a fixed rate
this.refreshTaskInstanceTimer = new Timer(true);
refreshTaskInstanceTimer.scheduleAtFixedRate(
new RefreshTaskInstanceTimerTask(), CACHE_REFRESH_TIME_MILLIS, CACHE_REFRESH_TIME_MILLIS
);
}
@PreDestroy
public void close() {
this.refreshTaskInstanceTimer.cancel();
}
/**
* get taskInstance by taskInstance id
*
* @param taskInstanceId taskInstanceId
* @return taskInstance
*/
@Override
public TaskInstance getByTaskInstanceId(Integer taskInstanceId) {
return taskInstanceCache.computeIfAbsent(taskInstanceId, k -> processService.findTaskInstanceById(taskInstanceId));
}
/**
* cache taskInstance
*
* @param taskExecutionContext taskExecutionContext
*/
@Override
public void cacheTaskInstance(TaskExecutionContext taskExecutionContext) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(taskExecutionContext.getTaskInstanceId());
taskInstance.setName(taskExecutionContext.getTaskName());
taskInstance.setStartTime(taskExecutionContext.getStartTime());
taskInstance.setTaskType(taskExecutionContext.getTaskType());
taskInstance.setExecutePath(taskExecutionContext.getExecutePath());
taskInstanceCache.put(taskExecutionContext.getTaskInstanceId(), taskInstance);
}
/**
* cache taskInstance
*
* @param taskAckCommand taskAckCommand
*/
@Override
public void cacheTaskInstance(TaskExecuteAckCommand taskAckCommand) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setState(ExecutionStatus.of(taskAckCommand.getStatus()));
taskInstance.setStartTime(taskAckCommand.getStartTime());
taskInstance.setHost(taskAckCommand.getHost());
taskInstance.setExecutePath(taskAckCommand.getExecutePath());
taskInstance.setLogPath(taskAckCommand.getLogPath());
taskInstanceCache.put(taskAckCommand.getTaskInstanceId(), taskInstance);
}
/**
* cache taskInstance
*
* @param taskExecuteResponseCommand taskExecuteResponseCommand
*/
@Override
public void cacheTaskInstance(TaskExecuteResponseCommand taskExecuteResponseCommand) {
TaskInstance taskInstance = getByTaskInstanceId(taskExecuteResponseCommand.getTaskInstanceId());
taskInstance.setState(ExecutionStatus.of(taskExecuteResponseCommand.getStatus()));
taskInstance.setEndTime(taskExecuteResponseCommand.getEndTime());
taskInstanceCache.put(taskExecuteResponseCommand.getTaskInstanceId(), taskInstance);
}
/**
* remove taskInstance by taskInstanceId
* @param taskInstanceId taskInstanceId
*/
@Override
public void removeByTaskInstanceId(Integer taskInstanceId) {
taskInstanceCache.remove(taskInstanceId);
}
class RefreshTaskInstanceTimerTask extends TimerTask {
@Override
public void run() {
for (Entry<Integer, TaskInstance> taskInstanceEntry : taskInstanceCache.entrySet()) {
TaskInstance taskInstance = processService.findTaskInstanceById(taskInstanceEntry.getKey());
if (null != taskInstance && taskInstance.getState() == ExecutionStatus.NEED_FAULT_TOLERANCE) {
taskInstanceCache.computeIfPresent(taskInstanceEntry.getKey(), (k, v) -> taskInstance);
}
}
}
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
浏览文件 @ 28872d87
......@@ -130,14 +130,16 @@ public class TaskPriorityQueueConsumer extends Thread {
TaskExecutionContext context = taskPriority.getTaskExecutionContext();
ExecutionContext executionContext = new ExecutionContext(context.toCommand(), ExecutorType.WORKER, context.getWorkerGroup());
if (taskInstanceIsFinalState(taskPriority.getTaskId())) {
// when task finish, ignore this task, there is no need to dispatch anymore
return true;
} else {
result = dispatcher.dispatch(executionContext);
if (isTaskNeedToCheck(taskPriority)) {
if (taskInstanceIsFinalState(taskPriority.getTaskId())) {
// when task finish, ignore this task, there is no need to dispatch anymore
return true;
}
}
result = dispatcher.dispatch(executionContext);
} catch (ExecuteException e) {
logger.error("dispatch error: {}", e.getMessage(),e);
logger.error("dispatch error: {}", e.getMessage(), e);
}
return result;
}
......@@ -153,4 +155,18 @@ public class TaskPriorityQueueConsumer extends Thread {
TaskInstance taskInstance = processService.findTaskInstanceById(taskInstanceId);
return taskInstance.getState().typeIsFinished();
}
/**
* check if task need to check state, if true, refresh the checkpoint
* @param taskPriority
* @return
*/
private boolean isTaskNeedToCheck(TaskPriority taskPriority) {
long now = System.currentTimeMillis();
if (now - taskPriority.getCheckpoint() > Constants.SECOND_TIME_MILLIS) {
taskPriority.setCheckpoint(now);
return true;
}
return false;
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java
浏览文件 @ 28872d87
......@@ -24,8 +24,6 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
......@@ -49,14 +47,8 @@ public class TaskAckProcessor implements NettyRequestProcessor {
*/
private final TaskResponseService taskResponseService;
/**
* taskInstance cache manager
*/
private final TaskInstanceCacheManager taskInstanceCacheManager;
public TaskAckProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
}
/**
......@@ -71,8 +63,6 @@ public class TaskAckProcessor implements NettyRequestProcessor {
TaskExecuteAckCommand taskAckCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteAckCommand.class);
logger.info("taskAckCommand : {}", taskAckCommand);
taskInstanceCacheManager.cacheTaskInstance(taskAckCommand);
String workerAddress = ChannelUtils.toAddress(channel).getAddress();
ExecutionStatus ackStatus = ExecutionStatus.of(taskAckCommand.getStatus());
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
浏览文件 @ 28872d87
......@@ -23,8 +23,6 @@ import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
......@@ -48,14 +46,8 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
*/
private final TaskResponseService taskResponseService;
/**
* taskInstance cache manager
*/
private final TaskInstanceCacheManager taskInstanceCacheManager;
public TaskResponseProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
}
/**
......@@ -72,8 +64,6 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
TaskExecuteResponseCommand responseCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteResponseCommand.class);
logger.info("received command : {}", responseCommand);
taskInstanceCacheManager.cacheTaskInstance(responseCommand);
// TaskResponseEvent
TaskResponseEvent taskResponseEvent = TaskResponseEvent.newResult(ExecutionStatus.of(responseCommand.getStatus()),
responseCommand.getEndTime(),
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java
浏览文件 @ 28872d87
......@@ -133,6 +133,15 @@ public class StateEventResponseService {
}
WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecCacheManager.getByProcessInstanceId(stateEvent.getProcessInstanceId());
switch (stateEvent.getType()) {
case TASK_STATE_CHANGE:
workflowExecuteThread.refreshTaskInstance(stateEvent.getTaskInstanceId());
break;
case PROCESS_STATE_CHANGE:
workflowExecuteThread.refreshProcessInstance(stateEvent.getProcessInstanceId());
break;
default:
}
workflowExecuteThread.addStateEvent(stateEvent);
writeResponse(stateEvent, ExecutionStatus.SUCCESS);
} catch (Exception e) {
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseService.java
浏览文件 @ 28872d87
......@@ -142,54 +142,28 @@ public class TaskResponseService {
*/
private void persist(TaskResponseEvent taskResponseEvent) {
Event event = taskResponseEvent.getEvent();
Channel channel = taskResponseEvent.getChannel();
int taskInstanceId = taskResponseEvent.getTaskInstanceId();
int processInstanceId = taskResponseEvent.getProcessInstanceId();
TaskInstance taskInstance;
WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecCacheManager.getByProcessInstanceId(processInstanceId);
if (workflowExecuteThread != null && workflowExecuteThread.checkTaskInstanceById(taskInstanceId)) {
taskInstance = workflowExecuteThread.getTaskInstance(taskInstanceId);
} else {
taskInstance = processService.findTaskInstanceById(taskInstanceId);
}
TaskInstance taskInstance = processService.findTaskInstanceById(taskResponseEvent.getTaskInstanceId());
switch (event) {
case ACK:
try {
if (taskInstance != null) {
ExecutionStatus status = taskInstance.getState().typeIsFinished() ? taskInstance.getState() : taskResponseEvent.getState();
processService.changeTaskState(taskInstance, status,
taskResponseEvent.getStartTime(),
taskResponseEvent.getWorkerAddress(),
taskResponseEvent.getExecutePath(),
taskResponseEvent.getLogPath(),
taskResponseEvent.getTaskInstanceId());
}
// if taskInstance is null (maybe deleted) . retry will be meaningless . so ack success
DBTaskAckCommand taskAckCommand = new DBTaskAckCommand(ExecutionStatus.SUCCESS.getCode(), taskResponseEvent.getTaskInstanceId());
channel.writeAndFlush(taskAckCommand.convert2Command());
} catch (Exception e) {
logger.error("worker ack master error", e);
DBTaskAckCommand taskAckCommand = new DBTaskAckCommand(ExecutionStatus.FAILURE.getCode(), -1);
channel.writeAndFlush(taskAckCommand.convert2Command());
}
handleAckEvent(taskResponseEvent, taskInstance);
break;
case RESULT:
try {
if (taskInstance != null) {
processService.changeTaskState(taskInstance, taskResponseEvent.getState(),
taskResponseEvent.getEndTime(),
taskResponseEvent.getProcessId(),
taskResponseEvent.getAppIds(),
taskResponseEvent.getTaskInstanceId(),
taskResponseEvent.getVarPool()
);
}
// if taskInstance is null (maybe deleted) . retry will be meaningless . so response success
DBTaskResponseCommand taskResponseCommand = new DBTaskResponseCommand(ExecutionStatus.SUCCESS.getCode(), taskResponseEvent.getTaskInstanceId());
channel.writeAndFlush(taskResponseCommand.convert2Command());
} catch (Exception e) {
logger.error("worker response master error", e);
DBTaskResponseCommand taskResponseCommand = new DBTaskResponseCommand(ExecutionStatus.FAILURE.getCode(), -1);
channel.writeAndFlush(taskResponseCommand.convert2Command());
}
handleResultEvent(taskResponseEvent, taskInstance);
break;
default:
throw new IllegalArgumentException("invalid event type : " + event);
}
WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecCacheManager.getByProcessInstanceId(taskResponseEvent.getProcessInstanceId());
if (workflowExecuteThread != null) {
StateEvent stateEvent = new StateEvent();
stateEvent.setProcessInstanceId(taskResponseEvent.getProcessInstanceId());
......@@ -200,7 +174,59 @@ public class TaskResponseService {
}
}
public BlockingQueue<TaskResponseEvent> getEventQueue() {
return eventQueue;
/**
* handle ack event
* @param taskResponseEvent
* @param taskInstance
*/
private void handleAckEvent(TaskResponseEvent taskResponseEvent, TaskInstance taskInstance) {
Channel channel = taskResponseEvent.getChannel();
try {
if (taskInstance != null) {
if (taskInstance.getState().typeIsFinished()) {
logger.warn("task is finish, ack is meaningless, taskInstanceId:{}, state:{}", taskInstance.getId(), taskInstance.getState());
} else {
processService.changeTaskState(taskInstance, taskResponseEvent.getState(),
taskResponseEvent.getStartTime(),
taskResponseEvent.getWorkerAddress(),
taskResponseEvent.getExecutePath(),
taskResponseEvent.getLogPath()
);
}
}
// if taskInstance is null (maybe deleted) or finish. retry will be meaningless . so ack success
DBTaskAckCommand taskAckCommand = new DBTaskAckCommand(ExecutionStatus.SUCCESS.getCode(), taskResponseEvent.getTaskInstanceId());
channel.writeAndFlush(taskAckCommand.convert2Command());
} catch (Exception e) {
logger.error("worker ack master error", e);
DBTaskAckCommand taskAckCommand = new DBTaskAckCommand(ExecutionStatus.FAILURE.getCode(), -1);
channel.writeAndFlush(taskAckCommand.convert2Command());
}
}
/**
* handle result event
* @param taskResponseEvent
* @param taskInstance
*/
private void handleResultEvent(TaskResponseEvent taskResponseEvent, TaskInstance taskInstance) {
Channel channel = taskResponseEvent.getChannel();
try {
if (taskInstance != null) {
processService.changeTaskState(taskInstance, taskResponseEvent.getState(),
taskResponseEvent.getEndTime(),
taskResponseEvent.getProcessId(),
taskResponseEvent.getAppIds(),
taskResponseEvent.getVarPool()
);
}
// if taskInstance is null (maybe deleted) . retry will be meaningless . so response success
DBTaskResponseCommand taskResponseCommand = new DBTaskResponseCommand(ExecutionStatus.SUCCESS.getCode(), taskResponseEvent.getTaskInstanceId());
channel.writeAndFlush(taskResponseCommand.convert2Command());
} catch (Exception e) {
logger.error("worker response master error", e);
DBTaskResponseCommand taskResponseCommand = new DBTaskResponseCommand(ExecutionStatus.FAILURE.getCode(), -1);
channel.writeAndFlush(taskResponseCommand.convert2Command());
}
}
}
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java
浏览文件 @ 28872d87
......@@ -161,8 +161,6 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
* @return TaskExecutionContext
*/
protected TaskExecutionContext getTaskExecutionContext(TaskInstance taskInstance) {
processService.setTaskInstanceDetail(taskInstance);
int userId = taskInstance.getProcessDefine() == null ? 0 : taskInstance.getProcessDefine().getUserId();
Tenant tenant = processService.getTenantForProcess(taskInstance.getProcessInstance().getTenantId(), userId);
......@@ -172,12 +170,12 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
taskInstance.getStartTime(),
taskInstance.getHost(),
null,
null,
taskInstance.getId());
null
);
return null;
}
// set queue for process instance, user-specified queue takes precedence over tenant queue
String userQueue = processService.queryUserQueueByProcessInstanceId(taskInstance.getProcessInstanceId());
String userQueue = processService.queryUserQueueByProcessInstance(taskInstance.getProcessInstance());
taskInstance.getProcessInstance().setQueue(StringUtils.isEmpty(userQueue) ? tenant.getQueue() : userQueue);
taskInstance.getProcessInstance().setTenantCode(tenant.getTenantCode());
taskInstance.setResources(getResourceFullNames(taskInstance));
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessor.java
浏览文件 @ 28872d87
......@@ -65,7 +65,7 @@ public class CommonTaskProcessor extends BaseTaskProcessor {
@Override
public boolean submit(TaskInstance task, ProcessInstance processInstance, int maxRetryTimes, int commitInterval) {
this.processInstance = processInstance;
this.taskInstance = processService.submitTask(task, maxRetryTimes, commitInterval);
this.taskInstance = processService.submitTaskWithRetry(processInstance, task, maxRetryTimes, commitInterval);
if (this.taskInstance == null) {
return false;
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/ConditionTaskProcessor.java
浏览文件 @ 28872d87
......@@ -72,7 +72,7 @@ public class ConditionTaskProcessor extends BaseTaskProcessor {
@Override
public boolean submit(TaskInstance task, ProcessInstance processInstance, int masterTaskCommitRetryTimes, int masterTaskCommitInterval) {
this.processInstance = processInstance;
this.taskInstance = processService.submitTask(task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
this.taskInstance = processService.submitTaskWithRetry(processInstance, task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
if (this.taskInstance == null) {
return false;
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/DependentTaskProcessor.java
浏览文件 @ 28872d87
......@@ -34,7 +34,6 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.utils.DependentExecute;
import org.apache.dolphinscheduler.server.utils.LogUtils;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.util.ArrayList;
import java.util.Date;
......@@ -81,7 +80,7 @@ public class DependentTaskProcessor extends BaseTaskProcessor {
public boolean submit(TaskInstance task, ProcessInstance processInstance, int masterTaskCommitRetryTimes, int masterTaskCommitInterval) {
this.processInstance = processInstance;
this.taskInstance = task;
this.taskInstance = processService.submitTask(task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
this.taskInstance = processService.submitTaskWithRetry(processInstance, task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
if (this.taskInstance == null) {
return false;
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SubTaskProcessor.java
浏览文件 @ 28872d87
......@@ -54,7 +54,7 @@ public class SubTaskProcessor extends BaseTaskProcessor {
taskDefinition = processService.findTaskDefinition(
task.getTaskCode(), task.getTaskDefinitionVersion()
);
this.taskInstance = processService.submitTask(task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
this.taskInstance = processService.submitTaskWithRetry(processInstance, task, masterTaskCommitRetryTimes, masterTaskCommitInterval);
if (this.taskInstance == null) {
return false;
......
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/SwitchTaskProcessor.java
浏览文件 @ 28872d87
......@@ -63,7 +63,7 @@ public class SwitchTaskProcessor extends BaseTaskProcessor {
public boolean submit(TaskInstance taskInstance, ProcessInstance processInstance, int masterTaskCommitRetryTimes, int masterTaskCommitInterval) {
this.processInstance = processInstance;
this.taskInstance = processService.submitTask(taskInstance, masterTaskCommitRetryTimes, masterTaskCommitInterval);
this.taskInstance = processService.submitTaskWithRetry(processInstance, taskInstance, masterTaskCommitRetryTimes, masterTaskCommitInterval);
if (this.taskInstance == null) {
return false;
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/ConditionsTaskTest.java
浏览文件 @ 28872d87
......@@ -89,7 +89,7 @@ public class ConditionsTaskTest {
// for MasterBaseTaskExecThread.submit
Mockito.when(processService
.submitTask(taskInstance))
.submitTask(processInstance, taskInstance))
.thenReturn(taskInstance);
// for MasterBaseTaskExecThread.call
Mockito.when(processService
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/DependentTaskTest.java
浏览文件 @ 28872d87
......@@ -93,6 +93,7 @@ public class DependentTaskTest {
Mockito.when(applicationContext.getBean(ProcessService.class)).thenReturn(processService);
processInstance = getProcessInstance();
taskInstance = getTaskInstance();
// for MasterBaseTaskExecThread.call
// for DependentTaskExecThread.waitTaskQuit
......@@ -102,7 +103,7 @@ public class DependentTaskTest {
// for MasterBaseTaskExecThread.submit
Mockito.when(processService
.submitTask(Mockito.argThat(taskInstance -> taskInstance.getId() == 1000)))
.submitTask(processInstance, taskInstance))
.thenAnswer(i -> taskInstance);
// for DependentTaskExecThread.initTaskParameters
......@@ -346,6 +347,12 @@ public class DependentTaskTest {
return processInstance;
}
private TaskInstance getTaskInstance() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(1000);
return taskInstance;
}
/**
* task that dependent on others (and to be tested here)
* notice: should be filled with setDependence() and be passed to setupTaskInstance()
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/SubProcessTaskTest.java
浏览文件 @ 28872d87
......@@ -74,6 +74,8 @@ public class SubProcessTaskTest {
Mockito.when(applicationContext.getBean(AlertDao.class)).thenReturn(alertDao);
processInstance = getProcessInstance();
TaskInstance taskInstance = getTaskInstance();
Mockito.when(processService
.findProcessInstanceById(processInstance.getId()))
.thenReturn(processInstance);
......@@ -85,7 +87,7 @@ public class SubProcessTaskTest {
// for MasterBaseTaskExecThread.submit
Mockito.when(processService
.submitTask(Mockito.any()))
.submitTask(processInstance, taskInstance))
.thenAnswer(t -> t.getArgument(0));
TaskDefinition taskDefinition = new TaskDefinition();
......@@ -147,6 +149,12 @@ public class SubProcessTaskTest {
return processInstance;
}
private TaskInstance getTaskInstance() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(1000);
return taskInstance;
}
private ProcessInstance getSubProcessInstance(ExecutionStatus executionStatus) {
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(102);
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/SwitchTaskTest.java
浏览文件 @ 28872d87
......@@ -81,7 +81,7 @@ public class SwitchTaskTest {
// for MasterBaseTaskExecThread.submit
Mockito.when(processService
.submitTask(taskInstance))
.submitTask(processInstance, taskInstance))
.thenReturn(taskInstance);
// for MasterBaseTaskExecThread.call
Mockito.when(processService
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/WorkflowExecuteThreadTest.java
浏览文件 @ 28872d87
......@@ -162,39 +162,52 @@ public class WorkflowExecuteThreadTest {
public void testGetPreVarPool() {
try {
Set<String> preTaskName = new HashSet<>();
preTaskName.add("test1");
preTaskName.add("test2");
Map<String, TaskInstance> completeTaskList = new ConcurrentHashMap<>();
preTaskName.add(Long.toString(1));
preTaskName.add(Long.toString(2));
TaskInstance taskInstance = new TaskInstance();
TaskInstance taskInstance1 = new TaskInstance();
taskInstance1.setId(1);
taskInstance1.setName("test1");
taskInstance1.setTaskCode(1);
taskInstance1.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test1\",\"type\":\"VARCHAR\",\"value\":\"1\"}]");
taskInstance1.setEndTime(new Date());
TaskInstance taskInstance2 = new TaskInstance();
taskInstance2.setId(2);
taskInstance2.setName("test2");
taskInstance2.setTaskCode(2);
taskInstance2.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test2\",\"type\":\"VARCHAR\",\"value\":\"2\"}]");
taskInstance2.setEndTime(new Date());
completeTaskList.put("test1", taskInstance1);
completeTaskList.put("test2", taskInstance2);
Map<Integer, TaskInstance> taskInstanceMap = new ConcurrentHashMap<>();
taskInstanceMap.put(taskInstance1.getId(), taskInstance1);
taskInstanceMap.put(taskInstance2.getId(), taskInstance2);
Map<String, Integer> completeTaskList = new ConcurrentHashMap<>();
completeTaskList.put(Long.toString(taskInstance1.getTaskCode()), taskInstance1.getId());
completeTaskList.put(Long.toString(taskInstance1.getTaskCode()), taskInstance2.getId());
Class<WorkflowExecuteThread> masterExecThreadClass = WorkflowExecuteThread.class;
Field field = masterExecThreadClass.getDeclaredField("completeTaskList");
field.setAccessible(true);
field.set(workflowExecuteThread, completeTaskList);
Field completeTaskMapField = masterExecThreadClass.getDeclaredField("completeTaskMap");
completeTaskMapField.setAccessible(true);
completeTaskMapField.set(workflowExecuteThread, completeTaskList);
Field taskInstanceMapField = masterExecThreadClass.getDeclaredField("taskInstanceMap");
taskInstanceMapField.setAccessible(true);
taskInstanceMapField.set(workflowExecuteThread, taskInstanceMap);
workflowExecuteThread.getPreVarPool(taskInstance, preTaskName);
Assert.assertNotNull(taskInstance.getVarPool());
taskInstance2.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test1\",\"type\":\"VARCHAR\",\"value\":\"2\"}]");
completeTaskList.put("test2", taskInstance2);
field.setAccessible(true);
field.set(workflowExecuteThread, completeTaskList);
completeTaskList.put(Long.toString(taskInstance2.getTaskCode()), taskInstance2.getId());
completeTaskMapField.setAccessible(true);
completeTaskMapField.set(workflowExecuteThread, completeTaskList);
taskInstanceMapField.setAccessible(true);
taskInstanceMapField.set(workflowExecuteThread, taskInstanceMap);
workflowExecuteThread.getPreVarPool(taskInstance, preTaskName);
Assert.assertNotNull(taskInstance.getVarPool());
} catch (Exception e) {
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/cache/impl/TaskInstanceCacheManagerImplTest.java 已删除 100644 → 0
浏览文件 @ f2d242c7
/*
* 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.cache.impl;
import static org.apache.dolphinscheduler.common.Constants.CACHE_REFRESH_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class TaskInstanceCacheManagerImplTest {
@InjectMocks
private TaskInstanceCacheManagerImpl taskInstanceCacheManager;
@Mock(name = "processService")
private ProcessService processService;
@Before
public void before() {
TaskExecuteAckCommand taskExecuteAckCommand = new TaskExecuteAckCommand();
taskExecuteAckCommand.setStatus(1);
taskExecuteAckCommand.setExecutePath("/dolphinscheduler/worker");
taskExecuteAckCommand.setHost("worker007");
taskExecuteAckCommand.setLogPath("/temp/worker.log");
taskExecuteAckCommand.setStartTime(new Date(1970, Calendar.AUGUST,7));
taskExecuteAckCommand.setTaskInstanceId(0);
taskInstanceCacheManager.cacheTaskInstance(taskExecuteAckCommand);
}
@Test
public void testInit() throws InterruptedException {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(0);
taskInstance.setState(ExecutionStatus.NEED_FAULT_TOLERANCE);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setProcessInstanceId(0);
Mockito.when(processService.findTaskInstanceById(0)).thenReturn(taskInstance);
taskInstanceCacheManager.init();
TimeUnit.MILLISECONDS.sleep(CACHE_REFRESH_TIME_MILLIS + 1000);
Assert.assertEquals(taskInstance.getState(), taskInstanceCacheManager.getByTaskInstanceId(0).getState());
}
@Test
public void getByTaskInstanceIdFromCache() {
TaskInstance instanceGot = taskInstanceCacheManager.getByTaskInstanceId(0);
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(0);
taskInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setStartTime(new Date(1970, Calendar.AUGUST,7));
Assert.assertEquals(taskInstance.toString(), instanceGot.toString());
}
@Test
public void getByTaskInstanceIdFromDatabase() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(1);
taskInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setStartTime(new Date(1970, Calendar.AUGUST,7));
Mockito.when(processService.findTaskInstanceById(1)).thenReturn(taskInstance);
TaskInstance instanceGot = taskInstanceCacheManager.getByTaskInstanceId(1);
Assert.assertEquals(taskInstance, instanceGot);
}
@Test
public void cacheTaskInstanceByTaskExecutionContext() {
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
taskExecutionContext.setTaskInstanceId(2);
taskExecutionContext.setTaskName("blackberrier test");
taskExecutionContext.setStartTime(new Date(1970, Calendar.AUGUST,7));
taskExecutionContext.setTaskType(TaskType.SPARK.getDesc());
taskExecutionContext.setExecutePath("/tmp");
taskInstanceCacheManager.cacheTaskInstance(taskExecutionContext);
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(2);
Assert.assertEquals(taskInstance.getId(), 2);
Assert.assertEquals(taskInstance.getName(), "blackberrier test");
Assert.assertEquals(taskInstance.getStartTime(), new Date(1970, Calendar.AUGUST, 7));
Assert.assertEquals(taskInstance.getTaskType(), TaskType.SPARK.getDesc());
Assert.assertEquals(taskInstance.getExecutePath(), "/tmp");
}
@Test
public void testCacheTaskInstanceByTaskExecuteAckCommand() {
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(0);
Assert.assertEquals(ExecutionStatus.RUNNING_EXECUTION, taskInstance.getState());
Assert.assertEquals(new Date(1970, Calendar.AUGUST, 7), taskInstance.getStartTime());
Assert.assertEquals("worker007", taskInstance.getHost());
Assert.assertEquals("/dolphinscheduler/worker", taskInstance.getExecutePath());
Assert.assertEquals("/temp/worker.log", taskInstance.getLogPath());
}
@Test
public void testCacheTaskInstanceByTaskExecuteResponseCommand() {
TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand();
responseCommand.setTaskInstanceId(0);
responseCommand.setStatus(9);
responseCommand.setEndTime(new Date(1970, Calendar.AUGUST, 8));
taskInstanceCacheManager.cacheTaskInstance(responseCommand);
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(0);
Assert.assertEquals(new Date(1970, Calendar.AUGUST, 8), taskInstance.getEndTime());
Assert.assertEquals(ExecutionStatus.KILL, taskInstance.getState());
}
@Test
public void removeByTaskInstanceId() {
taskInstanceCacheManager.removeByTaskInstanceId(0);
Assert.assertNull(taskInstanceCacheManager.getByTaskInstanceId(0));
}
}
\ No newline at end of file
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
浏览文件 @ 28872d87
......@@ -76,8 +76,6 @@ public class TaskPriorityQueueConsumerTest {
tenant.setUpdateTime(new Date());
Mockito.doReturn(tenant).when(processService).getTenantForProcess(1, 2);
Mockito.doReturn("default").when(processService).queryUserQueueByProcessInstanceId(1);
}
@Test
......@@ -101,7 +99,6 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
taskPriorityQueue.put(taskPriority);
......@@ -129,7 +126,6 @@ public class TaskPriorityQueueConsumerTest {
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
taskPriorityQueue.put(taskPriority);
......@@ -171,7 +167,6 @@ public class TaskPriorityQueueConsumerTest {
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
taskPriorityQueue.put(taskPriority);
......@@ -211,7 +206,6 @@ public class TaskPriorityQueueConsumerTest {
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
taskPriorityQueue.put(taskPriority);
......@@ -271,7 +265,6 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "NoWorkGroup");
......@@ -310,7 +303,6 @@ public class TaskPriorityQueueConsumerTest {
taskDefinition.setTimeoutFlag(TimeoutFlag.OPEN);
taskInstance.setTaskDefine(taskDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
TaskPriority taskPriority = new TaskPriority();
......@@ -342,7 +334,6 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setUserId(2);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "NoWorkGroup");
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessorTest.java
浏览文件 @ 28872d87
......@@ -18,20 +18,16 @@
package org.apache.dolphinscheduler.server.master.processor;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.net.InetSocketAddress;
import java.util.Date;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
......@@ -39,7 +35,7 @@ import org.powermock.modules.junit4.PowerMockRunner;
import io.netty.channel.Channel;
/**
* task ack processor test
* task ack processor test
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest({SpringApplicationContext.class, TaskResponseEvent.class})
......@@ -47,7 +43,6 @@ public class TaskAckProcessorTest {
private TaskAckProcessor taskAckProcessor;
private TaskResponseService taskResponseService;
private TaskInstanceCacheManagerImpl taskInstanceCacheManager;
private ProcessService processService;
private TaskExecuteAckCommand taskExecuteAckCommand;
private TaskResponseEvent taskResponseEvent;
......@@ -60,9 +55,6 @@ public class TaskAckProcessorTest {
taskResponseService = PowerMockito.mock(TaskResponseService.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskResponseService.class)).thenReturn(taskResponseService);
taskInstanceCacheManager = PowerMockito.mock(TaskInstanceCacheManagerImpl.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class)).thenReturn(taskInstanceCacheManager);
processService = PowerMockito.mock(ProcessService.class);
PowerMockito.when(SpringApplicationContext.getBean(ProcessService.class)).thenReturn(processService);
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/runner/task/CommonTaskProcessorTest.java
浏览文件 @ 28872d87
......@@ -86,7 +86,6 @@ public class CommonTaskProcessorTest {
taskDefinition.setTimeoutFlag(TimeoutFlag.OPEN);
taskInstance.setTaskDefine(taskDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
TaskExecutionContext taskExecutionContext = commonTaskProcessor.getTaskExecutionContext(taskInstance);
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java
浏览文件 @ 28872d87
......@@ -36,7 +36,6 @@ import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.dispatch.host.HostManager;
import org.apache.dolphinscheduler.server.master.dispatch.host.RandomHostManager;
......@@ -67,11 +66,6 @@ public class DependencyConfig {
return Mockito.mock(AlertMapper.class);
}
@Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@Bean
public ProcessService processService() {
return Mockito.mock(ProcessService.class);
......
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java
浏览文件 @ 28872d87
......@@ -36,7 +36,6 @@ import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.mockito.Mockito;
......@@ -59,11 +58,6 @@ public class TaskCallbackServiceTestConfig {
return Mockito.mock(AlertMapper.class);
}
@Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@Bean
public ProcessService processService() {
return Mockito.mock(ProcessService.class);
......
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java
浏览文件 @ 28872d87
......@@ -205,43 +205,54 @@ public class ProcessAlertManager {
List<TaskInstance> taskInstances,
ProjectUser projectUser) {
if (Flag.YES == processInstance.getIsSubProcess()) {
if (!isNeedToSendWarning(processInstance)) {
return;
}
boolean sendWarnning = false;
Alert alert = new Alert();
String cmdName = getCommandCnName(processInstance.getCommandType());
String success = processInstance.getState().typeIsSuccess() ? "success" : "failed";
alert.setTitle(cmdName + " " + success);
String content = getContentProcessInstance(processInstance, taskInstances,projectUser);
alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date());
alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert);
}
/**
* check if need to be send warning
*
* @param processInstance
* @return
*/
public boolean isNeedToSendWarning(ProcessInstance processInstance) {
if (Flag.YES == processInstance.getIsSubProcess()) {
return false;
}
boolean sendWarning = false;
WarningType warningType = processInstance.getWarningType();
switch (warningType) {
case ALL:
if (processInstance.getState().typeIsFinished()) {
sendWarnning = true;
sendWarning = true;
}
break;
case SUCCESS:
if (processInstance.getState().typeIsSuccess()) {
sendWarnning = true;
sendWarning = true;
}
break;
case FAILURE:
if (processInstance.getState().typeIsFailure()) {
sendWarnning = true;
sendWarning = true;
}
break;
default:
}
if (!sendWarnning) {
return;
}
Alert alert = new Alert();
String cmdName = getCommandCnName(processInstance.getCommandType());
String success = processInstance.getState().typeIsSuccess() ? "success" : "failed";
alert.setTitle(cmdName + " " + success);
String content = getContentProcessInstance(processInstance, taskInstances,projectUser);
alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date());
alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert);
return sendWarning;
}
/**
......
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
浏览文件 @ 28872d87
......@@ -100,6 +100,7 @@ import org.apache.dolphinscheduler.dao.utils.DagHelper;
import org.apache.dolphinscheduler.remote.command.StateEventChangeCommand;
import org.apache.dolphinscheduler.remote.processor.StateEventCallbackService;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.apache.dolphinscheduler.service.log.LogClientService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
......@@ -1087,24 +1088,17 @@ public class ProcessService {
/**
* retry submit task to db
*/
public TaskInstance submitTask(TaskInstance taskInstance, int commitRetryTimes, int commitInterval) {
public TaskInstance submitTaskWithRetry(ProcessInstance processInstance, TaskInstance taskInstance, int commitRetryTimes, int commitInterval) {
int retryTimes = 1;
boolean submitDB = false;
TaskInstance task = null;
while (retryTimes <= commitRetryTimes) {
try {
if (!submitDB) {
// submit task to db
task = submitTask(taskInstance);
if (task != null && task.getId() != 0) {
submitDB = true;
break;
}
}
if (!submitDB) {
logger.error("task commit to db failed , taskId {} has already retry {} times, please check the database", taskInstance.getId(), retryTimes);
// submit task to db
task = SpringApplicationContext.getBean(ProcessService.class).submitTask(processInstance, taskInstance);
if (task != null && task.getId() != 0) {
break;
}
logger.error("task commit to db failed , taskId {} has already retry {} times, please check the database", taskInstance.getId(), retryTimes);
Thread.sleep(commitInterval);
} catch (Exception e) {
logger.error("task commit to mysql failed", e);
......@@ -1118,12 +1112,12 @@ public class ProcessService {
* submit task to db
* submit sub process to command
*
* @param processInstance processInstance
* @param taskInstance taskInstance
* @return task instance
*/
@Transactional(rollbackFor = Exception.class)
public TaskInstance submitTask(TaskInstance taskInstance) {
ProcessInstance processInstance = this.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
public TaskInstance submitTask(ProcessInstance processInstance, TaskInstance taskInstance) {
logger.info("start submit task : {}, instance id:{}, state: {}",
taskInstance.getName(), taskInstance.getProcessInstanceId(), processInstance.getState());
//submit to db
......@@ -1131,8 +1125,9 @@ public class ProcessService {
if (task == null) {
logger.error("end submit task to db error, task name:{}, process id:{} state: {} ",
taskInstance.getName(), taskInstance.getProcessInstance(), processInstance.getState());
return task;
return null;
}
if (!task.getState().typeIsFinished()) {
createSubWorkProcess(processInstance, task);
}
......@@ -1383,7 +1378,7 @@ public class ProcessService {
}
taskInstance.setExecutorId(processInstance.getExecutorId());
taskInstance.setProcessInstancePriority(processInstance.getProcessInstancePriority());
taskInstance.setState(getSubmitTaskState(taskInstance, processInstanceState));
taskInstance.setState(getSubmitTaskState(taskInstance, processInstance));
if (taskInstance.getSubmitTime() == null) {
taskInstance.setSubmitTime(new Date());
}
......@@ -1406,10 +1401,10 @@ public class ProcessService {
* if all of above are not satisfied, return submit success
*
* @param taskInstance taskInstance
* @param processInstanceState processInstanceState
* @param processInstance processInstance
* @return process instance state
*/
public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ExecutionStatus processInstanceState) {
public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ProcessInstance processInstance) {
ExecutionStatus state = taskInstance.getState();
// running, delayed or killed
// the task already exists in task queue
......@@ -1423,10 +1418,10 @@ public class ProcessService {
}
//return pasue /stop if process instance state is ready pause / stop
// or return submit success
if (processInstanceState == ExecutionStatus.READY_PAUSE) {
if (processInstance.getState() == ExecutionStatus.READY_PAUSE) {
state = ExecutionStatus.PAUSE;
} else if (processInstanceState == ExecutionStatus.READY_STOP
|| !checkProcessStrategy(taskInstance)) {
} else if (processInstance.getState() == ExecutionStatus.READY_STOP
|| !checkProcessStrategy(taskInstance, processInstance)) {
state = ExecutionStatus.KILL;
} else {
state = ExecutionStatus.SUBMITTED_SUCCESS;
......@@ -1440,8 +1435,7 @@ public class ProcessService {
* @param taskInstance taskInstance
* @return check strategy result
*/
private boolean checkProcessStrategy(TaskInstance taskInstance) {
ProcessInstance processInstance = this.findProcessInstanceById(taskInstance.getProcessInstanceId());
private boolean checkProcessStrategy(TaskInstance taskInstance, ProcessInstance processInstance) {
FailureStrategy failureStrategy = processInstance.getFailureStrategy();
if (failureStrategy == FailureStrategy.CONTINUE) {
return true;
......@@ -1535,39 +1529,15 @@ public class ProcessService {
}
/**
* package task instance，associate processInstance and processDefine
*
* @param taskInstId taskInstId
* @return task instance
* package task instance
*/
public TaskInstance getTaskInstanceDetailByTaskId(int taskInstId) {
// get task instance
TaskInstance taskInstance = findTaskInstanceById(taskInstId);
if (taskInstance == null) {
return null;
}
setTaskInstanceDetail(taskInstance);
return taskInstance;
}
/**
* package task instance，associate processInstance and processDefine
*
* @param taskInstance taskInstance
* @return task instance
*/
public void setTaskInstanceDetail(TaskInstance taskInstance) {
// get process instance
ProcessInstance processInstance = findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
// get process define
ProcessDefinition processDefine = findProcessDefinition(processInstance.getProcessDefinitionCode(),
processInstance.getProcessDefinitionVersion());
public void packageTaskInstance(TaskInstance taskInstance, ProcessInstance processInstance) {
taskInstance.setProcessInstance(processInstance);
taskInstance.setProcessDefine(processDefine);
TaskDefinition taskDefinition = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(
taskInstance.setProcessDefine(processInstance.getProcessDefinition());
TaskDefinition taskDefinition = this.findTaskDefinition(
taskInstance.getTaskCode(),
taskInstance.getTaskDefinitionVersion());
updateTaskDefinitionResources(taskDefinition);
this.updateTaskDefinitionResources(taskDefinition);
taskInstance.setTaskDefine(taskDefinition);
}
......@@ -1576,7 +1546,7 @@ public class ProcessService {
*
* @param taskDefinition the given {@link TaskDefinition}
*/
private void updateTaskDefinitionResources(TaskDefinition taskDefinition) {
public void updateTaskDefinitionResources(TaskDefinition taskDefinition) {
Map<String, Object> taskParameters = JSONUtils.parseObject(
taskDefinition.getTaskParams(),
new TypeReference<Map<String, Object>>() {
......@@ -1757,12 +1727,10 @@ public class ProcessService {
* @param host host
* @param executePath executePath
* @param logPath logPath
* @param taskInstId taskInstId
*/
public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state, Date startTime, String host,
String executePath,
String logPath,
int taskInstId) {
String logPath) {
taskInstance.setState(state);
taskInstance.setStartTime(startTime);
taskInstance.setHost(host);
......@@ -1786,14 +1754,12 @@ public class ProcessService {
*
* @param state state
* @param endTime endTime
* @param taskInstId taskInstId
* @param varPool varPool
*/
public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state,
Date endTime,
int processId,
String appIds,
int taskInstId,
String varPool) {
taskInstance.setPid(processId);
taskInstance.setAppLink(appIds);
......@@ -2047,15 +2013,14 @@ public class ProcessService {
}
/**
* query user queue by process instance id
* query user queue by process instance
*
* @param processInstanceId processInstanceId
* @param processInstance processInstance
* @return queue
*/
public String queryUserQueueByProcessInstanceId(int processInstanceId) {
public String queryUserQueueByProcessInstance(ProcessInstance processInstance) {
String queue = "";
ProcessInstance processInstance = processInstanceMapper.selectById(processInstanceId);
if (processInstance == null) {
return queue;
}
......
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriority.java
浏览文件 @ 28872d87
......@@ -23,7 +23,7 @@ import java.util.Map;
import java.util.Objects;
/**
* task priority info
* task priority info
*/
public class TaskPriority implements Comparable<TaskPriority> {
......@@ -62,7 +62,14 @@ public class TaskPriority implements Comparable<TaskPriority> {
*/
private Map<String, String> context;
public TaskPriority(){}
/**
* checkpoint
*/
private long checkpoint;
public TaskPriority() {
this.checkpoint = System.currentTimeMillis();
}
public TaskPriority(int processInstancePriority,
int processInstanceId,
......@@ -73,6 +80,7 @@ public class TaskPriority implements Comparable<TaskPriority> {
this.taskInstancePriority = taskInstancePriority;
this.taskId = taskId;
this.groupName = groupName;
this.checkpoint = System.currentTimeMillis();
}
public int getProcessInstancePriority() {
......@@ -131,6 +139,14 @@ public class TaskPriority implements Comparable<TaskPriority> {
this.taskExecutionContext = taskExecutionContext;
}
public long getCheckpoint() {
return checkpoint;
}
public void setCheckpoint(long checkpoint) {
this.checkpoint = checkpoint;
}
@Override
public int compareTo(TaskPriority other) {
if (this.getProcessInstancePriority() > other.getProcessInstancePriority()) {
......@@ -174,7 +190,7 @@ public class TaskPriority implements Comparable<TaskPriority> {
}
TaskPriority that = (TaskPriority) o;
return processInstancePriority == that.processInstancePriority
&& processInstanceId == that.processInstanceId
&& processInstanceId == that.processInstanceId
&& taskInstancePriority == that.taskInstancePriority
&& taskId == that.taskId
&& Objects.equals(groupName, that.groupName);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或