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

[Fix-13244] [Master] Stopping a workflow does not update task status correctly (#13375) 

Co-authored-by: JinyLeeChina <jiny.li@foxmail.com>
This commit is contained in:
JinYong Li 2023-01-13 21:06:09 +08:00 committed by GitHub
parent a7ddc9047b
commit feeda4aa32
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 135 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/HadoopUtils.java
View File

@ -92,8 +92,9 @@ public class HadoopUtils implements Closeable {
private FileSystem fs;
private HadoopUtils() {
init();
initHdfsPath();
if(init()) {
initHdfsPath();
}
}
public static HadoopUtils getInstance() {
@ -120,7 +121,7 @@ public class HadoopUtils implements Closeable {
/**
* init hadoop configuration
*/
private void init() {
private boolean init() {
try {
configuration = new HdfsConfiguration();
@ -168,11 +169,13 @@ public class HadoopUtils implements Closeable {
configuration.set(Constants.FS_S3A_ACCESS_KEY, PropertyUtils.getString(Constants.FS_S3A_ACCESS_KEY));
configuration.set(Constants.FS_S3A_SECRET_KEY, PropertyUtils.getString(Constants.FS_S3A_SECRET_KEY));
fs = FileSystem.get(configuration);
} else {
return false;
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return true;
}
/**

2
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.java
View File

@ -84,4 +84,6 @@ public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
TaskInstance queryLastTaskInstance(@Param("taskCode") long taskCode, @Param("startTime") Date startTime, @Param("endTime") Date endTime);
List<TaskInstance> queryLastTaskInstanceList(@Param("taskCodes") Set<Long> taskCodes, @Param("startTime") Date startTime, @Param("endTime") Date endTime);
List<TaskInstance> queryTaskInstanceListByIds(@Param("ids") Set<Integer> ids);
}

12
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskInstanceMapper.xml
View File

@ -196,4 +196,16 @@
and start_time <![CDATA[ >= ]]> #{startTime} and start_time <![CDATA[ <= ]]> #{endTime}
</if>
</select>
<select id="queryTaskInstanceListByIds" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
from t_ds_task_instance
where 1=1
<if test="ids != null and ids.size() != 0">
and id in
<foreach collection="ids" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
</select>
</mapper>

6
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskKillResponseProcessor.java
View File

@ -71,10 +71,8 @@ public class TaskKillResponseProcessor implements NettyRequestProcessor {
TaskKillResponseCommand responseCommand = JSONUtils.parseObject(command.getBody(), TaskKillResponseCommand.class);
logger.info("received task kill response command : {}", responseCommand);
// TaskResponseEvent
TaskResponseEvent taskResponseEvent = TaskResponseEvent.newActionStop(ExecutionStatus.of(responseCommand.getStatus()),
responseCommand.getTaskInstanceId(),
responseCommand.getProcessInstanceId()
);
TaskResponseEvent taskResponseEvent = TaskResponseEvent.newKillResponse(ExecutionStatus.of(responseCommand.getStatus()),
responseCommand.getTaskInstanceId(), channel, responseCommand.getProcessInstanceId());
taskResponseService.addResponse(taskResponseEvent);
}

13
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseEvent.java
View File

@ -100,6 +100,19 @@ public class TaskResponseEvent {
*/
private long opaque;
public static TaskResponseEvent newKillResponse(ExecutionStatus state,
int taskInstanceId,
Channel channel,
int processInstanceId) {
TaskResponseEvent event = new TaskResponseEvent();
event.setState(state);
event.setTaskInstanceId(taskInstanceId);
event.setEvent(Event.ACTION_STOP);
event.setChannel(channel);
event.setProcessInstanceId(processInstanceId);
return event;
}
public static TaskResponseEvent newActionStop(ExecutionStatus state,
int taskInstanceId,
int processInstanceId) {

8
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponsePersistThread.java
View File

@ -26,7 +26,6 @@ import org.apache.dolphinscheduler.remote.command.DBTaskAckCommand;
import org.apache.dolphinscheduler.remote.command.DBTaskResponseCommand;
import org.apache.dolphinscheduler.remote.command.TaskKillAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskRecallAckCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRemoteChannel;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
import org.apache.dolphinscheduler.server.master.runner.task.ITaskProcessor;
import org.apache.dolphinscheduler.server.master.runner.task.TaskAction;
@ -159,6 +158,10 @@ public class TaskResponsePersistThread implements Runnable {
taskProcessor.persist(TaskAction.STOP);
logger.debug("ACTION_STOP: task instance id:{}, process instance id:{}", taskResponseEvent.getTaskInstanceId(), taskResponseEvent.getProcessInstanceId());
}
workflowExecuteThread.getActiveTaskProcessorMaps().remove(taskResponseEvent.getTaskInstanceId());
if (workflowExecuteThread.activeTaskFinish()) {
this.processInstanceMapper.remove(taskResponseEvent.getProcessInstanceId());
}
}
if (channel != null) {
@ -197,7 +200,8 @@ public class TaskResponsePersistThread implements Runnable {
}
WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(taskResponseEvent.getProcessInstanceId());
if (workflowExecuteThread != null && taskResponseEvent.getState().typeIsFinished()) {
if (workflowExecuteThread != null && taskResponseEvent.getState().typeIsFinished()
&& event != Event.ACTION_STOP && !workflowExecuteThread.getProcessInstance().getState().typeIsStop()) {
StateEvent stateEvent = new StateEvent();
stateEvent.setProcessInstanceId(taskResponseEvent.getProcessInstanceId());
stateEvent.setTaskInstanceId(taskResponseEvent.getTaskInstanceId());

6
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseService.java
View File

@ -177,7 +177,7 @@ public class TaskResponseService {
Thread.currentThread().interrupt();
break;
} catch (Exception e) {
logger.error("persist task error", e);
logger.error("handle task error", e);
}
}
logger.info("StateEventResponseWorker stopped");
@ -227,7 +227,7 @@ public class TaskResponseService {
FutureCallback futureCallback = new FutureCallback() {
@Override
public void onSuccess(Object o) {
logger.info("persist events {} succeeded.", taskResponsePersistThread.getProcessInstanceId());
logger.info("handle events {} succeeded.", taskResponsePersistThread.getProcessInstanceId());
if (!processInstanceMap.containsKey(taskResponsePersistThread.getProcessInstanceId())) {
processTaskResponseMap.remove(taskResponsePersistThread.getProcessInstanceId());
logger.info("remove process instance: {}", taskResponsePersistThread.getProcessInstanceId());
@ -237,7 +237,7 @@ public class TaskResponseService {
@Override
public void onFailure(Throwable throwable) {
logger.error("persist events failed: {}", throwable);
logger.error("handle events failed: {}", throwable.getMessage());
if (!processInstanceMap.containsKey(taskResponsePersistThread.getProcessInstanceId())) {
processTaskResponseMap.remove(taskResponsePersistThread.getProcessInstanceId());
logger.info("remove process instance: {}", taskResponsePersistThread.getProcessInstanceId());

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/EventExecuteService.java
View File

@ -133,7 +133,7 @@ public class EventExecuteService extends Thread {
FutureCallback futureCallback = new FutureCallback() {
@Override
public void onSuccess(Object o) {
if (workflowExecuteThread.workFlowFinish()) {
if (workflowExecuteThread.workFlowFinish() && workflowExecuteThread.activeTaskFinish()) {
processInstanceExecMaps.remove(processInstanceId);
notifyProcessChanged();
logger.info("process instance {} finished.", processInstanceId);

18
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
View File

@ -942,11 +942,6 @@ public class WorkflowExecuteThread implements Runnable {
taskInstance.setVarPool(JSONUtils.toJsonString(allProperty.values()));
}
}
// else {
// if (StringUtils.isNotEmpty(processInstance.getVarPool())) {
// taskInstance.setVarPool(processInstance.getVarPool());
// }
// }
}
private void setVarPoolValue(Map<String, Property> allProperty, Map<String, TaskInstance> allTaskInstance, TaskInstance preTaskInstance, Property thisProperty) {
@ -1457,6 +1452,19 @@ public class WorkflowExecuteThread implements Runnable {
return this.processInstance.getState().typeIsFinished();
}
public boolean activeTaskFinish() {
if (activeTaskProcessorMaps.isEmpty()) {
return true;
}
List<TaskInstance> taskInstanceList = processService.findTaskInstanceListByIds(activeTaskProcessorMaps.keySet());
for (TaskInstance taskInstance : taskInstanceList) {
if (!taskInstance.getState().typeIsFinished()) {
return false;
}
}
return true;
}
/**
* handling the list of tasks to be submitted
*/

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
View File

@ -88,8 +88,6 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
* @param taskExecutionContext task
*/
private void setTaskCache(TaskExecutionContext taskExecutionContext) {
TaskExecutionContext preTaskCache = new TaskExecutionContext();
preTaskCache.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
TaskRequest taskRequest = JSONUtils.parseObject(JSONUtils.toJsonString(taskExecutionContext), TaskRequest.class);
TaskExecutionContextCacheManager.cacheTaskExecutionContext(taskRequest);
}

11
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskKillAckProcessor.java
View File

@ -41,20 +41,19 @@ public class TaskKillAckProcessor implements NettyRequestProcessor {
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_KILL_RESPONSE_ACK == command.getType(),
String.format("invalid command type : %s", command.getType()));
TaskKillAckCommand taskKillAckCommand = JSONUtils.parseObject(
command.getBody(), TaskKillAckCommand.class);
TaskKillAckCommand taskKillAckCommand = JSONUtils.parseObject(command.getBody(), TaskKillAckCommand.class);
if (taskKillAckCommand == null) {
logger.warn("Cannot parse command, command type: {}", command.getType());
return;
}
logger.info("received kill ack command : {}", taskKillAckCommand);
if (taskKillAckCommand.getStatus() == ExecutionStatus.SUCCESS.getCode()) {
ResponceCache.get().removeKillResponseCache(taskKillAckCommand.getTaskInstanceId());
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskKillAckCommand.getTaskInstanceId());
logger.debug("removeKillResponseCache: task instance id:{}", taskKillAckCommand.getTaskInstanceId());
logger.info("removeKillResponseCache: task instance id:{}", taskKillAckCommand.getTaskInstanceId());
TaskCallbackService.remove(taskKillAckCommand.getTaskInstanceId());
logger.debug("remove REMOTE_CHANNELS, task instance id:{}", taskKillAckCommand.getTaskInstanceId());
logger.info("remove REMOTE_CHANNELS, task instance id:{}", taskKillAckCommand.getTaskInstanceId());
}
}
}

16
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskKillProcessor.java
View File

@ -95,16 +95,18 @@ public class TaskKillProcessor implements NettyRequestProcessor {
TaskKillRequestCommand killCommand = JSONUtils.parseObject(command.getBody(), TaskKillRequestCommand.class);
logger.info("received kill command : {}", killCommand);
taskCallbackService.addRemoteChannel(killCommand.getTaskInstanceId(),
new NettyRemoteChannel(channel, command.getOpaque()));
TaskRequest taskRequest = TaskExecutionContextCacheManager.getByTaskInstanceId(killCommand.getTaskInstanceId());
if (taskRequest == null) {
logger.warn("Cannot find taskInstanceId {} in taskContextCacheManager", killCommand.getTaskInstanceId());
return;
}
taskRequest.setCurrentExecutionStatus(org.apache.dolphinscheduler.spi.task.ExecutionStatus.STOP);
TaskExecutionContextCacheManager.updateTaskExecutionContext(taskRequest);
taskCallbackService.addRemoteChannel(killCommand.getTaskInstanceId(), new NettyRemoteChannel(channel, command.getOpaque()));
Pair<Boolean, List<String>> result = doKill(killCommand);
TaskRequest taskRequest = TaskExecutionContextCacheManager.getByTaskInstanceId(killCommand.getTaskInstanceId());
if (taskRequest == null) {
return;
}
TaskKillResponseCommand taskKillResponseCommand = buildKillTaskResponseCommand(taskRequest, result);
ResponceCache.get().cache(taskKillResponseCommand.getTaskInstanceId(), taskKillResponseCommand.convert2Command(), Event.ACTION_STOP);
taskCallbackService.sendResult(taskKillResponseCommand.getTaskInstanceId(), taskKillResponseCommand.convert2Command());

76
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
View File

@ -20,10 +20,14 @@ package org.apache.dolphinscheduler.server.worker.runner;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.Event;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.*;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.LoggerUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.server.utils.LogUtils;
@ -51,15 +55,12 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.rholder.retry.RetryException;
/**
* task scheduler thread
*/
@ -141,11 +142,11 @@ public class TaskExecuteThread implements Runnable, Delayed {
taskExecutionContext.setStartTime(new Date());
}
if (taskExecutionContext.getCurrentExecutionStatus() != ExecutionStatus.RUNNING_EXECUTION) {
changeTaskExecutionStatusToRunning();
//changeTaskExecutionStatusToRunning();
logger.info("the task begins to execute. task instance id: {}", taskExecutionContext.getTaskInstanceId());
taskExecutionContext.setCurrentExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
sendTaskExecuteRunningCommand(taskExecutionContext);
}
logger.info("the task begins to execute. task instance id: {}", taskExecutionContext.getTaskInstanceId());
taskExecutionContext.setCurrentExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
sendTaskExecuteRunningCommand(taskExecutionContext);
int dryRun = taskExecutionContext.getDryRun();
// copy hdfs/minio file to local
if (dryRun == Constants.DRY_RUN_FLAG_NO) {
@ -168,13 +169,19 @@ public class TaskExecuteThread implements Runnable, Delayed {
throw new RuntimeException(String.format("%s Task Plugin Not Found,Please Check Config File.", taskExecutionContext.getTaskType()));
}
TaskRequest taskRequest = JSONUtils.parseObject(JSONUtils.toJsonString(taskExecutionContext), TaskRequest.class);
if (null == taskRequest) {
throw new RuntimeException("The taskExecutionContext parse error");
}
String taskLogName = LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
taskExecutionContext.getProcessDefineCode(),
taskExecutionContext.getProcessDefineVersion(),
taskExecutionContext.getProcessInstanceId(),
taskExecutionContext.getTaskInstanceId());
taskRequest.setTaskLogName(taskLogName);
if (!TaskExecutionContextCacheManager.updateTaskExecutionContext(taskRequest)) {
TaskExecutionContextCacheManager.cacheTaskExecutionContext(taskRequest);
logger.info("taskRequest reCache successfully, taskInstanceId: {}", taskExecutionContext.getTaskInstanceId());
}
// set the name of the current thread
Thread.currentThread().setName(String.format(TaskConstants.TASK_LOGGER_THREAD_NAME_FORMAT,taskLogName));
@ -212,9 +219,14 @@ public class TaskExecuteThread implements Runnable, Delayed {
responseCommand.setProcessId(task.getProcessId());
responseCommand.setAppIds(task.getAppIds());
} finally {
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
if (TaskExecutionContextCacheManager.statusIsStop(taskExecutionContext.getTaskInstanceId())) {
logger.info("task has exited, taskInstanceId:{}, exitStatusCode:{}, task executionStatus:{}",
taskExecutionContext.getTaskInstanceId(), this.task.getExitStatusCode(), ExecutionStatus.STOP);
} else {
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
}
clearTaskExecPath();
}
}
@ -352,42 +364,6 @@ public class TaskExecuteThread implements Runnable, Delayed {
}
}
/**
* send an ack to change the status of the task.
*/
private void changeTaskExecutionStatusToRunning() {
taskExecutionContext.setCurrentExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
Command ackCommand = buildAckCommand().convert2Command();
try {
RetryerUtils.retryCall(() -> {
taskCallbackService.sendAck(taskExecutionContext.getTaskInstanceId(), ackCommand);
return Boolean.TRUE;
});
} catch (ExecutionException | RetryException e) {
logger.error(e.getMessage(), e);
}
}
/**
* build ack command.
*
* @return TaskExecuteAckCommand
*/
private TaskExecuteAckCommand buildAckCommand() {
TaskExecuteAckCommand ackCommand = new TaskExecuteAckCommand();
ackCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
ackCommand.setStatus(taskExecutionContext.getCurrentExecutionStatus().getCode());
ackCommand.setStartTime(taskExecutionContext.getStartTime());
ackCommand.setLogPath(taskExecutionContext.getLogPath());
ackCommand.setHost(taskExecutionContext.getHost());
if (TaskType.SQL.getDesc().equalsIgnoreCase(taskExecutionContext.getTaskType()) || TaskType.PROCEDURE.getDesc().equalsIgnoreCase(taskExecutionContext.getTaskType())) {
ackCommand.setExecutePath(null);
} else {
ackCommand.setExecutePath(taskExecutionContext.getExecutePath());
}
return ackCommand;
}
/**
* get current TaskExecutionContext
*

15
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
View File

@ -1489,12 +1489,25 @@ public class ProcessService {
* find task instance by id
*
* @param taskId task id
* @return task intance
* @return task instance
*/
public TaskInstance findTaskInstanceById(Integer taskId) {
return taskInstanceMapper.selectById(taskId);
}
/**
* find task instance list by ids
*
* @param taskIds task id list
* @return task instance list
*/
public List<TaskInstance> findTaskInstanceListByIds(Set<Integer> taskIds) {
if (CollectionUtils.isEmpty(taskIds)) {
return new ArrayList<>();
}
return taskInstanceMapper.queryTaskInstanceListByIds(taskIds);
}
/**
* package task instance，associate processInstance and processDefine
*

8
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/task/TaskExecutionContextCacheManager.java
View File

@ -71,4 +71,12 @@ public class TaskExecutionContextCacheManager {
public static Collection<TaskRequest> getAllTaskRequestList() {
return taskRequestContextCache.values();
}
public static boolean statusIsStop(Integer taskInstanceId) {
TaskRequest taskRequest = taskRequestContextCache.get(taskInstanceId);
if (taskRequest == null) {
return true;
}
return taskRequest.getCurrentExecutionStatus().typeIsStop();
}
}

14
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/task/request/TaskRequest.java
View File

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.spi.task.request;
import org.apache.dolphinscheduler.spi.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.spi.task.ExecutionStatus;
import org.apache.dolphinscheduler.spi.task.Property;
import java.util.Date;
@ -183,6 +184,11 @@ public class TaskRequest {
*/
private int delayTime;
/**
* current execution status
*/
private ExecutionStatus currentExecutionStatus;
/**
* Task Logger name should be like: Task-{processDefinitionId}-{processInstanceId}-{taskInstanceId}
*/
@ -473,6 +479,14 @@ public class TaskRequest {
this.delayTime = delayTime;
}
public ExecutionStatus getCurrentExecutionStatus() {
return currentExecutionStatus;
}
public void setCurrentExecutionStatus(ExecutionStatus currentExecutionStatus) {
this.currentExecutionStatus = currentExecutionStatus;
}
public SQLTaskExecutionContext getSqlTaskExecutionContext() {
return sqlTaskExecutionContext;
}