Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[improve-#13045] after a submit failure, stop the processInstance to avoid an endless loop #13051

Merged
merged 10 commits into from
Feb 20, 2023
Merged

[improve-#13045] after a submit failure, stop the processInstance to avoid an endless loop #13051

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0c139b5
[improve-#13045] add max submit number of workflow
Nov 30, 2022
e9a6cda
[improve-#13045] format
Nov 30, 2022
fb7e0a5
[improve-#13053] fix
Dec 15, 2022
29dd0f7
Merge branch 'apache:dev' into improve-#13045
fuchanghai Dec 26, 2022
622d7ca
Merge branch 'dev' into improve-#13045
fuchanghai Feb 9, 2023
ddbbb10
[improve-#13045] add max submit number of workflow
Nov 30, 2022
8c231fc
[fix-13201]fix error
fuchanghai Feb 15, 2023
c2a12fa
[fix-13405] update logic
fuchanghai Feb 17, 2023
0ce6b70
[fix-13405] remove max times
fuchanghai Feb 19, 2023
e657d1b
Update dolphinscheduler-master/src/main/resources/application.yaml
fuchanghai Feb 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...heduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/StateEventType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
15 changes: 11 additions & 4 deletions .../main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStartEventHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -66,11 +68,16 @@ public void handleWorkflowEvent(final WorkflowEvent workflowEvent) throws Workfl
if (processInstance.getTimeout() > 0) {
stateWheelExecuteThread.addProcess4TimeoutCheck(processInstance);
}
} else {
// submit failed will resend the event to workflow event queue
log.error("Failed to submit the workflow instance, will resend the workflow start event: {}",
} else if (WorkflowSubmitStatue.FAILED == workflowSubmitStatue) {
log.error(
"Failed to submit the workflow instance, will resend the workflow start event: {}",
workflowEvent);
workflowEventQueue.addEvent(workflowEvent);
fuchanghai marked this conversation as resolved.
Show resolved Hide resolved
WorkflowStateEvent stateEvent = WorkflowStateEvent.builder()
.processInstanceId(processInstance.getId())
.type(StateEventType.PROCESS_SUBMIT_FAILED)
.status(WorkflowExecutionStatus.FAILURE)
.build();
workflowExecuteRunnable.addStateEvent(stateEvent);
}
});
}
Expand Down
3 changes: 3 additions & 0 deletions .../main/java/org/apache/dolphinscheduler/server/master/event/WorkflowStateEventHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable,
return true;
}
if (workflowStateEvent.getStatus().isFinished()) {
if (workflowStateEvent.getType().equals(StateEventType.PROCESS_SUBMIT_FAILED)) {
workflowExecuteRunnable.updateProcessInstanceState(workflowStateEvent);
}
workflowExecuteRunnable.endProcess();
}
if (processInstance.getState().isReadyStop()) {
Expand Down