Adding batch handling for popping items from RealFIFO

Signed-off-by: Min Jin <minkimzz@amazon.com>

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

optimizing fifo loop

Co-authored-by: Marek Siarkowicz <marek.siarkowicz@protonmail.com>
Signed-off-by: Min Jin <minkimzz@amazon.com>

refactoring PopBatch to accept []Delta

Signed-off-by: Min Jin <minkimzz@amazon.com>

staging/src/k8s.io/client-go/features/known_features.go

@@ -58,6 +58,12 @@ const (
 	// Refactor informers to deliver watch stream events in order instead of out of order.
 	InOrderInformers Feature = "InOrderInformers"
 
+	// owner: @yue9944882
+	// beta: v1.35
+	//
+	// Allow InOrderInformer to process incoming events in batches to expedite the process rate.
+	InOrderInformersBatchProcess Feature = "InOrderInformersBatchProcess"
+
 	// owner: @enj, @michaelasp
 	// alpha: v1.30
 	// GA: v1.35
@@ -89,6 +95,9 @@ var defaultVersionedKubernetesFeatureGates = map[Feature]VersionedSpecs{
 	InOrderInformers: {
 		{Version: version.MustParse("1.33"), Default: true, PreRelease: Beta},
 	},
+	InOrderInformersBatchProcess: {
+		{Version: version.MustParse("1.35"), Default: true, PreRelease: Beta},
+	},
 	InformerResourceVersion: {
 		{Version: version.MustParse("1.30"), Default: false, PreRelease: Alpha},
 		{Version: version.MustParse("1.35"), Default: true, PreRelease: GA},

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

@@ -19,6 +19,7 @@ package cache
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -48,9 +49,20 @@ type Config struct {
 	// Something that can list and watch your objects.
 	ListerWatcher
 
-	// Something that can process a popped Deltas.
+	// Process can process a popped Deltas.
 	Process ProcessFunc
 
+	// ProcessBatch can process a batch of popped Deltas, which should return `TransactionError` if not all items
+	// in the batch were successfully processed.
+	//
+	// For batch processing to be used:
+	// * ProcessBatch must be non-nil
+	// * Queue must implement QueueWithBatch
+	// * The client InOrderInformersBatchProcess feature gate must be enabled
+	//
+	// If any of those are false, Process is used and no batch processing is done.
+	ProcessBatch ProcessBatchFunc
+
 	// ObjectType is an example object of the type this controller is
 	// expected to handle.
 	ObjectType runtime.Object
@@ -94,6 +106,10 @@ type ShouldResyncFunc func() bool
 // ProcessFunc processes a single object.
 type ProcessFunc func(obj interface{}, isInInitialList bool) error
 
+// ProcessBatchFunc processes multiple objects in batch.
+// The deltas must not contain multiple entries for the same object.
+type ProcessBatchFunc func(deltas []Delta, isInInitialList bool) error
+
 // `*controller` implements Controller
 type controller struct {
 	config         Config
@@ -203,12 +219,23 @@ func (c *controller) LastSyncResourceVersion() string {
 // to make sure that we don't end up processing the same object multiple times
 // concurrently.
 func (c *controller) processLoop(ctx context.Context) {
+	useBatchProcess := false
+	batchQueue, ok := c.config.Queue.(QueueWithBatch)
+	if ok && c.config.ProcessBatch != nil && clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.InOrderInformersBatchProcess) {
+		useBatchProcess = true
+	}
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		default:
-			_, err := c.config.Pop(PopProcessFunc(c.config.Process))
+			var err error
+			if useBatchProcess {
+				err = batchQueue.PopBatch(c.config.ProcessBatch)
+			} else {
+				// otherwise fallback to non-batch process behavior
+				_, err = c.config.Pop(PopProcessFunc(c.config.Process))
+			}
 			if err != nil {
 				if errors.Is(err, ErrFIFOClosed) {
 					return
@@ -585,6 +612,91 @@ func processDeltas(
 	return nil
 }
 
+// processDeltasInBatch applies a batch of Delta objects to the given Store and
+// notifies the ResourceEventHandler of add, update, or delete events.
+//
+// If the Store supports transactions (TransactionStore), all Deltas are applied
+// atomically in a single transaction and corresponding handler callbacks are
+// executed afterward. Otherwise, each Delta is processed individually.
+//
+// Returns an error if any Delta or transaction fails. For TransactionError,
+// only successful operations trigger callbacks.
+func processDeltasInBatch(
+	handler ResourceEventHandler,
+	clientState Store,
+	deltas []Delta,
+	isInInitialList bool,
+) error {
+	// from oldest to newest
+	txns := make([]Transaction, 0)
+	callbacks := make([]func(), 0)
+	txnStore, txnSupported := clientState.(TransactionStore)
+	if !txnSupported {
+		var errs []error
+		for _, delta := range deltas {
+			if err := processDeltas(handler, clientState, Deltas{delta}, isInInitialList); err != nil {
+				errs = append(errs, err)
+			}
+		}
+		if len(errs) > 0 {
+			return fmt.Errorf("unexpected error when handling deltas: %v", errs)
+		}
+		return nil
+	}
+	// deltasList is a list of unique objects
+	for _, d := range deltas {
+		obj := d.Object
+		switch d.Type {
+		case Sync, Replaced, Added, Updated:
+			// it will only return one old object for each because items are unique
+			if old, exists, err := clientState.Get(obj); err == nil && exists {
+				txn := Transaction{
+					Type:   TransactionTypeUpdate,
+					Object: obj,
+				}
+				txns = append(txns, txn)
+				callbacks = append(callbacks, func() {
+					handler.OnUpdate(old, obj)
+				})
+			} else {
+				txn := Transaction{
+					Type:   TransactionTypeAdd,
+					Object: obj,
+				}
+				txns = append(txns, txn)
+				callbacks = append(callbacks, func() {
+					handler.OnAdd(obj, isInInitialList)
+				})
+			}
+		case Deleted:
+			txn := Transaction{
+				Type:   TransactionTypeDelete,
+				Object: obj,
+			}
+			txns = append(txns, txn)
+			callbacks = append(callbacks, func() {
+				handler.OnDelete(obj)
+			})
+		}
+	}
+
+	err := txnStore.Transaction(txns...)
+	if err != nil {
+		// if txn had error, only execute the callbacks for the successful ones
+		for _, i := range err.SuccessfulIndices {
+			if i < len(callbacks) {
+				callbacks[i]()
+			}
+		}
+		// formatting the error so txns doesn't escape and keeps allocated in the stack.
+		return fmt.Errorf("not all items in the batch successfully processed: %s", err.Error())
+	}
+	for _, callback := range callbacks {
+		callback()
+	}
+	return nil
+}
+
 // newInformer returns a controller for populating the store while also
 // providing event notifications.
 //
@@ -624,6 +736,9 @@ func newInformer(clientState Store, options InformerOptions) Controller {
 			}
 			return errors.New("object given as Process argument is not Deltas")
 		},
+		ProcessBatch: func(deltaList []Delta, isInInitialList bool) error {
+			return processDeltasInBatch(options.Handler, clientState, deltaList, isInInitialList)
+		},
 	}
 	return New(cfg)
 }

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

@@ -18,6 +18,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"sync"
@@ -25,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -747,3 +749,147 @@ func toListWatcherWithUnSupportedWatchListSemantics(lw *ListWatch) ListerWatcher
 		lw,
 	}
 }
+
+func TestProcessDeltasInBatch(t *testing.T) {
+	cm1 := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testname1",
+			Namespace: "testnamespace",
+		},
+	}
+	cm2 := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testname2",
+			Namespace: "testnamespace",
+		},
+	}
+	cm3 := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testname3",
+			Namespace: "testnamespace",
+		},
+	}
+	testDelta1 := Delta{
+		Type:   Added,
+		Object: cm1,
+	}
+	testDelta2 := Delta{
+		Type:   Added,
+		Object: cm2,
+	}
+	testDelta3 := Delta{
+		Type:   Added,
+		Object: cm3,
+	}
+	testCases := []struct {
+		name                            string
+		deltaList                       []Delta
+		succeedingObjects               []runtime.Object
+		failingObjects                  []runtime.Object
+		expectedListenerReceivedObjects []interface{}
+		expectedSuccessCount            int
+		assertErr                       func(error) bool
+	}{
+		{
+			name:                            "all transaction succeeding should works",
+			deltaList:                       []Delta{testDelta1},
+			expectedSuccessCount:            1,
+			expectedListenerReceivedObjects: []interface{}{cm1},
+		},
+		{
+			name:                 "all transaction failing should not trigger listener",
+			deltaList:            []Delta{testDelta1},
+			failingObjects:       []runtime.Object{cm1},
+			expectedSuccessCount: 0,
+			assertErr: func(err error) bool {
+				return assert.Contains(t, err.Error(), "failed to execute (1/1) transactions")
+			},
+			expectedListenerReceivedObjects: make([]interface{}, 0),
+		},
+		{
+			name:                 "partial transaction failing should only trigger successful events to listener #1",
+			deltaList:            []Delta{testDelta1, testDelta2},
+			failingObjects:       []runtime.Object{cm2},
+			expectedSuccessCount: 1,
+			assertErr: func(err error) bool {
+				return assert.Contains(t, err.Error(), "failed to execute (1/2) transactions")
+			},
+			expectedListenerReceivedObjects: []interface{}{cm1},
+		},
+		{
+			name:                 "partial transaction failing should only trigger successful events to listener #2",
+			deltaList:            []Delta{testDelta1, testDelta2, testDelta3},
+			failingObjects:       []runtime.Object{cm2},
+			expectedSuccessCount: 2,
+			assertErr: func(err error) bool {
+				return assert.Contains(t, err.Error(), "failed to execute (1/3) transactions")
+			},
+			expectedListenerReceivedObjects: []interface{}{cm1, cm3},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockStore := &mockTxnStore{
+				Store:       NewStore(MetaNamespaceKeyFunc),
+				failingObjs: tc.failingObjects,
+			}
+			actualListenerReceivedObjects := make([]interface{}, 0)
+			dummyListener := ResourceEventHandlerFuncs{
+				AddFunc: func(obj interface{}) {
+					actualListenerReceivedObjects = append(actualListenerReceivedObjects, obj)
+				},
+				UpdateFunc: func(oldObj, newObj interface{}) {
+					actualListenerReceivedObjects = append(actualListenerReceivedObjects, newObj)
+				},
+				DeleteFunc: func(obj interface{}) {
+					actualListenerReceivedObjects = append(actualListenerReceivedObjects, obj)
+				},
+			}
+			err := processDeltasInBatch(
+				dummyListener,
+				mockStore,
+				tc.deltaList,
+				true)
+			if tc.assertErr != nil {
+				assert.True(t, tc.assertErr(err))
+			}
+			assert.Equal(t, tc.expectedSuccessCount, mockStore.succeedCount)
+			assert.Equal(t, tc.expectedListenerReceivedObjects, actualListenerReceivedObjects)
+		})
+	}
+}
+
+var _ TransactionStore = &mockTxnStore{}
+
+type mockTxnStore struct {
+	failingObjs  []runtime.Object
+	succeedCount int
+	Store
+}
+
+func (m *mockTxnStore) Transaction(txns ...Transaction) *TransactionError {
+	successfuls := make([]int, 0)
+	fails := make([]int, 0)
+	errs := make([]error, 0)
+	for i := range txns {
+		txn := txns[i]
+		for _, fail := range m.failingObjs {
+			if apiequality.Semantic.DeepEqual(fail, txn.Object) {
+				fails = append(fails, i)
+				errs = append(errs, errors.New("test error"))
+			} else {
+				successfuls = append(successfuls, i)
+			}
+		}
+	}
+	if len(fails) > 0 {
+		m.succeedCount = len(successfuls)
+		return &TransactionError{
+			TotalTransactions: len(txns),
+			SuccessfulIndices: successfuls,
+			Errors:            errs,
+		}
+	}
+	m.succeedCount = len(txns)
+	return nil
+}

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

