[dolphinscheduler] branch 2.0.1-prepare updated: fix timeout (#7224)

[leonbao]

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 <caishunf...@users.noreply.github.com>
AuthorDate: Mon Dec 6 22:54:49 2021 +0800

    fix timeout (#7224)
    
    Co-authored-by: caishunfeng <534328...@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);
         }