stopCh := make(chan struct{})
	defer close(stopCh)

	handler := NewServiceHandlerMock()

	sharedInformers := informers.NewSharedInformerFactory(client, time.Minute)

	serviceConfig := NewServiceConfig(ctx, sharedInformers.Core().V1().Services(), time.Minute)
	serviceConfig.RegisterEventHandler(handler)
	go sharedInformers.Start(stopCh)
	go serviceConfig.Run(stopCh)

	stopCh := make(chan struct{})
	defer close(stopCh)

	sharedInformers := informers.NewSharedInformerFactory(client, time.Minute)

	config := NewServiceConfig(ctx, sharedInformers.Core().V1().Services(), time.Minute)
	handler := NewServiceHandlerMock()
	config.RegisterEventHandler(handler)
	go sharedInformers.Start(stopCh)
	go config.Run(stopCh)

	service := &v1.Service{

			manager := newTestVolumeManager(t, tmpDir, podManager, kubeClient, node)

			stopCh := runVolumeManager(manager)
			defer close(stopCh)

			podManager.SetPods([]*v1.Pod{pod})

			// Fake node status update
			go simulateVolumeInUseUpdate(
				v1.UniqueVolumeName(node.Status.VolumesAttached[0].Name),
				stopCh,
				manager)

			err = manager.WaitForAttachAndMount(context.Background(), pod)

}

// monitor runs a Controller with a local stop channel.
type monitor struct {
	controller cache.Controller

	// stopCh stops Controller. If stopCh is nil, the monitor is considered to be
	// not yet started.
	stopCh chan struct{}
}

// Run is intended to be called in a goroutine. Multiple calls of this is an
// error.
func (m *monitor) Run() {
	m.controller.Run(m.stopCh)
}

	return createHandlerWithConfig(kubeClient, informerFactory, nil, stopCh)
}

func createHandlerWithConfig(kubeClient kubernetes.Interface, informerFactory informers.SharedInformerFactory, config *resourcequotaapi.Configuration, stopCh chan struct{}) (*resourcequota.QuotaAdmission, error) {
	if config == nil {

	fakePodControl := controller.FakePodControl{}
	stopCh := make(chan struct{})
	defer close(stopCh)
	manager, informers := testNewReplicaSetControllerFromClient(t, client, stopCh, BurstReplicas)

	// 2 running pods, a controller with 2 replicas, sync is a no-op
	labelMap := map[string]string{"foo": "bar"}

	// lock is to ensure the access and modify of lastAccessTime, stopped, and immutable are thread safety,
	// and protecting from closing stopCh multiple times.
	lock           sync.Mutex
	lastAccessTime time.Time
	stopped        bool
	immutable      bool
	stopCh         chan struct{}
}

func (i *objectCacheItem) stop() bool {
	i.lock.Lock()
	defer i.lock.Unlock()
	return i.stopThreadUnsafe()
}

		wg.Wait()
		c <- struct{}{}
	}()
	select {
	case <-c:
		return true
	case <-time.After(timeout):
		return false
	}
}

func TestSendPushesManyPushes(t *testing.T) {
	stopCh := make(chan struct{})
	defer close(stopCh)

	semaphore := make(chan struct{}, 2)
	queue := NewPushQueue()
	defer queue.ShutDown()

	proxies := createProxies(5)

	pushes := make(map[string]int)
	pushesMu := &sync.Mutex{}

}

// Run starts the EstablishingController.
func (ec *EstablishingController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ec.queue.ShutDown()

	klog.Info("Starting EstablishingController")
	defer klog.Info("Shutting down EstablishingController")

	if !cache.WaitForCacheSync(stopCh, ec.crdSynced) {
		return
	}

	// only start one worker thread since its a slow moving API

	cond    *sync.Cond
	waiting int
	stopped bool
}

func (pr *conditionalProgressRequester) Run(stopCh <-chan struct{}) {
	ctx := wait.ContextForChannel(stopCh)
	if pr.contextMetadata != nil {
		ctx = metadata.NewOutgoingContext(ctx, pr.contextMetadata)
	}
	go func() {
		defer utilruntime.HandleCrash()
		<-stopCh
		pr.mux.Lock()
		defer pr.mux.Unlock()
		pr.stopped = true
		pr.cond.Signal()
	}()