return fmt.Errorf("error syncing pod %q: %v", format.Pod(pod), err)
		}
		if retry >= runOnceMaxRetries {
			return fmt.Errorf("timeout error: pod %q containers not running after %d retries", format.Pod(pod), runOnceMaxRetries)
		}
		// TODO(proppy): health checking would be better than waiting + checking the state at the next iteration.
		klog.InfoS("Pod's containers synced, waiting", "pod", klog.KObj(pod), "duration", delay)

	qm.monitorLock.RLock()
	defer qm.monitorLock.RUnlock()

	if len(qm.monitors) == 0 {
		logger.V(4).Info("quota monitor not synced: no monitors")
		return false
	}

	for resource, monitor := range qm.monitors {
		if !monitor.controller.HasSynced() {
			logger.V(4).Info("quota monitor not synced", "resource", resource)
			return false
		}
	}
	return true
}

func (o *ReplicaSetControllerOptions) AddFlags(fs *pflag.FlagSet) {
	if o == nil {
		return
	}

	fs.Int32Var(&o.ConcurrentRSSyncs, "concurrent-replicaset-syncs", o.ConcurrentRSSyncs, "The number of replica sets that are allowed to sync concurrently. Larger number = more responsive replica management, but more CPU (and network) load")
}

// ApplyTo fills up ReplicaSetController config with options.

func (o *EndpointControllerOptions) AddFlags(fs *pflag.FlagSet) {
	if o == nil {
		return
	}

	fs.Int32Var(&o.ConcurrentEndpointSyncs, "concurrent-endpoint-syncs", o.ConcurrentEndpointSyncs, "The number of endpoint syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")

A queue is used to give a few properties:
* Ability to serially process updates received from a variety of different sources. This avoids need for other synchronization mechanisms like mutexes.
* Correctness at startup; with the sequencing above, items are only processed once all informers are synced. This means queries will not return stale data at startup.

	factory.Start(wait.NeverStop)
	syncedTypes := factory.WaitForCacheSync(wait.NeverStop)
	if len(syncedTypes) != 1 {
		t.Fatalf("informers are not synced")
	}
	for ty, ok := range syncedTypes {
		if !ok {
			t.Fatalf("failed to sync: %#v", ty)
		}
	}

	host := volumetest.NewFakeVolumeHostWithCSINodeName(t,
		tmpDir,
		client,
		csi.ProbeVolumePlugins(),
		"fakeNode",
		csiDriverLister,

type HPAControllerConfiguration struct {
	// ConcurrentHorizontalPodAutoscalerSyncs is the number of HPA objects that are allowed to sync concurrently.
	// Larger number = more responsive HPA processing, but more CPU (and network) load.
	ConcurrentHorizontalPodAutoscalerSyncs int32
	// horizontalPodAutoscalerSyncPeriod is the period for syncing the number of
	// pods in horizontal pod autoscaler.
	HorizontalPodAutoscalerSyncPeriod metav1.Duration

Laura Kassovic <******@****.***> 1701107622 -0800

	if err := cc.syncFunc(ctx, cKey); err != nil {
		cc.queue.AddRateLimited(cKey)
		if _, ignorable := err.(ignorableError); !ignorable {
			utilruntime.HandleError(fmt.Errorf("Sync %v failed with : %v", cKey, err))
		} else {
			klog.FromContext(ctx).V(4).Info("Sync certificate request failed", "csr", cKey, "err", err)
		}
		return true
	}

	cc.queue.Forget(cKey)
	return true

}

// registerHandlerAsBatched is a helper to register the provided handler as a batched handler. This allows collections to
// only implement RegisterBatch.
func registerHandlerAsBatched[T any](c internalCollection[T], f func(o Event[T])) Syncer {
	return c.RegisterBatch(func(events []Event[T], initialSync bool) {
		for _, o := range events {
			f(o)
		}
	}, true)
}

// castEvent converts an Event[I] to Event[O].