add watch bookmark

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/decorated_watcher.go

@@ -55,7 +55,7 @@ func (d *decoratedWatcher) run(ctx context.Context) {
 				return
 			}
 			switch recv.Type {
-			case watch.Added, watch.Modified, watch.Deleted:
+			case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
 				err := d.decorator(recv.Object)
 				if err != nil {
 					send = makeStatusErrorEvent(err)

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -1075,6 +1075,7 @@ func (e *Store) Watch(ctx context.Context, options *metainternalversion.ListOpti
 	resourceVersion := ""
 	if options != nil {
 		resourceVersion = options.ResourceVersion
+		predicate.AllowWatchBookmarks = options.AllowWatchBookmarks
 	}
 	return e.WatchPredicate(ctx, predicate, resourceVersion)
 }

staging/src/k8s.io/apiserver/pkg/storage/cacher/BUILD

@@ -59,8 +59,11 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/example:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/example/v1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/features:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
     ],
 )

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go

@@ -32,13 +32,17 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apimachinery/pkg/util/diff"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/apis/example"
 	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilfeaturetesting "k8s.io/apiserver/pkg/util/feature/testing"
 )
 
 // verifies the cacheWatcher.process goroutine is properly cleaned up even if
@@ -63,7 +67,7 @@ func TestCacheWatcherCleanupNotBlockedByResult(t *testing.T) {
 	}
 	// set the size of the buffer of w.result to 0, so that the writes to
 	// w.result is blocked.
-	w = newCacheWatcher(0, filter, forget, testVersioner{})
+	w = newCacheWatcher(0, filter, forget, testVersioner{}, time.Now(), false)
 	go w.process(context.Background(), initEvents, 0)
 	w.Stop()
 	if err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
@@ -183,8 +187,9 @@ TestCase:
 			testCase.events[j].ResourceVersion = uint64(j) + 1
 		}
 
-		w := newCacheWatcher(0, filter, forget, testVersioner{})
+		w := newCacheWatcher(0, filter, forget, testVersioner{}, time.Now(), false)
 		go w.process(context.Background(), testCase.events, 0)
