try {
            watcher.shutdown();
            if (!watcher.awaitTermination(5, TimeUnit.SECONDS)) {
                throw new RuntimeException("Watcher did not terminate within 5 seconds");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new InterruptedIOException("Awaiting termination of watcher was interrupted");
        } finally {

	}
}

// serveWatch will serve a watch response according to the watcher and watchServer.
// Before watchServer.HandleHTTP, an error may occur like k8s.io/apiserver/pkg/endpoints/handlers/watch.go#serveWatch does.
func serveWatch(watcher watch.Interface, watchServer *handlers.WatchServer, preServeErr error) http.HandlerFunc {
	return func(w http.ResponseWriter, req *http.Request) {
		defer watcher.Stop()

		if preServeErr != nil {

)

// possible states of the cache watcher
const (
	// cacheWatcherWaitingForBookmark indicates the cacher
	// is waiting for a bookmark event with a specific RV set
	cacheWatcherWaitingForBookmark = iota

	// cacheWatcherBookmarkReceived indicates that the cacher
	// has received a bookmark event with required RV
	cacheWatcherBookmarkReceived

	// cacheWatcherBookmarkSent indicates that the cacher

	err := prober.watcher.Init(func(event fsnotify.Event) {
		if err := prober.handleWatchEvent(event); err != nil {
			klog.Errorf("Flexvolume prober watch: %s", err)
		}
	}, func(err error) {
		klog.Errorf("Received an error from watcher: %s", err)
	})
	if err != nil {
		return fmt.Errorf("error initializing watcher: %s", err)
	}

	if err := prober.addWatchRecursive(prober.pluginDir); err != nil {

	"istio.io/istio/pkg/test"
	"istio.io/istio/pkg/test/util/retry"
)

func TestNamespaceController(t *testing.T) {
	client := kube.NewFakeClient()
	t.Cleanup(client.Shutdown)
	watcher := keycertbundle.NewWatcher()
	caBundle := []byte("caBundle")
	watcher.SetAndNotify(nil, nil, caBundle)
	meshWatcher := mesh.NewTestWatcher(&meshconfig.MeshConfig{})
	stop := test.NewStop(t)
	discoveryNamespacesFilter := filter.NewDiscoveryNamespacesFilter(

			inf.Start(s.started)
		}
	}
}

type delayedFilter struct {
	Watcher  kubetypes.CrdWatcher
	Resource schema.GroupVersionResource
}

func (d *delayedFilter) HasSynced() bool {
	return d.Watcher.HasSynced()
}

func (d *delayedFilter) KnownOrCallback(f func(stop <-chan struct{})) bool {
	return d.Watcher.KnownOrCallback(d.Resource, f)
}

                new File(new File(buildDir, ".gradle"), "file-system.probe"));
            W watcher = createFileWatcher(fileEvents);
            WatchableHierarchies watchableHierarchies = new WatchableHierarchies(probeRegistry, immutableLocationsFilter);
            FileWatcherUpdater fileWatcherUpdater = createFileWatcherUpdater(watcher, probeRegistry, watchableHierarchies);
            return new DefaultFileWatcherRegistry(

		s.kubeClient, args.Namespace, configMapName, configMapKey, multiWatch, s.internalStop)
	s.environment.Watcher = multiWatcher
	s.environment.NetworksWatcher = multiWatcher
	log.Infof("initializing mesh networks from mesh config watcher")

	if multiWatch {
		kubemesh.AddUserMeshConfig(s.kubeClient, s.environment.Watcher, args.Namespace, configMapKey, features.SharedMeshConfig, s.internalStop)
	}
}

	if err != nil {
		return nil, err
	}

	watcher := &realWatcher{
		recorder:    recorder,
		oomStreamer: oomStreamer,
	}

	return watcher, nil
}

const (
	systemOOMEvent           = "SystemOOM"
	recordEventContainerName = "/"
)

// Start watches for system oom's and records an event for every system oom encountered.
func (ow *realWatcher) Start(ref *v1.ObjectReference) error {

				}
			}

			watcher := keycertbundle.NewWatcher()
			watcher.SetAndNotify(nil, nil, tc.pemData)
			whPatcher, err := NewWebhookCertPatcher(client, tc.revision, tc.webhookName, watcher)
			if err != nil {
				t.Fatal(err)
			}

			stop := test.NewStop(t)
			client.RunAndWait(stop)
			retry.UntilOrFail(t, whPatcher.webhooks.HasSynced)

			err = whPatcher.patchMutatingWebhookConfig(tc.configName)