This is an automated email from the ASF dual-hosted git repository.
caishunfeng pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 3e5781b [Fix-9065] [master] when task submit failed, remove from
queue and add state event to thread (#9186)
3e5781b is described below
commit 3e5781b6e1c38675ba134048b21941be63723cb6
Author: guoshupei <[email protected]>
AuthorDate: Fri Apr 1 22:23:35 2022 +0800
[Fix-9065] [master] when task submit failed, remove from queue and add
state event to thread (#9186)
* [Fix-9065] [master] when task submit failed, remove from queue and add
state event to thread
This closes #9065
* Update
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
yes, your idea is better
Co-authored-by: caishunfeng <[email protected]>
* [Bug-9065] [master] update logger info
Co-authored-by: guoshupei <[email protected]>
Co-authored-by: caishunfeng <[email protected]>
---
.../server/master/runner/WorkflowExecuteThread.java | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
index 1d3413b..9c1aa90 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
@@ -1637,7 +1637,9 @@ public class WorkflowExecuteThread {
stateEvent.setExecutionStatus(processInstance.getState());
stateEvent.setProcessInstanceId(this.processInstance.getId());
stateEvent.setType(StateEventType.PROCESS_STATE_CHANGE);
- this.processStateChangeHandler(stateEvent);
+// this.processStateChangeHandler(stateEvent);
+ // replace with `stateEvents`, make sure `WorkflowExecuteThread`
can be deleted to avoid memory leaks
+ this.stateEvents.add(stateEvent);
}
}
@@ -1800,6 +1802,11 @@ public class WorkflowExecuteThread {
TaskInstance taskInstance = submitTaskExec(task);
if (taskInstance == null) {
this.taskFailedSubmit = true;
+ // Remove and add to complete map and error map
+ removeTaskFromStandbyList(task);
+ completeTaskMap.put(task.getTaskCode(), task.getId());
+ errorTaskMap.put(task.getTaskCode(), task.getId());
+ logger.error("process {}, task {}, code:{} submit task
failed.", task.getProcessInstanceId(), task.getName(), task.getTaskCode());
} else {
removeTaskFromStandbyList(task);
}
