storage/cacher/cache_watcher: processInterval sets RV from the snapshot

Kubernetes-commit: de735be512767dd2eced78530693d2e3ae997e6e

pkg/storage/cacher/cache_watcher.go

@@ -454,6 +454,13 @@ func (c *cacheWatcher) processInterval(ctx context.Context, cacheInterval *watch
 	const initProcessThreshold = 500 * time.Millisecond
 	startTime := time.Now()
 
+	// cacheInterval may be created from a version being more fresh than requested
+	// (e.g. for NotOlderThan semantic). In such a case, we need to prevent watch event
+	// with lower resourceVersion from being delivered to ensure watch contract.
+	if cacheInterval.resourceVersion > resourceVersion {
+		resourceVersion = cacheInterval.resourceVersion
+	}
+
 	initEventCount := 0
 	for {
 		event, err := cacheInterval.Next()