})
		ads.ExpectEmptyResponse()

		// Create pod, due to wildcard subscribe we should receive it
		createPod(s, "pod", "sa", "127.0.0.1", "not-node")
		expect(ads.ExpectResponse(), "Kubernetes//Pod/default/pod")

		// A new pod should push only that one
		createPod(s, "pod2", "sa", "127.0.0.2", "node")
		expect(ads.ExpectResponse(), "Kubernetes//Pod/default/pod2")

}

func TestHasInitialEventsEndBookmarkAnnotation(t *testing.T) {
	createPod := func(name string) *example.Pod {
		return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: name}}
	}
	createAnnotatedPod := func(name, value string) *example.Pod {
		p := createPod(name)
		p.Annotations = map[string]string{}
		p.Annotations[metav1.InitialEventsAnnotationKey] = value
		return p
	}

		client := &mockRecyclerClient{
			events: test.eventSequence,
			pod:    test.existingPod,
		}
		err := internalRecycleVolumeByWatchingPodUntilCompletion(test.createPod.Name, test.createPod, client)
		receivedError := ""
		if err != nil {
			receivedError = err.Error()
		}
		if receivedError != test.expectedError {
			t.Errorf("Test %q failed, expected error %q, got %q", test.name, test.expectedError, receivedError)

	})
	initTestEnv(t, c.client.Kube(), fx)
	createPod := func(ip, name string) {
		addPods(t, c, fx, generatePod(ip, name, "ns", "1", "", map[string]string{}, map[string]string{}))
	}

	createPod("128.0.0.1", "pod1")
	assert.Equal(t, c.pods.getPodKeys("128.0.0.1"), []types.NamespacedName{{Name: "pod1", Namespace: "ns"}})
	events.WaitOrdered("pod1/add", "pod1/update")
	createPod("128.0.0.1", "pod2")

	if err != nil {
		klog.V(4).Infof("cannot start watcher for pod %s/%s: %v", pod.Namespace, pod.Name, err)
		return err
	}

	// Start the pod
	_, err = recyclerClient.CreatePod(pod)
	if err != nil {
		if errors.IsAlreadyExists(err) {
			deleteErr := recyclerClient.DeletePod(pod.Name, pod.Namespace)
			if deleteErr != nil {

// with a clientset for writes and listers for reads; for tests we provide stubs.
type StatefulPodControlObjectManager interface {
	CreatePod(ctx context.Context, pod *v1.Pod) error
	GetPod(namespace, podName string) (*v1.Pod, error)
	UpdatePod(pod *v1.Pod) error
	DeletePod(pod *v1.Pod) error
	CreateClaim(claim *v1.PersistentVolumeClaim) error

	testContext := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
	destroyFunc, registry := NewTestGenericStoreRegistry(t)
	registry.EnableGarbageCollection = true
	defer destroyFunc()

	createPod := func(i int, finalizers []string) *example.Pod {
		return &example.Pod{
			ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("pod-%d", i), Finalizers: finalizers, Generation: initialGeneration},

		revisionInformer.Informer().GetIndexer(),
		newRequestTracker(0, nil, 0),
		newRequestTracker(0, nil, 0),
		newRequestTracker(0, nil, 0),
	}
}

func (om *fakeObjectManager) CreatePod(ctx context.Context, pod *v1.Pod) error {
	defer om.createPodTracker.trackParallelRequests()
	if err := om.createPodTracker.incWithOptionalError(); err != nil {
		return err
	}