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

cherry-pick after a submit failure, stop the processInstance to avoid an endless loop #13051

This commit is contained in:
fuchanghai 2023-02-20 14:10:59 +08:00 committed by zhuangchong
parent 92c744a2d3
commit f2789054fa
3 changed files with 23 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/StateEventType.java
View File

@ -27,7 +27,8 @@ public enum StateEventType {
TASK_TIMEOUT(3, "task timeout"),
WAKE_UP_TASK_GROUP(4, "wait task group"),
TASK_RETRY(5, "task retry"),
PROCESS_BLOCKED(6, "process blocked");
PROCESS_BLOCKED(6, "process blocked"),
PROCESS_SUBMIT_FAILED(7, "process submit failed");
StateEventType(int code, String descp) {
this.code = code;

29
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStartEventHandler.java
View File

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.server.master.event;
import org.apache.dolphinscheduler.common.enums.StateEventType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
@ -61,18 +63,23 @@ public class WorkflowStartEventHandler implements WorkflowEventHandler {
ProcessInstanceMetrics.incProcessInstanceByState("submit");
ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
CompletableFuture.supplyAsync(workflowExecuteRunnable::call, workflowExecuteThreadPool)
.thenAccept(workflowSubmitStatue -> {
if (WorkflowSubmitStatue.SUCCESS == workflowSubmitStatue) {
// submit failed will resend the event to workflow event queue
logger.info("Success submit the workflow instance");
if (processInstance.getTimeout() > 0) {
stateWheelExecuteThread.addProcess4TimeoutCheck(processInstance);
.thenAccept(workflowSubmitStatue -> {
if (WorkflowSubmitStatue.SUCCESS == workflowSubmitStatue) {
logger.info("Success submit the workflow instance");
if (processInstance.getTimeout() > 0) {
stateWheelExecuteThread.addProcess4TimeoutCheck(processInstance);
}
} else if (WorkflowSubmitStatue.FAILED == workflowSubmitStatue) {
logger.error(
"Failed to submit the workflow instance, will resend the workflow start event: {}",
workflowEvent);
WorkflowStateEvent stateEvent = WorkflowStateEvent.builder()
.processInstanceId(processInstance.getId())
.type(StateEventType.PROCESS_SUBMIT_FAILED)
.status(WorkflowExecutionStatus.FAILURE)
.build();
workflowExecuteRunnable.addStateEvent(stateEvent);
}
} else {
logger.error("Failed to submit the workflow instance, will resend the workflow start event: {}",
workflowEvent);
workflowEventQueue.addEvent(workflowEvent);
}
});
}

3
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStateEventHandler.java
View File

@ -57,6 +57,9 @@ public class WorkflowStateEventHandler implements StateEventHandler {
return true;
}
if (workflowStateEvent.getStatus().isFinished()) {
if (workflowStateEvent.getType().equals(StateEventType.PROCESS_SUBMIT_FAILED)) {
workflowExecuteRunnable.updateProcessInstanceState(workflowStateEvent);
}
workflowExecuteRunnable.endProcess();
}
if (processInstance.getState().isReadyStop()) {