+
 		ch := w.ResultChan()
 		for j, event := range testCase.expected {
 			e := <-ch
@@ -461,7 +466,7 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
 	// timeout to zero and run the Stop goroutine concurrently.
 	// May sure that the watch will not be blocked on Stop.
 	for i := 0; i < maxRetriesToProduceTheRaceCondition; i++ {
-		w = newCacheWatcher(0, filter, forget, testVersioner{})
+		w = newCacheWatcher(0, filter, forget, testVersioner{}, time.Now(), false)
 		go w.Stop()
 		select {
 		case <-done:
@@ -470,11 +475,12 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
 		}
 	}
 
+	deadline := time.Now().Add(time.Hour)
 	// After that, verifies the cacheWatcher.process goroutine works correctly.
 	for i := 0; i < maxRetriesToProduceTheRaceCondition; i++ {
-		w = newCacheWatcher(2, filter, emptyFunc, testVersioner{})
+		w = newCacheWatcher(2, filter, emptyFunc, testVersioner{}, deadline, false)
 		w.input <- &watchCacheEvent{Object: &v1.Pod{}, ResourceVersion: uint64(i + 1)}
-		ctx, _ := context.WithTimeout(context.Background(), time.Hour)
+		ctx, _ := context.WithDeadline(context.Background(), deadline)
 		go w.process(ctx, nil, 0)
 		select {
 		case <-w.ResultChan():
@@ -484,3 +490,210 @@ func TestCacheWatcherStoppedInAnotherGoroutine(t *testing.T) {
 		w.Stop()
 	}
 }
+
+func TestTimeBucketWatchersBasic(t *testing.T) {
+	filter := func(_ string, _ labels.Set, _ fields.Set) bool {
+		return true
+	}
+	forget := func() {}
+
+	newWatcher := func(deadline time.Time) *cacheWatcher {
+		return newCacheWatcher(0, filter, forget, testVersioner{}, deadline, true)
+	}
+
+	clock := clock.NewFakeClock(time.Now())
+	watchers := newTimeBucketWatchers(clock)
+	now := clock.Now()
+	watchers.addWatcher(newWatcher(now.Add(10 * time.Second)))
+	watchers.addWatcher(newWatcher(now.Add(20 * time.Second)))
+	watchers.addWatcher(newWatcher(now.Add(20 * time.Second)))
+
+	if len(watchers.watchersBuckets) != 2 {
+		t.Errorf("unexpected bucket size: %#v", watchers.watchersBuckets)
+	}
+	watchers0 := watchers.popExpiredWatchers()
+	if len(watchers0) != 0 {
+		t.Errorf("unexpected bucket size: %#v", watchers0)
+	}
+
+	clock.Step(10 * time.Second)
+	watchers1 := watchers.popExpiredWatchers()
+	if len(watchers1) != 1 || len(watchers1[0]) != 1 {
+		t.Errorf("unexpected bucket size: %v", watchers1)
+	}
+	watchers1 = watchers.popExpiredWatchers()
+	if len(watchers1) != 0 {
+		t.Errorf("unexpected bucket size: %#v", watchers1)
+	}
+
+	clock.Step(12 * time.Second)
+	watchers2 := watchers.popExpiredWatchers()
+	if len(watchers2) != 1 || len(watchers2[0]) != 2 {
+		t.Errorf("unexpected bucket size: %#v", watchers2)
+	}
+}
+
+func testCacherSendBookmarkEvents(t *testing.T, watchCacheEnabled, allowWatchBookmarks, expectedBookmarks bool) {
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, watchCacheEnabled)()
+	backingStorage := &dummyStorage{}
+	cacher, _ := newTestCacher(backingStorage, 1000)
+	defer cacher.Stop()
+
+	// Wait until cacher is initialized.
+	cacher.ready.wait()
+	pred := storage.Everything
+	pred.AllowWatchBookmarks = allowWatchBookmarks
+
+	ctx, _ := context.WithTimeout(context.Background(), 3*time.Second)
+	w, err := cacher.Watch(ctx, "pods/ns", "0", pred)
+	if err != nil {
+		t.Fatalf("Failed to create watch: %v", err)
+	}
+
+	resourceVersion := uint64(1000)
+	go func() {
+		deadline := time.Now().Add(time.Second)
+		for i := 0; time.Now().Before(deadline); i++ {
+			err = cacher.watchCache.Add(&examplev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:            fmt.Sprintf("pod-%d", i),
+					Namespace:       "ns",
+					ResourceVersion: fmt.Sprintf("%v", resourceVersion+uint64(i)),
+				}})
+			if err != nil {
+				t.Fatalf("failed to add a pod: %v", err)
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+	}()
+
+	timeoutCh := time.After(2 * time.Second)
+	lastObservedRV := uint64(0)
+	for {
+		select {
+		case event, ok := <-w.ResultChan():
+			if !ok {
+				t.Fatal("Unexpected closed")
+			}
+			rv, err := cacher.versioner.ObjectResourceVersion(event.Object)
+			if err != nil {
+				t.Errorf("failed to parse resource version from %#v", event.Object)
+			}
+			if event.Type == watch.Bookmark {
+				if !expectedBookmarks {
+					t.Fatalf("Unexpected bookmark events received")
+				}
+
+				if rv < lastObservedRV {
+					t.Errorf("Unexpected bookmark event resource version %v (last %v)", rv, lastObservedRV)
+				}
+				return
+			}
+			lastObservedRV = rv
+		case <-timeoutCh:
+			if expectedBookmarks {
+				t.Fatal("Unexpected timeout to receive a bookmark event")
+			}
+			return
+		}
+	}
+}
+
+func TestCacherSendBookmarkEvents(t *testing.T) {
+	testCases := []struct {
+		watchCacheEnabled   bool
+		allowWatchBookmarks bool
+		expectedBookmarks   bool
+	}{
+		{
+			watchCacheEnabled:   true,
+			allowWatchBookmarks: true,
+			expectedBookmarks:   true,
+		},
+		{
+			watchCacheEnabled:   true,
+			allowWatchBookmarks: false,
+			expectedBookmarks:   false,
+		},
+		{
+			watchCacheEnabled:   false,
+			allowWatchBookmarks: true,
+			expectedBookmarks:   false,
+		},
+		{
+			watchCacheEnabled:   false,
+			allowWatchBookmarks: false,
+			expectedBookmarks:   false,
+		},
+	}
+
+	for _, tc := range testCases {
+		testCacherSendBookmarkEvents(t, tc.watchCacheEnabled, tc.allowWatchBookmarks, tc.expectedBookmarks)
+	}
+}
+
+func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) {
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, true)()
+	backingStorage := &dummyStorage{}
+	cacher, _ := newTestCacher(backingStorage, 1000)
+	defer cacher.Stop()
+
+	// Wait until cacher is initialized.
+	cacher.ready.wait()
+
+	// Ensure there is some budget for slowing down processing.
+	cacher.dispatchTimeoutBudget.returnUnused(100 * time.Millisecond)
+
+	resourceVersion := uint64(1000)
+	err := cacher.watchCache.Add(&examplev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            fmt.Sprintf("pod-0"),
+			Namespace:       "ns",
+			ResourceVersion: fmt.Sprintf("%v", resourceVersion),
+		}})
+	if err != nil {
+		t.Fatalf("failed to add a pod: %v", err)
+	}
+
+	for i := 0; i < 1000; i++ {
+		pred := storage.Everything
+		pred.AllowWatchBookmarks = true
+		ctx, _ := context.WithTimeout(context.Background(), time.Second)
+		w, err := cacher.Watch(ctx, "pods/ns", "999", pred)
+		if err != nil {
+			t.Fatalf("Failed to create watch: %v", err)
+		}
+		bookmark := &watchCacheEvent{
+			Type:            watch.Bookmark,
+			ResourceVersion: uint64(i),
+			Object:          cacher.newFunc(),
+		}
+		err = cacher.versioner.UpdateObject(bookmark.Object, bookmark.ResourceVersion)
+		if err != nil {
+			t.Fatalf("failure to update version of object (%d) %#v", bookmark.ResourceVersion, bookmark.Object)
+		}
+
+		go func() {
+			cacher.dispatchEvent(bookmark)
+		}()
+
+		go func() {
+			w.Stop()
+		}()
+
+		done := make(chan struct{})
+		go func() {
+			for range w.ResultChan() {
+			}
+			close(done)
+		}()
+
+		select {
+		case <-done:
+			break
+		case <-time.After(time.Second):
+			t.Fatal("receive result timeout")
+		}
+		w.Stop()
+	}
+}

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog"
 	utiltrace "k8s.io/utils/trace"
 )
 
