fix priority queue

Signed-off-by: dongjiang <[email protected]>

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()