Make PopBatch process a single-item batch identically to Pop

staging/src/k8s.io/client-go/tools/cache/controller.go

@@ -231,7 +231,7 @@ func (c *controller) processLoop(ctx context.Context) {
 		default:
 			var err error
 			if useBatchProcess {
-				err = batchQueue.PopBatch(c.config.ProcessBatch)
+				err = batchQueue.PopBatch(c.config.ProcessBatch, PopProcessFunc(c.config.Process))
 			} else {
 				// otherwise fallback to non-batch process behavior
 				_, err = c.config.Pop(PopProcessFunc(c.config.Process))

staging/src/k8s.io/client-go/tools/cache/fifo.go

@@ -73,8 +73,12 @@ type QueueWithBatch interface {
 
 	// PopBatch behaves similarly to Queue#Pop, but processes multiple keys
 	// as a batch. The implementation determines the batching strategy,
-	// such as the number of keys to include per batch.
-	PopBatch(ProcessBatchFunc) error
+	// such as the number of keys to include per batch. The ProcessBatchFunc
+	// is called when a batch is ready to be processed. The PopProcessFunc
+	// is called when a singleton item is ready to be processed. The
+	// ProcessBatchFunc and PopProcessFunc must do the same processing to
+	// ensure consistent behavior.
+	PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error
 }
 
 // Pop is helper function for popping from Queue.

staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go

@@ -264,7 +264,9 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 	return Deltas{item}, err
 }
 
-func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error {
+// PopBatch pops as many items as possible to be processed as a batch using processBatch,
+// or pop a single item using processSingle if multiple items cannot be batched.
+func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProcessFunc) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
@@ -304,6 +306,7 @@ func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error {
 			break
 		}
 		if unique.Has(id) {
+			// close the batch if a duplicate item is encountered
 			break
 		}
 		unique.Insert(id)
@@ -329,8 +332,10 @@ func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error {
 		defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second))
 	}
 
-	err := process(deltas, isInInitialList)
-	return err
+	if len(deltas) == 1 {
+		return processSingle(Deltas{deltas[0]}, isInInitialList)
+	}
+	return processBatch(deltas, isInInitialList)
 }
 
 // Replace

staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go

@@ -1141,7 +1141,13 @@ func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) {
 							receivedInitial <- obj
 						}
 						return nil
-					})
+					}, PopProcessFunc(func(obj interface{}, isInInitialList bool) error {
+						received <- []Delta{*obj.(Deltas).Newest()}
+						if isInInitialList {
+							receivedInitial <- []Delta{*obj.(Deltas).Newest()}
+						}
+						return nil
+					}))
 				}()
 				timer := time.NewTimer(time.Millisecond * 50)
 				select {
@@ -1265,7 +1271,10 @@ func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) {
 					_ = f.PopBatch(func(obj []Delta, isInInitialList bool) error {
 						received <- obj
 						return nil
-					})
+					}, PopProcessFunc(func(obj interface{}, isInInitialList bool) error {
+						received <- []Delta{*obj.(Deltas).Newest()}
+						return nil
+					}))
 				}()
 				timer := time.NewTimer(time.Millisecond * 50)
 				select {