@@ -386,6 +387,7 @@ func (w *watchCache) Replace(objs []interface{}, resourceVersion string) error {
 		w.onReplace()
 	}
 	w.cond.Broadcast()
+	klog.V(3).Infof("Replace watchCache (rev: %v) ", resourceVersion)
 	return nil
 }
 

staging/src/k8s.io/apiserver/pkg/storage/selection_predicate.go

@@ -71,12 +71,13 @@ func (f AttrFunc) WithFieldMutation(fieldMutator FieldMutationFunc) AttrFunc {
 
 // SelectionPredicate is used to represent the way to select objects from api storage.
 type SelectionPredicate struct {
-	Label       labels.Selector
-	Field       fields.Selector
-	GetAttrs    AttrFunc
-	IndexFields []string
-	Limit       int64
-	Continue    string
+	Label               labels.Selector
+	Field               fields.Selector
+	GetAttrs            AttrFunc
+	IndexFields         []string
+	Limit               int64
+	Continue            string
+	AllowWatchBookmarks bool
 }
 
 // Matches returns true if the given object's labels and fields (as

staging/src/k8s.io/apiserver/pkg/storage/tests/BUILD

@@ -26,6 +26,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/example:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/example/v1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/features:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/cacher:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd:go_default_library",
@@ -33,6 +34,8 @@ go_test(
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd/testing:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd3:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/value:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
     ],
 )
 

staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go

@@ -40,6 +40,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/apis/example"
 	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage"
 	cacherstorage "k8s.io/apiserver/pkg/storage/cacher"
 	etcdstorage "k8s.io/apiserver/pkg/storage/etcd"
