comps := make([]string, 0, len(p.components))
	wait := make([]string, 0, len(p.components))
	for c, l := range p.components {
		comps = append(comps, name.UserFacingComponentName(name.ComponentName(c)))
		wait = append(wait, l.waitingResources()...)
	}
	sort.Strings(comps)
	sort.Strings(wait)
	msg := fmt.Sprintf(`Processing resources for %s.`, strings.Join(comps, ", "))
	if len(wait) > 0 {

		go func() {
			defer waitGroup.Done()
			wait.Until(c.runWorker, time.Second, ctx.Done())
		}()
	}

	klog.Infof("Started %v workers for %v", c.options.Workers, c.options.Name)

	// Wait for context cancel.
	<-ctx.Done()

	// Forcefully shutdown workqueue. Drop any enqueued items.
	c.queue.ShutDown()

	// Workqueue shutdown signals for workers to stop. Wait for all workers to
	// clean up
	waitGroup.Wait()

	}
}

func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.ConditionFunc) error {
	backoff := wait.Backoff{
		Duration: initialDuration,
		Factor:   3,
		Jitter:   0,
		Steps:    6,
	}
	return wait.ExponentialBackoff(backoff, fn)
}

func TestPluginRegistration(t *testing.T) {
	defer cleanup(t)

				w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
				return w.Run(func() error {
					wait := deleteCleanupSleeper.Timer(ctx)
					removeAll(pathJoin(drivePath, minioMetaTmpDeletedBucket, ddir))
					wait()
					return nil
				})
			})
		}(disk)
	}
	wg.Wait()
}

// nsScanner will start scanning buckets and send updated totals as they are traversed.

	if client.Discovery().RESTClient() == nil {
		fmt.Printf("[upgrade/health] Would create the Job with the prefix %q in namespace %q and wait until it completes\n", prefix, ns)
		return nil
	}

	// Check if there is at least one Node where a Job's Pod can schedule. If not, skip this preflight check.
	err = wait.PollUntilContextTimeout(ctx, time.Second*1, timeout, true, func(_ context.Context) (bool, error) {

	epsHandler := NewEndpointSliceHandlerMock()
	epsConfig.RegisterEventHandler(epsHandler)

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

	err := wait.PollImmediate(time.Millisecond*10, wait.ForeverTestTimeout, func() (bool, error) {
		svcHandler.lock.Lock()
		defer svcHandler.lock.Unlock()
		if reflect.DeepEqual(svcHandler.state, expectedSvcState) {
			return true, nil
		}

limitations under the License.
*/

package cacher

import (
	"context"
	"sync"
	"time"

	"google.golang.org/grpc/metadata"

	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"

	"k8s.io/klog/v2"
	"k8s.io/utils/clock"
)

const (
	// progressRequestPeriod determines period of requesting progress
	// from etcd when there is a request waiting for watch cache to be fresh.

			default:
				cacher.Lock()
				cacher.bookmarkWatchers.popExpiredWatchersThreadUnsafe()
				cacher.Unlock()
			}
		}
	}()

	// wait for adding/removing watchers to end
	wg.Wait()

	if watchErr != nil {
		t.Fatal(watchErr)
	}

	// wait out the expiration period and pop expired watchers
	time.Sleep(2 * time.Second)
	cacher.Lock()
	defer cacher.Unlock()

	backingStorage := &dummyStorage{}
	cacher, _, err := newTestCacher(backingStorage)
	if err != nil {
		t.Fatalf("Couldn't create cacher: %v", err)
	}
	defer cacher.Stop()

	// Wait until cacher is initialized.
	if err := cacher.ready.wait(context.Background()); err != nil {
		t.Fatalf("unexpected error waiting for the cache to be ready")
	}

	if err := adscConn.Run(); err != nil {
		f.t.Fatalf("ADSC: failed running: %v", err)
	}
	if len(wait) > 0 {
		_, err := adscConn.Wait(10*time.Second, wait...)
		if err != nil {
			f.t.Fatalf("Error getting initial for %v config: %v", wait, err)
		}
	}
	return adscConn
}

func (f *FakeDiscoveryServer) Endpoints(p *model.Proxy) []*endpoint.ClusterLoadAssignment {
	loadAssignments := make([]*endpoint.ClusterLoadAssignment, 0)