Merge 656988f into 23b2c8a

lihongyi87 · web-flow · commit d3259742e7da · 2023-10-11T06:44:29.000Z
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceDispatchOperationFunction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicITaskInstanceDispatchOperationFunction.java
@@ -62,19 +62,23 @@ public LogicTaskDispatchResponse operate(LogicTaskDispatchRequest taskDispatchRe
             LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath());
 
             MasterTaskExecutionContextHolder.putTaskExecutionContext(taskExecutionContext);
-            // todo: calculate the delay in master dispatcher then we don't need to use a queue to store the task
-            final long remainTime =
-                    DateUtils.getRemainTime(DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
-                            TimeUnit.SECONDS.toMillis(taskExecutionContext.getDelayTime()));
-            if (remainTime > 0) {
-                log.info("Current taskInstance: {} is choosing delay execution, delay time: {}/ms, remainTime: {}/ms",
-                        taskExecutionContext.getTaskName(),
-                        TimeUnit.SECONDS.toMillis(taskExecutionContext.getDelayTime()), remainTime);
-                taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
-                // todo: send delay execution message
-                return LogicTaskDispatchResponse.success(taskExecutionContext.getTaskInstanceId());
-            }
 
+            int delayTime = taskExecutionContext.getDelayTime();
+            if (delayTime > 0) {
+                // todo: calculate the delay in master dispatcher then we don't need to use a queue to store the task
+                final long remainTime =
+                        DateUtils.getRemainTime(DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
+                                TimeUnit.SECONDS.toMillis(delayTime));
+                if (remainTime > 0) {
+                    log.info(
+                            "Current taskInstance: {} is choosing delay execution, delay time: {}/ms, remainTime: {}/ms",
+                            taskExecutionContext.getTaskName(),
+                            TimeUnit.SECONDS.toMillis(taskExecutionContext.getDelayTime()), remainTime);
+                    taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
+                    // todo: send delay execution message
+                    return LogicTaskDispatchResponse.success(taskExecutionContext.getTaskInstanceId());
+                }
+            }
             final MasterDelayTaskExecuteRunnable masterDelayTaskExecuteRunnable =
                     masterTaskExecuteRunnableFactoryBuilder
                             .createWorkerDelayTaskExecuteRunnableFactory(taskExecutionContext.getTaskType())
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueueLooper.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/GlobalTaskInstanceDispatchQueueLooper.java
@@ -75,16 +75,21 @@ public void run() {
                 TaskExecutionContext taskExecutionContext = globalTaskInstanceDispatchQueue.take();
                 LogUtils.setTaskInstanceLogFullPathMDC(taskExecutionContext.getLogPath());
                 LogUtils.setTaskInstanceIdMDC(taskExecutionContext.getTaskInstanceId());
-                // delay task process
-                long remainTime =
-                        DateUtils.getRemainTime(DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
-                                taskExecutionContext.getDelayTime() * 60L);
-                if (remainTime > 0) {
-                    log.info("Current taskInstance is choose delay execution, delay time: {}s", remainTime);
-                    taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
-                    // todo: use delay running event
-                    workerMessageSender.sendMessage(taskExecutionContext,
-                            ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType.FINISH);
+
+                int delayTime = taskExecutionContext.getDelayTime();
+                if (delayTime > 0) {
+                    // delay task process
+                    long remainTime =
+                            DateUtils.getRemainTime(
+                                    DateUtils.timeStampToDate(taskExecutionContext.getFirstSubmitTime()),
+                                    delayTime * 60L);
+                    if (remainTime > 0) {
+                        log.info("Current taskInstance is choose delay execution, delay time: {}s", remainTime);
+                        taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.DELAY_EXECUTION);
+                        // todo: use delay running event
+                        workerMessageSender.sendMessage(taskExecutionContext,
+                                ITaskInstanceExecutionEvent.TaskInstanceExecutionEventType.FINISH);
+                    }
                 }
                 WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder
                         .createWorkerDelayTaskExecuteRunnableFactory(