@@ -47,6 +48,8 @@ import (
 	etcdtesting "k8s.io/apiserver/pkg/storage/etcd/testing"
 	"k8s.io/apiserver/pkg/storage/etcd3"
 	"k8s.io/apiserver/pkg/storage/value"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilfeaturetesting "k8s.io/apiserver/pkg/util/feature/testing"
 )
 
 var (
@@ -121,6 +124,12 @@ func makeTestPod(name string) *example.Pod {
 	}
 }
 
+func createPod(s storage.Interface, obj *example.Pod) error {
+	key := "pods/" + obj.Namespace + "/" + obj.Name
+	out := &example.Pod{}
+	return s.Create(context.TODO(), key, obj, out, 0)
+}
+
 func updatePod(t *testing.T, s storage.Interface, obj, old *example.Pod) *example.Pod {
 	updateFn := func(input runtime.Object, res storage.ResponseMeta) (runtime.Object, *uint64, error) {
 		return obj.DeepCopyObject(), nil, nil
@@ -773,4 +782,129 @@ func TestCacherListerWatcherPagination(t *testing.T) {
 	if limit2.Items[0].Name != podFoo.Name {
 		t.Errorf("Expected list2.Items[0] to be %s but got %s", podFoo.Name, limit2.Items[0].Name)
 	}
+
+}
+
+func TestWatchDispatchBookmarkEvents(t *testing.T) {
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, true)()
+
+	server, etcdStorage := newEtcdTestStorage(t, etcdtest.PathPrefix())
+	defer server.Terminate(t)
+	cacher, v := newTestCacher(etcdStorage, 10)
+	defer cacher.Stop()
+
+	fooCreated := updatePod(t, etcdStorage, makeTestPod("foo"), nil)
+	rv, err := v.ParseResourceVersion(fooCreated.ResourceVersion)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	startVersion := strconv.Itoa(int(rv))
+
+	tests := []struct {
+		timeout            time.Duration
+		expected           bool
+		allowWatchBookmark bool
+	}{
+		{ // test old client won't get Bookmark event
+			timeout:            2 * time.Second,
+			expected:           false,
+			allowWatchBookmark: false,
+		},
+		{
+			timeout:            2 * time.Second,
+			expected:           true,
+			allowWatchBookmark: true,
+		},
+	}
+
+	for i, c := range tests {
+		pred := storage.Everything
+		pred.AllowWatchBookmarks = c.allowWatchBookmark
+		ctx, _ := context.WithTimeout(context.Background(), c.timeout)
+		watcher, err := cacher.Watch(ctx, "pods/ns/foo", startVersion, pred)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		// Create events of other pods
+		updatePod(t, etcdStorage, makeTestPod(fmt.Sprintf("foo-whatever-%d", i)), nil)
+
+		// Now wait for Bookmark event
+		select {
+		case event, ok := <-watcher.ResultChan():
+			if !ok && c.expected {
+				t.Errorf("Unexpected object watched (no objects)")
+			}
+			if c.expected && event.Type != watch.Bookmark {
+				t.Errorf("Unexpected object watched %#v", event)
+			}
+		case <-time.After(time.Second * 3):
+			if c.expected {
+				t.Errorf("Unexpected object watched (timeout)")
+			}
+		}
+		watcher.Stop()
+	}
+}
+
+func TestWatchBookmarksWithCorrectResourceVersion(t *testing.T) {
+	defer utilfeaturetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchBookmark, true)()
+
+	server, etcdStorage := newEtcdTestStorage(t, etcdtest.PathPrefix())
+	defer server.Terminate(t)
+	cacher, v := newTestCacher(etcdStorage, 10)
+	defer cacher.Stop()
+
+	pred := storage.Everything
+	pred.AllowWatchBookmarks = true
+	ctx, _ := context.WithTimeout(context.Background(), 2*time.Second)
+	watcher, err := cacher.WatchList(ctx, "pods/ns", "0", pred)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer watcher.Stop()
+
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		for i := 0; i < 100; i++ {
+			select {
+			case <-done:
+				return
+			default:
+				pod := fmt.Sprintf("foo-%d", i)
+				err := createPod(etcdStorage, makeTestPod(pod))
+				if err != nil {
+					t.Fatalf("failed to create pod %v", pod)
+				}
+				time.Sleep(time.Second / 100)
+			}
+		}
+	}()
+
+	bookmarkReceived := false
+	lastObservedResourceVersion := uint64(0)
+	for event := range watcher.ResultChan() {
+		rv, err := v.ObjectResourceVersion(event.Object)
+		if err != nil {
+			t.Fatalf("failed to parse resourceVersion from %#v", event)
+		}
+		if event.Type == watch.Bookmark {
+			bookmarkReceived = true
+			// bookmark event has a RV greater than or equal to the before one
+			if rv < lastObservedResourceVersion {
+				t.Fatalf("Unexpected bookmark resourceVersion %v less than observed %v)", rv, lastObservedResourceVersion)
+			}
+		} else {
+			// non-bookmark event has a RV greater than anything before
+			if rv <= lastObservedResourceVersion {
+				t.Fatalf("Unexpected event resourceVersion %v less than or equal to bookmark %v)", rv, lastObservedResourceVersion)
+			}
+		}
+		lastObservedResourceVersion = rv
+	}
+	// Make sure we have received a bookmark event
+	if !bookmarkReceived {
+		t.Fatalf("Unpexected error, we did not received a bookmark event")
+	}
 }