@@ -60,6 +60,23 @@ type Queue interface {
 	Close()
 }
 
+// QueueWithBatch extends the Queue interface with support for batch processing.
+//
+// In addition to the standard single-item Pop method, QueueWithBatch provides
+// PopBatch, which allows multiple items to be popped and processed together as
+// a batch. This can be used to improve processing efficiency when it is
+// beneficial to handle multiple queued keys or accumulators in a single
+// operation.
+// TODO: Consider merging this interface into Queue after feature gate GA
+type QueueWithBatch interface {
+	Queue
+
+	// 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
+}
+
 // Pop is helper function for popping from Queue.
 // WARNING: Do NOT use this function in non-test code to avoid races
 // unless you really really really really know what you are doing.

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

@@ -563,6 +563,7 @@ func (s *sharedIndexInformer) RunWithContext(ctx context.Context) {
 			ShouldResync:      s.processor.shouldResync,
 
 			Process:                      s.HandleDeltas,
+			ProcessBatch:                 s.HandleBatchDeltas,
 			WatchErrorHandlerWithContext: s.watchErrorHandler,
 		}
 
@@ -735,6 +736,12 @@ func (s *sharedIndexInformer) HandleDeltas(obj interface{}, isInInitialList bool
 	return errors.New("object given as Process argument is not Deltas")
 }
 
