Adjust the polling method of failed tasks.

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskDispatchFailedQueueConsumer.java

@@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
-import org.apache.dolphinscheduler.service.exceptions.TaskPriorityQueueException;
 import org.apache.dolphinscheduler.service.queue.TaskPriority;
 import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
 
@@ -94,22 +93,16 @@ public class TaskDispatchFailedQueueConsumer extends BaseDaemonThread {
     public void run() {
         while (Stopper.isRunning()) {
             try {
-                failedRetry();
+                if (taskDispatchFailedQueueImpl.size() > 0) {
+                    dispatchFailedBackToTaskPriorityQueue(masterConfig.getDispatchTaskNumber());
+                }
             } catch (Exception e) {
                 TaskMetrics.incTaskDispatchError();
                 logger.error("failed task retry error", e);
-            } finally {
-                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
             }
         }
     }
 
-    public void failedRetry() throws TaskPriorityQueueException {
-        if (taskDispatchFailedQueueImpl.size() > 0) {
-            retryConsumerThreadPoolExecutor.submit(() -> dispatchFailedBackToTaskPriorityQueue(masterConfig.getDispatchTaskNumber()));
-        }
-    }
-
     /**
      * put the failed dispatch task into the dispatch queue again
      */
@@ -120,21 +113,30 @@ public class TaskDispatchFailedQueueConsumer extends BaseDaemonThread {
                 if (Objects.isNull(dispatchFailedTaskPriority)) {
                     continue;
                 }
-                if (canRetry(dispatchFailedTaskPriority)) {
-                    dispatchFailedTaskPriority.setDispatchFailedRetryTimes(dispatchFailedTaskPriority.getDispatchFailedRetryTimes() + 1);
-                    taskPriorityQueueImpl.put(dispatchFailedTaskPriority);
-                } else {
-                    taskDispatchFailedQueueImpl.put(dispatchFailedTaskPriority);
-                }
+                retryConsumerThreadPoolExecutor.submit(() -> dispatchFailedTask(dispatchFailedTaskPriority));
             } catch (InterruptedException exception) {
                 logger.error("dispatch failed queue poll error", exception);
                 Thread.currentThread().interrupt();
             } catch (Exception e) {
                 logger.error("dispatch failed back to task priority queue error", e);
+            } finally {
+                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
             }
         }
     }
 
+    /**
+     * determine whether to retry and put into the dispatch queue when the conditions are met.
+     */
+    private void dispatchFailedTask(TaskPriority dispatchFailedTaskPriority) {
+        if (canRetry(dispatchFailedTaskPriority)) {
+            dispatchFailedTaskPriority.setDispatchFailedRetryTimes(dispatchFailedTaskPriority.getDispatchFailedRetryTimes() + 1);
+            taskPriorityQueueImpl.put(dispatchFailedTaskPriority);
+        } else {
+            taskDispatchFailedQueueImpl.put(dispatchFailedTaskPriority);
+        }
+    }
+
     /**
      * the time interval is adjusted according to the number of retries
      */