Implement ResilientWatchCacheInitialization

Kubernetes-commit: a8ef6e9f0104a44023162bb8229fb677ec80beb1

tools/cache/reflector_test.go

@@ -697,13 +697,59 @@ func TestBackoffOnTooManyRequests(t *testing.T) {
 	}
 
 	stopCh := make(chan struct{})
-	r.ListAndWatch(stopCh)
+	if err := r.ListAndWatch(stopCh); err != nil {
+		t.Fatal(err)
+	}
 	close(stopCh)
 	if bm.calls != 2 {
 		t.Errorf("unexpected watch backoff calls: %d", bm.calls)
 	}
 }
 
+func TestNoRelistOnTooManyRequests(t *testing.T) {
+	err := apierrors.NewTooManyRequests("too many requests", 1)
+	clock := &clock.RealClock{}
+	bm := &fakeBackoff{clock: clock}
+	listCalls, watchCalls := 0, 0
+
+	lw := &testLW{
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			listCalls++
+			return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "1"}}, nil
+		},
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			watchCalls++
+			if watchCalls < 5 {
+				return nil, err
+			}
+			w := watch.NewFake()
+			w.Stop()
+			return w, nil
+		},
+	}
+
+	r := &Reflector{
+		name:              "test-reflector",
+		listerWatcher:     lw,
+		store:             NewFIFO(MetaNamespaceKeyFunc),
+		backoffManager:    bm,
+		clock:             clock,
+		watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
+	}
+
+	stopCh := make(chan struct{})
+	if err := r.ListAndWatch(stopCh); err != nil {
+		t.Fatal(err)
+	}
+	close(stopCh)
+	if listCalls != 1 {
+		t.Errorf("unexpected list calls: %d", listCalls)
+	}
+	if watchCalls != 5 {
+		t.Errorf("unexpected watch calls: %d", watchCalls)
+	}
+}
+
 func TestRetryInternalError(t *testing.T) {
 	testCases := []struct {
 		name                string