apiserver/storage/cacher/listwatcher: error when the WatchList FG is disabled

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

@@ -18,6 +18,7 @@ package cacher
 
 import (
 	"context"
+	"fmt"
 
 	"google.golang.org/grpc/metadata"
 
@@ -26,7 +27,9 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -88,5 +91,17 @@ func (lw *listerWatcher) Watch(options metav1.ListOptions) (watch.Interface, err
 	if lw.contextMetadata != nil {
 		ctx = metadata.NewOutgoingContext(ctx, lw.contextMetadata)
 	}
+
+	// we need the below check because the listWatcher bypasses the REST layer,
+	// so the options are not validated. Without this, we might end up in a situation
+	// where streaming is requested, but the FeatureGate is disabled,
+	// and the bookmark will not be sent
+	//
+	// in such a case, client-go is going to fall back to a standard LIST on any error
+	// returned for watch-list requests
+	if isListWatchRequest(opts) && !utilfeature.DefaultFeatureGate.Enabled(features.WatchList) {
+		return nil, fmt.Errorf("sendInitialEvents is forbidden for watch unless the WatchList feature gate is enabled")
+	}
+
 	return lw.storage.Watch(ctx, lw.resourcePrefix, opts)
 }

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

@@ -177,3 +177,29 @@ func TestCacherListerWatcherListWatch(t *testing.T) {
 	storagetesting.TestCheckResultsInStrictOrder(t, w, expectedWatchEvents)
 	storagetesting.TestCheckNoMoreResultsWithIgnoreFunc(t, w, nil)
 }
+
+func TestCacherListerWatcherWhenListWatchDisabled(t *testing.T) {
+	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, false)
+
+	prefix := "pods"
+	fn := func() runtime.Object { return &example.PodList{} }
+	server, store := newEtcdTestStorage(t, prefix)
+	defer server.Terminate(t)
+
+	lw := NewListerWatcher(store, prefix, fn, nil)
+	target := cache.ToListerWatcherWithContext(lw)
+	watchListOptions := metav1.ListOptions{
+		Watch:               true,
+		AllowWatchBookmarks: true,
+		SendInitialEvents:   ptr.To(true),
+	}
+	_, err := target.WatchWithContext(context.TODO(), watchListOptions)
+	if err == nil {
+		t.Fatalf("Expected error, but got none")
+	}
+
+	expectedErrMsg := "sendInitialEvents is forbidden for watch unless the WatchList feature gate is enabled"
+	if err.Error() != expectedErrMsg {
+		t.Fatalf("Expected error %q, but got %q", expectedErrMsg, err.Error())
+	}
+}