kubernetes · deads2k · Jun 26, 2025
diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go
@@ -18,9 +18,12 @@ package cache
 
 import (
 	"fmt"
+	"k8s.io/apimachinery/pkg/api/meta"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	utiltrace "k8s.io/utils/trace"
+	"sort"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -326,8 +329,10 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error
 	}
 
 	// now that we have the deletes we need for items, we can add the newItems to the items queue
-	for _, obj := range newItems {
-		retErr := f.addToItems_locked(Replaced, false, obj)
+	sortedNewItems := createSortableItems(newItems)
+	sort.Sort(byResourceVersion(sortedNewItems))
+	for _, currSortableItem := range sortedNewItems {
+		retErr := f.addToItems_locked(Replaced, false, currSortableItem.item)
 		if retErr != nil {
 			return fmt.Errorf("couldn't enqueue object: %w", retErr)
 		}
@@ -405,3 +410,54 @@ func NewRealFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter, transformer Tran
 	f.cond.L = &f.lock
 	return f
 }
+
+func createSortableItems(items []interface{}) []sortableItem {
+	sortableItems := make([]sortableItem, len(items))
+	for i := range items {
+		sortableItems[i] = newSortableItem(items[i])
+	}
+
+	return sortableItems
+}
+
+type sortableItem struct {
+	item            interface{}
+	resourceVersion int
+}
+
+func newSortableItem(item interface{}) sortableItem {
+	return sortableItem{
+		item:            item,
+		resourceVersion: comparableResourceVersion(item),
+	}
+}
+
+type byResourceVersion []sortableItem
+
+// Len is part of sort.Interface.
+func (s byResourceVersion) Len() int {
+	return len(s)
+}
+
+// Swap is part of sort.Interface.
+func (s byResourceVersion) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+// Less is part of sort.Interface. It is implemented by calling the "by" closure in the sorter.
+func (s byResourceVersion) Less(i, j int) bool {
+	return s[i].resourceVersion < s[j].resourceVersion
+}
+
+func comparableResourceVersion(obj interface{}) int {
+	meta, err := meta.Accessor(obj)
+	if err != nil {
+		return 0 // resourceversions are never zero
+	}
+	rvString := meta.GetResourceVersion()
+	resourceVersion, err := strconv.Atoi(rvString)
+	if err != nil {
+		return 0 // resourceversions are never zero
+	}
+	return resourceVersion
+}
diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go
@@ -974,3 +974,74 @@ func TestRealFIFO_PopShouldUnblockWhenClosed(t *testing.T) {
 		}
 	}
 }
+
+func TestRealFIFO_ReplaceOrdersWhenPossible(t *testing.T) {
+	tests := []struct {
+		actions        []func(f *RealFIFO)
+		expectedSynced bool
+	}{
+		{
+			actions:        []func(f *RealFIFO){},
+			expectedSynced: false,
+		},
+		{
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Add(mkFifoObj("a", 1)) },
+			},
+			expectedSynced: true,
+		},
+		{
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Replace([]interface{}{}, "0") },
+			},
+			expectedSynced: true,
+		},
+		{
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
+			},
+			expectedSynced: false,
+		},
+		{
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
+				func(f *RealFIFO) { Pop(f) },
+			},
+			expectedSynced: false,
+		},
+		{
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("b", 2)}, "0") },
+				func(f *RealFIFO) { Pop(f) },
+				func(f *RealFIFO) { Pop(f) },
+			},
+			expectedSynced: true,
+		},
+		{
+			// This test case won't happen in practice since a Reflector, the only producer for delta_fifo today, always passes a complete snapshot consistent in time;
+			// there cannot be duplicate keys in the list or apiserver is broken.
+			actions: []func(f *RealFIFO){
+				func(f *RealFIFO) { f.Replace([]interface{}{mkFifoObj("a", 1), mkFifoObj("a", 2)}, "0") },
+				func(f *RealFIFO) { Pop(f) },
+				// ATTENTION: difference with delta_fifo_test, every event is delivered, so a is listed twice and must be popped twice to remove both
+				func(f *RealFIFO) { Pop(f) },
+			},
+			expectedSynced: true,
+		},
+	}
+
+	for i, test := range tests {
+		f := NewRealFIFO(
+			testFifoObjectKeyFunc,
+			emptyKnownObjects(),
+			nil,
+		)
+
+		for _, action := range test.actions {
+			action(f)
+		}
+		if e, a := test.expectedSynced, f.HasSynced(); a != e {
+			t.Errorf("test case %v failed, expected: %v , got %v", i, e, a)
+		}
+	}
+}