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

[Bug] [Fix-10672] Dependent task retry bug (#10707) 

* fix 10517

* fix dep warn bug

Co-authored-by: JinyLeeChina <jiny.li@foxmail.com>
This commit is contained in:
JinYong Li 2022-07-01 09:53:10 +08:00 committed by GitHub
parent 1f7a328669
commit 3c20fe7a8a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 18 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -348,13 +348,17 @@ public class WorkflowExecuteThread implements Runnable {
return true;
}
TaskTimeoutStrategy taskTimeoutStrategy = taskInstance.getTaskDefine().getTimeoutNotifyStrategy();
if (TaskTimeoutStrategy.FAILED == taskTimeoutStrategy && !taskInstance.getState().typeIsFinished()) {
if ((TaskTimeoutStrategy.FAILED == taskTimeoutStrategy || TaskTimeoutStrategy.WARNFAILED == taskTimeoutStrategy) && !taskInstance.getState().typeIsFinished()) {
ITaskProcessor taskProcessor = activeTaskProcessorMaps.get(stateEvent.getTaskInstanceId());
taskProcessor.action(TaskAction.TIMEOUT);
if (taskInstance.isDependTask()) {
TaskInstance task = processService.findTaskInstanceById(taskInstance.getId());
taskFinished(task);
}
if (TaskTimeoutStrategy.WARNFAILED == taskTimeoutStrategy) {
ProjectUser projectUser = processService.queryProjectWithUserByProcessInstanceId(processInstance.getId());
processAlertManager.sendTaskTimeoutAlert(processInstance, taskInstance, projectUser);
}
} else {
ProjectUser projectUser = processService.queryProjectWithUserByProcessInstanceId(processInstance.getId());
processAlertManager.sendTaskTimeoutAlert(processInstance, taskInstance, projectUser);

7
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/DependentExecute.java
View File

@ -196,7 +196,12 @@ public class DependentExecute {
TaskInstance taskInstance = processService.findLastTaskInstanceInterval(taskCode, dateInterval);
DependResult result;
if (taskInstance == null) {
logger.warn("Cannot find the task in the process instance when the ProcessInstance is finish, taskCode: {}", taskCode);
TaskDefinition taskDefinition = processService.findTaskDefinitionByCode(taskCode);
if (taskDefinition == null) {
logger.error("Cannot find the task definition, something error, taskCode: {}", taskCode);
} else {
logger.warn("Cannot find the task in the process instance when the ProcessInstance is finish, taskCode: {}, taskName: {}", taskCode, taskDefinition.getName());
}
result = DependResult.FAILED;
} else {
logger.info("The running task, taskId:{}, taskCode:{}, taskName:{}", taskInstance.getId(), taskInstance.getTaskCode(), taskInstance.getName());

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

@ -2489,6 +2489,13 @@ public class ProcessService {
return taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, taskDefinitionVersion);
}
/**
* find task definition by code
*/
public TaskDefinition findTaskDefinitionByCode(long taskCode) {
return taskDefinitionMapper.queryByCode(taskCode);
}
/**
* find process task relation list by process
*/