+func (s *sharedIndexInformer) HandleBatchDeltas(deltas []Delta, isInInitialList bool) error {
+	s.blockDeltas.Lock()
+	defer s.blockDeltas.Unlock()
+	return processDeltasInBatch(s, s.indexer, deltas, isInInitialList)
+}
+
 // Conforms to ResourceEventHandler
 func (s *sharedIndexInformer) OnAdd(obj interface{}, isInInitialList bool) {
 	// Invocation of this function is locked under s.blockDeltas, so it is

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

@@ -17,6 +17,7 @@ limitations under the License.
 package cache
 
 import (
+	"errors"
 	"fmt"
 	"strings"
 
@@ -71,6 +72,42 @@ type Store interface {
 	Resync() error
 }
 
+// TransactionType defines the type of a transaction operation. It is used to indicate whether
+// an object is being added, updated, or deleted.
+type TransactionType string
+
+const (
+	TransactionTypeAdd    TransactionType = "Add"
+	TransactionTypeUpdate TransactionType = "Update"
+	TransactionTypeDelete TransactionType = "Delete"
+)
+
+// Transaction represents a single operation or event in a process. It holds a generic Object
+// associated with the transaction and a Type indicating the kind of transaction being performed.
+type Transaction struct {
+	Object interface{}
+	Type   TransactionType
+}
+
+type TransactionStore interface {
+	// Transaction allows multiple operations to occur within a single lock acquisition to
+	// ensure progress can be made when there is contention.
+	Transaction(txns ...Transaction) *TransactionError
+}
+
+var _ error = &TransactionError{}
+
+type TransactionError struct {
+	SuccessfulIndices []int
+	TotalTransactions int
+	Errors            []error
+}
+
+func (t *TransactionError) Error() string {
+	return fmt.Sprintf("failed to execute (%d/%d) transactions failed due to: %v",
+		t.TotalTransactions-len(t.SuccessfulIndices), t.TotalTransactions, t.Errors)
+}
+
 // KeyFunc knows how to make a key from an object. Implementations should be deterministic.
 type KeyFunc func(obj interface{}) (string, error)
 
@@ -167,6 +204,40 @@ type cache struct {
 
 var _ Store = &cache{}
 
+func (c *cache) Transaction(txns ...Transaction) *TransactionError {
+	txnStore, ok := c.cacheStorage.(ThreadSafeStoreWithTransaction)
+	if !ok {
+		return &TransactionError{
+			TotalTransactions: len(txns),
+			Errors: []error{
+				errors.New("transaction not supported"),
+			},
+		}
+	}
+	keyedTxns := make([]ThreadSafeStoreTransaction, 0, len(txns))
+	successfulIndices := make([]int, 0, len(txns))
+	errs := make([]error, 0)
+	for i := range txns {
+		txn := txns[i]
+		key, err := c.keyFunc(txn.Object)
+		if err != nil {
+			errs = append(errs, KeyError{txn.Object, err})
+			continue
+		}
+		successfulIndices = append(successfulIndices, i)
+		keyedTxns = append(keyedTxns, ThreadSafeStoreTransaction{txn, key})
+	}
+	txnStore.Transaction(keyedTxns...)
+	if len(errs) > 0 {
+		return &TransactionError{
+			SuccessfulIndices: successfulIndices,
+			TotalTransactions: len(txns),
+			Errors:            errs,
+		}
+	}
+	return nil
+}
+
 // Add inserts an item into the cache.
 func (c *cache) Add(obj interface{}) error {
 	key, err := c.keyFunc(obj)

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

@@ -21,6 +21,7 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
@@ -186,3 +187,62 @@ func TestKeyError(t *testing.T) {
 		t.Errorf("not match target error: %v", err)
 	}
 }
+
+func TestCacheTransactionShouldIndexErrors(t *testing.T) {
+	successObj1 := 1
+	successObj2 := 2
+	failObj1 := 3
+	failObj2 := 4
+	testTxnType := TransactionTypeAdd
+	testCases := []struct {
+		name        string
+		objs        []interface{}
+		assertError func(*TransactionError) bool
+	}{
+		{
+			name: "txn all success objects should work",
+			objs: []interface{}{successObj1, successObj2},
+		},
+		{
+			name: "txn all fail objects should work",
+			objs: []interface{}{failObj1, failObj2},
+			assertError: func(err *TransactionError) bool {
+				return assert.Equal(t, []int{}, err.SuccessfulIndices)
+			},
+		},
+		{
+			name: "txn mix success and fail objects should work",
+			objs: []interface{}{successObj1, failObj1, successObj2, failObj2},
+			assertError: func(err *TransactionError) bool {
+				return assert.Equal(t, []int{0, 2}, err.SuccessfulIndices)
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			testKeyFunc := func(obj interface{}) (string, error) {
+				if obj == successObj1 || obj == successObj2 {
+					return "", nil
+				}
+				return "", errors.New("test error")
+			}
+			testStore := NewStore(testKeyFunc)
+			txnStore := testStore.(TransactionStore)
+			txns := make([]Transaction, len(tc.objs))
+			for i := range tc.objs {
+				txns[i] = Transaction{
+					Object: tc.objs[i],
+					Type:   testTxnType,
+				}
+			}
+			txnErr := txnStore.Transaction(txns...)
+			if tc.assertError != nil {
+				tc.assertError(txnErr)
+				return
+			}
+			if txnErr != nil {
+				t.Errorf("unexpected error: %v", txnErr)
+			}
+		})
+	}
+}

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

@@ -45,6 +45,12 @@ type RealFIFOOptions struct {
 	Transformer TransformFunc
 }
 
+const (
+	defaultBatchSize = 1000
+)
+
+var _ QueueWithBatch = &RealFIFO{}
+
 // RealFIFO is a Queue in which every notification from the Reflector is passed
 // in order to the Queue via Pop.
 // This means that it
@@ -77,6 +83,9 @@ type RealFIFO struct {
 
 	// Called with every object if non-nil.
 	transformer TransformFunc
+
+	// batchSize determines the maximum number of objects we can combine into a batch.
+	batchSize int
 }
 
 var (
@@ -254,6 +263,74 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 	return Deltas{item}, err
 }
 
+func (f *RealFIFO) PopBatch(process ProcessBatchFunc) error {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	for len(f.items) == 0 {
+		// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
+		// When Close() is called, the f.closed is set and the condition is broadcasted.
+		// Which causes this loop to continue and return from the Pop().
+		if f.closed {
+			return ErrFIFOClosed
+		}
+
+		f.cond.Wait()
+	}
+
+	isInInitialList := !f.hasSynced_locked()
+	unique := sets.NewString()
+	deltas := make([]Delta, 0, min(len(f.items), f.batchSize))
+	// only bundle unique items into a batch
+	for i := 0; i < f.batchSize && i < len(f.items); i++ {
+		if f.initialPopulationCount > 0 && i >= f.initialPopulationCount {
+			break
+		}
+		item := f.items[i]
+		id, err := f.keyOf(item)
+		if err != nil {
+			// close the batch here if error happens
+			// TODO: log the error when RealFIFOOptions supports passing klog instance like deprecated DeltaFIFO
+			// still pop the broken item out of queue to be compatible with the non-batch behavior it should be safe
+			// when 1st element is broken, however for Nth broken element, there's possible risk that broken item
+			// still can be processed and broke the uniqueness of the batch unexpectedly.
+			deltas = append(deltas, item)
+			// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
+			f.items[i] = Delta{}
+			break
+		}
+		if unique.Has(id) {
+			break
+		}
+		unique.Insert(id)
+		deltas = append(deltas, item)
+		// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
+		f.items[i] = Delta{}
+	}
+	if f.initialPopulationCount > 0 {
+		f.initialPopulationCount -= len(deltas)
+	}
+	f.items = f.items[len(deltas):]
+
+	// Only log traces if the queue depth is greater than 10 and it takes more than
+	// 100 milliseconds to process one item from the queue (with a max of 1 second for the whole batch)
+	// Queue depth never goes high because processing an item is locking the queue,
+	// and new items can't be added until processing finish.
+	// https://github.com/kubernetes/kubernetes/issues/103789
+	if len(f.items) > 10 {
+		id, _ := f.keyOf(deltas[0])
+		trace := utiltrace.New("RealFIFO PopBatch Process",
+			utiltrace.Field{Key: "ID", Value: id},
+			utiltrace.Field{Key: "Depth", Value: len(f.items)},
+			utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"},
+			utiltrace.Field{Key: "BatchSize", Value: len(deltas)})
+		defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second))
+	}
+
+	err := process(deltas, isInInitialList)
+	return err
+}
+
 // Replace
 // 1. finds those items in f.items that are not in newItems and creates synthetic deletes for them
 // 2. finds items in knownObjects that are not in newItems and creates synthetic deletes for them
@@ -440,6 +517,7 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO {
 		keyFunc:      opts.KeyFunction,
 		knownObjects: opts.KnownObjects,
 		transformer:  opts.Transformer,
+		batchSize:    defaultBatchSize,
 	}
 
 	f.cond.L = &f.lock

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

@@ -17,11 +17,16 @@ limitations under the License.
 package cache
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"runtime"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
+	clientfeatures "k8s.io/client-go/features"
+	clientfeaturestesting "k8s.io/client-go/features/testing"
 )
 
 func (f *RealFIFO) getItems() []Delta {
@@ -974,3 +979,313 @@ func TestRealFIFO_PopShouldUnblockWhenClosed(t *testing.T) {
 		}
 	}
 }
