fix priority queue

Signed-off-by: dongjiang <dongjiang1989@126.com>

Author: dongjiang1989

pkg/controller/priorityqueue/priorityqueue.go

@@ -277,6 +277,13 @@ func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool)
 w.waiters.Add(1)
 
 w.notifyItemOrWaiterAdded()
+
+// ref: https://github.com/kubernetes-sigs/controller-runtime/issues/3239
+if w.shutdown.Load() == true {
+var zero T
+return zero, 0, true
+}
+
 item := <-w.get
 
 return item.Key, item.Priority, w.shutdown.Load()

pkg/controller/priorityqueue/priorityqueue_test.go

@@ -300,6 +300,26 @@ var _ = Describe("Controllerworkqueue", func() {
 Expect(metrics.depth["test"]).To(Equal(map[int]int{0: 2}))
 })
 
+// ref: https://github.com/kubernetes-sigs/controller-runtime/issues/3239
+It("Get from priority queue might get stuck when the priority queue is shut down", func() {
+q, _ := newQueue()
+
+q.Add("baz")
+// shut down
+q.ShutDown()
+q.AddWithOpts(AddOpts{After: time.Second}, "foo")
+
+item, priority, isShutDown := q.GetWithPriority()
+Expect(item).To(Equal(""))
+Expect(priority).To(Equal(0))
+Expect(isShutDown).To(Equal(true))
+
+item1, priority1, isShutDown := q.GetWithPriority()
+Expect(item1).To(Equal(""))
+Expect(priority1).To(Equal(0))
+Expect(isShutDown).To(Equal(true))
+})
+
 It("items are included in Len() and the queueDepth metric once they are ready", func() {
 q, metrics := newQueue()
 defer q.ShutDown()