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

fix 13346 (#13411) 

Co-authored-by: JinyLeeChina <jiny.li@foxmail.com>
This commit is contained in:
JinYong Li 2023-01-18 10:31:51 +08:00 committed by GitHub
parent feeda4aa32
commit cbafb7fe06
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 33 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
View File

@ -339,7 +339,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
}
break;
case RECOVER_SUSPENDED_PROCESS:
if (executionStatus.typeIsPause() || executionStatus.typeIsCancel()) {
if (executionStatus.typeIsCancel()) {
checkResult = true;
}
break;

20
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ExecutionStatus.java
View File

@ -99,8 +99,7 @@ public enum ExecutionStatus {
* @return status
*/
public boolean typeIsFinished() {
return typeIsSuccess() || typeIsFailure() || typeIsCancel() || typeIsPause()
|| typeIsStop();
return typeIsSuccess() || typeIsFailure() || typeIsCancel();
}
/**
@ -113,12 +112,12 @@ public enum ExecutionStatus {
}
/**
* status is pause
* status is ready pause
*
* @return status
*/
public boolean typeIsPause() {
return this == PAUSE;
public boolean typeIsReadyPause() {
return this == READY_PAUSE;
}
/**
@ -145,7 +144,16 @@ public enum ExecutionStatus {
* @return status
*/
public boolean typeIsCancel() {
return this == KILL || this == STOP;
return this == KILL || this == STOP || this == PAUSE;
}
/**
* status is ready cancel
*
* @return status
*/
public boolean typeIsReadyCancel() {
return this == READY_PAUSE || this == READY_STOP;
}
public int getCode() {

3
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java
View File

@ -546,8 +546,7 @@ public class TaskInstance implements Serializable {
}
public boolean isTaskComplete() {
return this.getState().typeIsPause()
|| this.getState().typeIsSuccess()
return this.getState().typeIsSuccess()
|| this.getState().typeIsCancel()
|| (this.getState().typeIsFailure() && !taskCanRetry());
}

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

@ -501,7 +501,7 @@ public class WorkflowExecuteThread implements Runnable {
logger.info("process:{} state {} change to {}", processInstance.getId(), processInstance.getState(), stateEvent.getExecutionStatus());
processInstance = processService.findProcessInstanceById(this.processInstance.getId());
if (stateEvent.getExecutionStatus() == ExecutionStatus.STOP) {
if (stateEvent.getExecutionStatus().typeIsCancel()) {
this.updateProcessInstanceState(stateEvent);
return true;
}
@ -529,7 +529,7 @@ public class WorkflowExecuteThread implements Runnable {
return false;
}
if (processInstance.getState() == ExecutionStatus.READY_STOP) {
if (processInstance.getState().typeIsReadyCancel()) {
return false;
}
@ -984,7 +984,12 @@ public class WorkflowExecuteThread implements Runnable {
continue;
}
TaskInstance task = createTaskInstance(processInstance, taskNodeObject);
taskInstances.add(task);
if (processInstance.getState().typeIsReadyPause() && (!activeTaskProcessorMaps.isEmpty() || !completeTaskList.isEmpty())) {
task.setState(ExecutionStatus.PAUSE);
completeTaskList.put(String.valueOf(task.getTaskCode()), task);
} else {
taskInstances.add(task);
}
}
// if previous node success , post node submit
@ -998,7 +1003,7 @@ public class WorkflowExecuteThread implements Runnable {
logger.info("task {} has already run success, task id:{}", task.getName(), task.getId());
continue;
}
if (task.getState().typeIsPause() || task.getState().typeIsCancel()) {
if (task.getState().typeIsCancel()) {
logger.info("task {} stopped, the state is {}, task id:{}", task.getName(), task.getState(), task.getId());
} else {
addTaskToStandByList(task);
@ -1030,7 +1035,7 @@ public class WorkflowExecuteThread implements Runnable {
return DependResult.WAITING;
}
ExecutionStatus depTaskState = completeTaskList.get(depsNode).getState();
if (depTaskState.typeIsPause() || depTaskState.typeIsCancel()) {
if (depTaskState.typeIsCancel()) {
return DependResult.NON_EXEC;
}
// ignore task state if current task is condition
@ -1198,9 +1203,7 @@ public class WorkflowExecuteThread implements Runnable {
}
List<TaskInstance> pauseList = getCompleteTaskByState(ExecutionStatus.PAUSE);
if (CollectionUtils.isNotEmpty(pauseList)
|| !isComplementEnd()
|| readyToSubmitTaskQueue.size() > 0) {
if (CollectionUtils.isNotEmpty(pauseList) || !isComplementEnd()) {
return ExecutionStatus.PAUSE;
} else {
return ExecutionStatus.SUCCESS;

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

@ -106,7 +106,7 @@ public class CommonTaskProcessor extends BaseTaskProcessor {
protected boolean persistTask(TaskAction taskAction) {
switch (taskAction) {
case STOP:
if (taskInstance.getState().typeIsFinished() && !taskInstance.getState().typeIsCancel()) {
if (taskInstance.getState().typeIsSuccess() || taskInstance.getState().typeIsFailure()) {
return true;
}
taskInstance.setState(ExecutionStatus.KILL);

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

@ -123,7 +123,7 @@ public class ConditionTaskProcessor extends BaseTaskProcessor {
protected boolean persistTask(TaskAction taskAction) {
switch (taskAction) {
case STOP:
if (taskInstance.getState().typeIsFinished() && !taskInstance.getState().typeIsCancel()) {
if (taskInstance.getState().typeIsSuccess() || taskInstance.getState().typeIsFailure()) {
return true;
}
this.taskInstance.setState(ExecutionStatus.KILL);

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

@ -154,7 +154,7 @@ public class DependentTaskProcessor extends BaseTaskProcessor {
protected boolean persistTask(TaskAction taskAction) {
switch (taskAction) {
case STOP:
if (taskInstance.getState().typeIsFinished() && !taskInstance.getState().typeIsCancel()) {
if (taskInstance.getState().typeIsSuccess() || taskInstance.getState().typeIsFailure()) {
return true;
}
this.taskInstance.setState(ExecutionStatus.KILL);

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

@ -99,7 +99,7 @@ public class SwitchTaskProcessor extends BaseTaskProcessor {
protected boolean persistTask(TaskAction taskAction) {
switch (taskAction) {
case STOP:
if (taskInstance.getState().typeIsFinished() && !taskInstance.getState().typeIsCancel()) {
if (taskInstance.getState().typeIsSuccess() || taskInstance.getState().typeIsFailure()) {
return true;
}
this.taskInstance.setState(ExecutionStatus.KILL);

5
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/task/ExecutionStatus.java
View File

@ -96,8 +96,7 @@ public enum ExecutionStatus {
* @return status
*/
public boolean typeIsFinished() {
return typeIsSuccess() || typeIsFailure() || typeIsCancel() || typeIsPause()
|| typeIsStop();
return typeIsSuccess() || typeIsFailure() || typeIsCancel();
}
/**
@ -142,7 +141,7 @@ public enum ExecutionStatus {
* @return status
*/
public boolean typeIsCancel() {
return this == KILL || this == STOP;
return this == KILL || this == STOP || this == PAUSE;
}
public int getCode() {