+
+func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.InOrderInformersBatchProcess, true)
+	const unlimitedBatchSize = 999999
+	obj1 := mkFifoObj("foo1", 5)
+	obj2 := mkFifoObj("foo2", 5)
+	obj3 := mkFifoObj("foo3", 5)
+	obj4 := mkFifoObj("foo4", 5)
+	testCases := []struct {
+		name            string
+		initialItems    []testFifoObject
+		actions         []func(f *RealFIFO)
+		batchSize       int
+		expectedBatches [][]Delta
+	}{
+		{
+			name: "non-split: pop unique items should work",
+			initialItems: []testFifoObject{
+				obj1, obj2, obj3,
+			},
+			actions:   []func(f *RealFIFO){},
+			batchSize: unlimitedBatchSize,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}, {Replaced, obj3}},
+			},
+		},
+		{
+			name: "split due to initial list: initial 2 items with 2 updates should have 2 batches",
+			initialItems: []testFifoObject{
+				obj1, obj2,
+			},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj3) },
+				func(f *RealFIFO) { _ = f.Update(obj4) },
+			},
+			batchSize: 2,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}},
+				{{Updated, obj3}, {Updated, obj4}},
+			},
+		},
+		{
+			name: "split due to non-unique#1: update single item for multiple items should have separate batch",
+			initialItems: []testFifoObject{
+				obj1,
+			},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj1) },
+				func(f *RealFIFO) { _ = f.Update(obj1) },
+			},
+			batchSize: unlimitedBatchSize,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}},
+				{{Updated, obj1}},
+				{{Updated, obj1}},
+			},
+		},
+		{
+			name: "split due to non-unique#2: update 3 item for with non-unique item at the end should have separate batch",
+			initialItems: []testFifoObject{
+				obj1, obj2,
+			},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj2) },
+				func(f *RealFIFO) { _ = f.Update(obj3) },
+			},
+			batchSize: unlimitedBatchSize,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}},
+				{{Updated, obj2}, {Updated, obj3}},
+			},
+		},
+		{
+			name: "split due to non-unique#3: update 3 item for with non-unique item in the mid should have separate batch",
+			initialItems: []testFifoObject{
+				obj1, obj2, obj3,
+			},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj2) },
+			},
+			batchSize: unlimitedBatchSize,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}, {Replaced, obj3}},
+				{{Updated, obj2}},
+			},
+		},
+		{
+			name: "split due to batch size#1: batching initial list should work",
+			initialItems: []testFifoObject{
+				obj1, obj2, obj3,
+			},
+			actions:   []func(f *RealFIFO){},
+			batchSize: 2,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}},
+				{{Replaced, obj3}},
+			},
+		},
+		{
+			name:         "split due to batch size#2: batching incoming non-initial deltas should work",
+			initialItems: []testFifoObject{},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj1) },
+				func(f *RealFIFO) { _ = f.Update(obj2) },
+				func(f *RealFIFO) { _ = f.Update(obj3) },
+			},
+			batchSize: 2,
+			expectedBatches: [][]Delta{
+				{{Updated, obj1}, {Updated, obj2}},
+				{{Updated, obj3}},
+			},
+		},
+		{
+			name: "split due to batch size#3: pop 4 mixed initial & non-initial items with 2 batch size should have 3 batch",
+			initialItems: []testFifoObject{
+				obj1, obj2, obj3,
+			},
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { _ = f.Update(obj4) },
+			},
+			batchSize: 2,
+			expectedBatches: [][]Delta{
+				{{Replaced, obj1}, {Replaced, obj2}},
+				{{Replaced, obj3}},
+				{{Updated, obj4}},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			f := NewRealFIFO(
+				testFifoObjectKeyFunc,
+				literalListerGetter(func() []testFifoObject {
+					return tc.initialItems
+				}),
+				nil)
+			f.batchSize = tc.batchSize
+
+			initialItems := make([]interface{}, len(tc.initialItems))
+			for i, item := range tc.initialItems {
+				initialItems[i] = item
+			}
+			_ = f.Replace(initialItems, "")
+			for _, action := range tc.actions {
+				action(f)
+			}
+
+			const maxAttempts = 10
+			receivedItems := make([][]Delta, 0)
+			receivedInitialDeltas := make([][]Delta, 0)
+
+			for i := 0; i < maxAttempts; i++ {
+				received := make(chan []Delta, 100)
+				receivedInitial := make(chan []Delta, 100)
+				go func() {
+					_ = f.PopBatch(func(obj []Delta, isInInitialList bool) error {
+						received <- obj
+						if isInInitialList {
+							receivedInitial <- obj
+						}
+						return nil
+					})
+				}()
+				timer := time.NewTimer(time.Millisecond * 50)
+				select {
+				case <-timer.C:
+					close(received)
+				case item := <-received:
+					receivedItems = append(receivedItems, item)
+					close(received)
+				}
+				timerInitial := time.NewTimer(time.Millisecond * 50)
+				select {
+				case <-timerInitial.C:
+					close(receivedInitial)
+				case item := <-receivedInitial:
+					receivedInitialDeltas = append(receivedInitialDeltas, item)
+					close(receivedInitial)
+				}
+			}
+
+			runtime.Gosched()
+			f.Close()
+
+			idx := 0
+			for _, batch := range tc.expectedBatches {
+				assert.Equal(t, receivedItems[idx], batch)
+				idx++
+			}
+			receivedInitialItems := make([]testFifoObject, 0)
+			for _, deltas := range receivedInitialDeltas {
+				for _, delta := range deltas {
+					receivedInitialItems = append(receivedInitialItems, delta.Object.(testFifoObject))
+				}
+			}
+
+			assert.Equal(t, tc.initialItems, receivedInitialItems)
+		})
+	}
+}
+
+func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.InOrderInformersBatchProcess, true)
+	const unlimitedBatchSize = 999999
+	sucessObj1 := mkFifoObj("foo1", 5)
+	sucessObj2 := mkFifoObj("foo2", 5)
+	failObj3 := mkFifoObj("foo3", 5)
+	failObj4 := mkFifoObj("foo4", 5)
+	testDeltaType := Added
+	testCases := []struct {
+		name            string
+		batchSize       int
+		incomingItems   []testFifoObject
+		expectedBatches [][]Delta
+	}{
+		{
+			name: "1st item is broken",
+			incomingItems: []testFifoObject{
+				failObj3,
+			},
+			expectedBatches: [][]Delta{
+				{{testDeltaType, failObj3}},
+			},
+		},
+		{
+			name: "nth item is broken",
+			incomingItems: []testFifoObject{
+				sucessObj1, sucessObj2, failObj3,
+			},
+			expectedBatches: [][]Delta{
+				{{testDeltaType, sucessObj1}, {testDeltaType, sucessObj2}, {testDeltaType, failObj3}},
+			},
+		},
+		{
+			name: "multiple nth items are broken",
+			incomingItems: []testFifoObject{
+				sucessObj1, sucessObj2, failObj3, failObj4,
+			},
+			expectedBatches: [][]Delta{
+				{{testDeltaType, sucessObj1}, {testDeltaType, sucessObj2}, {testDeltaType, failObj3}},
+				{{testDeltaType, failObj4}},
+			},
+		},
+		{
+			name: "mixed 1st and nth items are broken",
+			incomingItems: []testFifoObject{
+				failObj3, sucessObj1, failObj4,
+			},
+			expectedBatches: [][]Delta{
+				{{testDeltaType, failObj3}},
+				{{testDeltaType, sucessObj1}, {testDeltaType, failObj4}},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			testBrokenItemKeyFunc := func(obj interface{}) (string, error) {
+				if obj == failObj3 || obj == failObj4 {
+					return "", errors.New("test key func error")
+				}
+				// otherwise success
+				return testFifoObjectKeyFunc(obj)
+			}
+			f := NewRealFIFO(
+				testBrokenItemKeyFunc,
+				literalListerGetter(func() []testFifoObject {
+					return nil
+				}),
+				nil)
+			f.batchSize = unlimitedBatchSize
+
+			for _, item := range tc.incomingItems {
+				f.items = append(f.items, Delta{testDeltaType, item})
+			}
+
+			const maxAttempts = 10
+			receivedItems := make([][]Delta, 0)
+
+			for i := 0; i < maxAttempts; i++ {
+				received := make(chan []Delta, 100)
+				go func() {
+					_ = f.PopBatch(func(obj []Delta, isInInitialList bool) error {
+						received <- obj
+						return nil
+					})
+				}()
+				timer := time.NewTimer(time.Millisecond * 50)
+				select {
+				case <-timer.C:
+					close(received)
+				case item := <-received:
+					receivedItems = append(receivedItems, item)
+					close(received)
+				}
+			}
+
+			runtime.Gosched()
+			f.Close()
+
+			idx := 0
+			assert.Len(t, tc.expectedBatches, len(receivedItems))
+			for _, batch := range tc.expectedBatches {
+				assert.Equal(t, receivedItems[idx], batch)
+				idx++
+			}
+		})
+	}
+}

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