staging/src/k8s.io/client-go/rest/watch/decoder.go

@@ -54,7 +54,7 @@ func (d *Decoder) Decode() (watch.EventType, runtime.Object, error) {
 		return "", nil, fmt.Errorf("unable to decode to metav1.Event")
 	}
 	switch got.Type {
-	case string(watch.Added), string(watch.Modified), string(watch.Deleted), string(watch.Error):
+	case string(watch.Added), string(watch.Modified), string(watch.Deleted), string(watch.Error), string(watch.Bookmark):
 	default:
 		return "", nil, fmt.Errorf("got invalid watch event type: %v", got.Type)
 	}

staging/src/k8s.io/client-go/rest/watch/decoder_test.go

@@ -42,7 +42,7 @@ func getDecoder() runtime.Decoder {
 }
 
 func TestDecoder(t *testing.T) {
-	table := []watch.EventType{watch.Added, watch.Deleted, watch.Modified, watch.Error}
+	table := []watch.EventType{watch.Added, watch.Deleted, watch.Modified, watch.Error, watch.Bookmark}
 
 	for _, eventType := range table {
 		out, in := io.Pipe()

staging/src/k8s.io/client-go/rest/watch/encoder_test.go

@@ -56,6 +56,10 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {
 			watch.Deleted,
 			&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}},
 		},
+		{
+			watch.Bookmark,
+			&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}},
+		},
 	}
 	for i, testCase := range testCases {
 		buf := &bytes.Buffer{}

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

@@ -271,6 +271,11 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 			// We want to avoid situations of hanging watchers. Stop any wachers that do not
 			// receive any events within the timeout window.
 			TimeoutSeconds: &timeoutSeconds,
+			// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
+			// Reflector doesn't assume bookmarks are returned at all (if the server do not support
+			// watch bookmarks, it will ignore this field).
+			// Disabled in Alpha release of watch bookmarks feature.
+			AllowWatchBookmarks: false,
 		}
 
 		w, err := r.listerWatcher.Watch(options)
@@ -368,6 +373,8 @@ loop:
 				if err != nil {
 					utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
 				}
+			case watch.Bookmark:
+				// A `Bookmark` means watch has synced here, just update the resourceVersion
 			default:
 				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
 			}

staging/src/k8s.io/client-go/tools/watch/retrywatcher.go

@@ -153,7 +153,7 @@ func (rw *RetryWatcher) doReceive() (bool, time.Duration) {
 
 			// We need to inspect the event and get ResourceVersion out of it
 			switch event.Type {
-			case watch.Added, watch.Modified, watch.Deleted:
+			case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
 				metaObject, ok := event.Object.(resourceVersionGetter)
 				if !ok {
 					_ = rw.send(watch.Event{