cacher.reflector = reflector

	go cacher.dispatchEvents()
	go progressRequester.Run(stopCh)

	cacher.stopWg.Add(1)
	go func() {
		defer cacher.stopWg.Done()
		defer cacher.terminateAllWatchers()
		wait.Until(
			func() {
				if !cacher.isStopped() {
					cacher.startCaching(stopCh)
				}
			}, time.Second, stopCh,
		)
	}()

	return cacher, nil
}

	if svc != nil {
		t.Fatalf("expected nil, got %v", svc)
	}
}

func BenchmarkServiceEntryHandler(b *testing.B) {
	_, sd := initServiceDiscoveryWithoutEvents(b)
	stopCh := make(chan struct{})
	go sd.Run(stopCh)
	defer close(stopCh)
	for i := 0; i < b.N; i++ {
		sd.serviceEntryHandler(config.Config{}, *httpDNS, model.EventAdd)
		sd.serviceEntryHandler(config.Config{}, *httpDNSRR, model.EventAdd)

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			ns := metav1.NamespaceDefault
			client, esController := newController(t, []string{"node-1"}, tc.batchPeriod)
			stopCh := make(chan struct{})
			defer close(stopCh)

			_, ctx := ktesting.NewTestContext(t)
			go esController.Run(ctx, 1)

			esController.serviceStore.Add(&v1.Service{
				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: ns},

	pred.AllowWatchBookmarks = true

	// run the collision test for 3 seconds to let ~2 buckets expire
	stopCh := make(chan struct{})
	var watchErr error
	time.AfterFunc(3*time.Second, func() { close(stopCh) })

	wg := &sync.WaitGroup{}

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stopCh:
				return
			default:
				ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)

func TestReverseProxyFlushIntervalHeaders(t *testing.T) {
	const expected = "hi"
	stopCh := make(chan struct{})
	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Add("MyHeader", expected)
		w.WriteHeader(200)
		w.(http.Flusher).Flush()
		<-stopCh
	}))
	defer backend.Close()
	defer close(stopCh)

	backendURL, err := url.Parse(backend.URL)
	if err != nil {

		o.logger = &logger
	}
}

// defaultFrameworkOptions are applied when no option corresponding to those fields exist.
func defaultFrameworkOptions(stopCh <-chan struct{}) frameworkOptions {
	return frameworkOptions{
		metricsRecorder: metrics.NewMetricsAsyncRecorder(1000, time.Second, stopCh),
		parallelizer:    parallelize.NewParallelizer(parallelize.DefaultParallelism),
	}
}

var _ framework.Framework = &frameworkImpl{}

			go func() {
				defer wk.Give()
				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
				success := true
				if err := r.ReplicateFromSource(ctx, api, core, oi, retry); err != nil {
					// object must be deleted concurrently, allow these failures but do not count them
					if isErrVersionNotFound(err) || isErrObjectNotFound(err) {
						return
					}
					stopFn(oi, err)
					batchLogIf(ctx, err)

// gcMode indicates how concurrent a GC cycle should be.
type gcMode int

const (
	gcBackgroundMode gcMode = iota // concurrent GC and sweep
	gcForceMode                    // stop-the-world GC now, concurrent sweep
	gcForceBlockMode               // stop-the-world GC now and STW sweep (forced by user)
)

// A gcTrigger is a predicate for starting a GC cycle. Specifically,
// it is an exit condition for the _GCoff phase.

        def service = Mock(TestStopService)

        given:
        registry.add(TestStopService, service)

        when:
        registry.close()

        then:
        1 * service.stop()
    }

    def closeIgnoresServiceWithNoCloseOrStopMethod() {
        registry.add(String, "service")
        registry.getAll(String)

        when:
        registry.close()

			require.Equal(t, int64(1), resourceAllocatable.Value(), "Devices of plugin previously registered should be removed.")
			p2.Stop()
			p3.Stop()
			cleanup(t, m, p1)
		}
	}
}

// Tests that the device plugin manager correctly handles registration and re-registration by