Fix task mighe be dispatched even if it has been killed

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueue.java

@@ -19,6 +19,7 @@
 
 import java.util.concurrent.DelayQueue;
 
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.stereotype.Component;
@@ -37,7 +38,8 @@ public void submitTaskExecuteRunnable(DefaultTaskExecuteRunnable priorityTaskExe
         queue.put(priorityTaskExecuteRunnable);
     }
 
-    public DefaultTaskExecuteRunnable takeTaskExecuteRunnable() throws InterruptedException {
+    @SneakyThrows
+    public DefaultTaskExecuteRunnable takeTaskExecuteRunnable() {
         return queue.take();
     }
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueueLooper.java

@@ -19,6 +19,7 @@
 
 import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatchFactory;
 import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatcher;
 
@@ -65,14 +66,15 @@ public synchronized void start() {
     public void run() {
         DefaultTaskExecuteRunnable defaultTaskExecuteRunnable;
         while (RUNNING_FLAG.get()) {
+            defaultTaskExecuteRunnable = globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable();
             try {
-                defaultTaskExecuteRunnable = globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable();
-            } catch (InterruptedException e) {
-                log.warn("Get waiting dispatch task failed, the current thread has been interrupted, will stop loop");
-                Thread.currentThread().interrupt();
-                break;
-            }
-            try {
+                TaskExecutionStatus status = defaultTaskExecuteRunnable.getTaskInstance().getState();
+                if (status != TaskExecutionStatus.SUBMITTED_SUCCESS) {
+                    log.warn("The TaskInstance {} state is : {}, will not dispatch",
+                            defaultTaskExecuteRunnable.getTaskInstance().getName(), status);
+                    continue;
+                }
+
                 TaskDispatcher taskDispatcher =
                         taskDispatchFactory.getTaskDispatcher(defaultTaskExecuteRunnable.getTaskInstance());
                 taskDispatcher.dispatchTask(defaultTaskExecuteRunnable);
@@ -86,14 +88,15 @@ public void run() {
                 log.error("Dispatch Task: {} failed", defaultTaskExecuteRunnable.getTaskInstance().getName(), e);
             }
         }
-        log.info("GlobalTaskDispatchWaitingQueueLooper started...");
     }
 
     @Override
     public void close() throws Exception {
         if (RUNNING_FLAG.compareAndSet(true, false)) {
             log.info("GlobalTaskDispatchWaitingQueueLooper stopping...");
             log.info("GlobalTaskDispatchWaitingQueueLooper stopped...");
+        } else {
+            log.error("GlobalTaskDispatchWaitingQueueLooper is not started");
         }
     }
 }

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueueLooperTest.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.runner;
+
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatchFactory;
+import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatcher;
+import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class GlobalTaskDispatchWaitingQueueLooperTest {
+
+    @InjectMocks
+    private GlobalTaskDispatchWaitingQueueLooper globalTaskDispatchWaitingQueueLooper;
+
+    @Mock
+    private GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue;
+
+    @Mock
+    private TaskDispatchFactory taskDispatchFactory;
+
+    @Test
+    void testTaskExecutionRunnableStatusIsNotSubmitted() throws Exception {
+        ProcessInstance processInstance = new ProcessInstance();
+        TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setState(TaskExecutionStatus.KILL);
+        TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
+        TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager =
+                new TaskExecuteRunnableOperatorManager();
+        DefaultTaskExecuteRunnable defaultTaskExecuteRunnable = new DefaultTaskExecuteRunnable(processInstance,
+                taskInstance, taskExecutionContext, taskExecuteRunnableOperatorManager);
+
+        when(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable()).thenReturn(defaultTaskExecuteRunnable);
+        globalTaskDispatchWaitingQueueLooper.start();
+        Thread.sleep(1000);
+        verify(taskDispatchFactory, Mockito.never()).getTaskDispatcher(taskInstance);
+        globalTaskDispatchWaitingQueueLooper.close();
+    }
+
+    @Test
+    void testTaskExecutionRunnableStatusIsSubmitted() throws Exception {
+        ProcessInstance processInstance = new ProcessInstance();
+        TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setState(TaskExecutionStatus.SUBMITTED_SUCCESS);
+        TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
+        TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager =
+                new TaskExecuteRunnableOperatorManager();
+        DefaultTaskExecuteRunnable defaultTaskExecuteRunnable = new DefaultTaskExecuteRunnable(processInstance,
+                taskInstance, taskExecutionContext, taskExecuteRunnableOperatorManager);
+
+        TaskDispatcher taskDispatcher = mock(TaskDispatcher.class);
+        when(taskDispatchFactory.getTaskDispatcher(taskInstance)).thenReturn(taskDispatcher);
+        doNothing().when(taskDispatcher).dispatchTask(Mockito.any());
+
+        when(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable()).thenReturn(defaultTaskExecuteRunnable);
+        globalTaskDispatchWaitingQueueLooper.start();
+        Thread.sleep(1000);
+        verify(taskDispatchFactory, Mockito.atLeastOnce()).getTaskDispatcher(Mockito.any(TaskInstance.class));
+        verify(taskDispatcher, Mockito.atLeastOnce()).dispatchTask(Mockito.any(TaskExecuteRunnable.class));
+        globalTaskDispatchWaitingQueueLooper.close();
+
+    }
+}