You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by le...@apache.org on 2021/12/06 14:54:56 UTC
[dolphinscheduler] branch 2.0.1-prepare updated: fix timeout (#7224)
This is an automated email from the ASF dual-hosted git repository.
leonbao pushed a commit to branch 2.0.1-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/2.0.1-prepare by this push:
new 3084847 fix timeout (#7224)
3084847 is described below
commit 30848478bfe5d18bea9730dc74161207a8b60c7e
Author: wind <ca...@users.noreply.github.com>
AuthorDate: Mon Dec 6 22:54:49 2021 +0800
fix timeout (#7224)
Co-authored-by: caishunfeng <53...@qq.com>
---
.../master/runner/StateWheelExecuteThread.java | 6 +++---
.../master/runner/WorkflowExecuteThread.java | 25 ++++++++++++----------
2 files changed, 17 insertions(+), 14 deletions(-)
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java
index d0e9d94..7eca55c 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/StateWheelExecuteThread.java
@@ -96,7 +96,7 @@ public class StateWheelExecuteThread extends Thread {
for (TaskInstance taskInstance : taskInstanceTimeoutCheckList.values()) {
if (TimeoutFlag.OPEN == taskInstance.getTaskDefine().getTimeoutFlag()) {
long timeRemain = DateUtils.getRemainTime(taskInstance.getStartTime(), taskInstance.getTaskDefine().getTimeout() * Constants.SEC_2_MINUTES_TIME_UNIT);
- if (0 >= timeRemain) {
+ if (timeRemain < 0) {
addTaskTimeoutEvent(taskInstance);
taskInstanceTimeoutCheckList.remove(taskInstance.getId());
}
@@ -127,7 +127,7 @@ public class StateWheelExecuteThread extends Thread {
for (ProcessInstance processInstance : this.processInstanceTimeoutCheckList.values()) {
long timeRemain = DateUtils.getRemainTime(processInstance.getStartTime(), processInstance.getTimeout() * Constants.SEC_2_MINUTES_TIME_UNIT);
- if (0 >= timeRemain) {
+ if (timeRemain < 0) {
addProcessTimeoutEvent(processInstance);
processInstanceTimeoutCheckList.remove(processInstance.getId());
}
@@ -162,7 +162,7 @@ public class StateWheelExecuteThread extends Thread {
}
private void addEvent(StateEvent stateEvent) {
- if (!processInstanceExecMaps.contains(stateEvent.getProcessInstanceId())) {
+ if (!processInstanceExecMaps.containsKey(stateEvent.getProcessInstanceId())) {
return;
}
WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecMaps.get(stateEvent.getProcessInstanceId());
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
index 53c0bf4..3b2ddca 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
@@ -669,12 +669,14 @@ public class WorkflowExecuteThread implements Runnable {
if (taskTimeoutCheckList.containsKey(taskInstance.getId())) {
return;
}
- TaskDefinition taskDefinition = processService.findTaskDefinition(
- taskInstance.getTaskCode(),
- taskInstance.getTaskDefinitionVersion()
- );
- taskInstance.setTaskDefine(taskDefinition);
- if (TimeoutFlag.OPEN == taskDefinition.getTimeoutFlag()) {
+ if (taskInstance.getTaskDefine() == null) {
+ TaskDefinition taskDefinition = processService.findTaskDefinition(
+ taskInstance.getTaskCode(),
+ taskInstance.getTaskDefinitionVersion()
+ );
+ taskInstance.setTaskDefine(taskDefinition);
+ }
+ if (TimeoutFlag.OPEN == taskInstance.getTaskDefine().getTimeoutFlag()) {
this.taskTimeoutCheckList.put(taskInstance.getId(), taskInstance);
}
if (taskInstance.isDependTask() || taskInstance.isSubProcess()) {
@@ -686,12 +688,13 @@ public class WorkflowExecuteThread implements Runnable {
if (taskRetryCheckList.containsKey(taskInstance.getId())) {
return;
}
- TaskDefinition taskDefinition = taskInstance.getTaskDefine();
- if (taskDefinition == null) {
- logger.error("taskDefinition is null, taskId:{}", taskInstance.getId());
- return;
+ if (taskInstance.getTaskDefine() == null) {
+ TaskDefinition taskDefinition = processService.findTaskDefinition(
+ taskInstance.getTaskCode(),
+ taskInstance.getTaskDefinitionVersion()
+ );
+ taskInstance.setTaskDefine(taskDefinition);
}
-
if (taskInstance.taskCanRetry()) {
this.taskRetryCheckList.put(taskInstance.getId(), taskInstance);
}