From 5faa5d170331b64b85b145716b41c9e49927f0d0 Mon Sep 17 00:00:00 2001
From: wuxiaofei <woshiwuxiaofei@126.com>
Date: Mon, 22 Feb 2021 10:20:22 +0800
Subject: [PATCH] [BUG][PeerTaskInstancePriorityQueue]poll implemented in
 PriorityQueue

---
 .../queue/PeerTaskInstancePriorityQueue.java    | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/PeerTaskInstancePriorityQueue.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/PeerTaskInstancePriorityQueue.java
index 5ce42242e..72d0bf612 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/PeerTaskInstancePriorityQueue.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/PeerTaskInstancePriorityQueue.java
@@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.service.exceptions.TaskPriorityQueueException
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.PriorityQueue;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Task instances priority queue implementation
@@ -61,6 +62,22 @@ public class PeerTaskInstancePriorityQueue implements TaskPriorityQueue<TaskInst
         return queue.poll();
     }
 
+    /**
+     * poll task info with timeout
+     * @param timeout
+     * @param unit
+     * @return
+     * @throws TaskPriorityQueueException
+     * @throws InterruptedException
+     */
+    @Override
+    public TaskInstance poll(long timeout, TimeUnit unit) throws TaskPriorityQueueException, InterruptedException {
+        if (queue.isEmpty()) {
+            unit.sleep(timeout);
+        }
+        return queue.poll();
+    }
+
     /**
      * peek taskInfo
      *
-- 
GitLab