@@ -19,8 +19,10 @@ package cache
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	utiltrace "k8s.io/utils/trace"
 )
 
 // ThreadSafeStore is an interface that allows concurrent indexed
@@ -58,6 +60,19 @@ type ThreadSafeStore interface {
 	Resync() error
 }
 
+// ThreadSafeStoreWithTransaction is a store that can batch execute multiple transactions.
+type ThreadSafeStoreWithTransaction interface {
+	ThreadSafeStore
+	// Transaction allows performing multiple writes in one call.
+	Transaction(fns ...ThreadSafeStoreTransaction)
+}
+
+// ThreadSafeStoreTransaction embeds a Transaction and includes the specific Key identifying the affected object.
+type ThreadSafeStoreTransaction struct {
+	Transaction
+	Key string
+}
+
 // storeIndex implements the indexing functionality for Store interface
 type storeIndex struct {
 	// indexers maps a name to an IndexFunc
@@ -229,13 +244,41 @@ type threadSafeMap struct {
 	index *storeIndex
 }
 
+func (c *threadSafeMap) Transaction(txns ...ThreadSafeStoreTransaction) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	trace := utiltrace.New("ThreadSafeMap Transaction Process",
+		utiltrace.Field{Key: "Size", Value: len(txns)},
+		utiltrace.Field{Key: "Reason", Value: "Slow batch process due to too many items"})
+	defer trace.LogIfLong(min(500*time.Millisecond*time.Duration(len(txns)), 5*time.Second))
+
+	for _, txn := range txns {
+		switch txn.Type {
+		case TransactionTypeAdd:
+			c.addLocked(txn.Key, txn.Object)
+		case TransactionTypeUpdate:
+			c.updateLocked(txn.Key, txn.Object)
+		case TransactionTypeDelete:
+			c.deleteLocked(txn.Key)
+		}
+	}
+}
+
 func (c *threadSafeMap) Add(key string, obj interface{}) {
 	c.Update(key, obj)
 }
 
+func (c *threadSafeMap) addLocked(key string, obj interface{}) {
+	c.updateLocked(key, obj)
+}
+
 func (c *threadSafeMap) Update(key string, obj interface{}) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
+	c.updateLocked(key, obj)
+}
+
+func (c *threadSafeMap) updateLocked(key string, obj interface{}) {
 	oldObject := c.items[key]
 	c.items[key] = obj
 	c.index.updateIndices(oldObject, obj, key)
@@ -244,6 +287,10 @@ func (c *threadSafeMap) Update(key string, obj interface{}) {
 func (c *threadSafeMap) Delete(key string) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
+	c.deleteLocked(key)
+}
+
+func (c *threadSafeMap) deleteLocked(key string) {
 	if obj, exists := c.items[key]; exists {
 		c.index.updateIndices(obj, nil, key)
 		delete(c.items, key)