// pattern.
//
// The delta protocol changes the request, adding unsubscribe/subscribe instead of sending full
// list of resources. On the response it adds 'removed resources' and sends changes for everything.
func (s *DiscoveryServer) DeltaAggregatedResources(stream discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error {
	return s.StreamDeltas(stream)
}

// queues any Services that are past the threshold.
func (c *Controller) checkNodeTopologyDistribution(logger klog.Logger) {
	if c.topologyCache == nil {
		return
	}
	nodes, err := c.nodeLister.List(labels.Everything())
	if err != nil {
		logger.Error(err, "Error listing Nodes")
		return
	}
	c.topologyCache.SetNodes(logger, nodes)
	serviceKeys := c.topologyCache.GetOverloadedServices()

	if runExistingState {
		h.recomputeMu.Lock()
		defer h.recomputeMu.Unlock()
	}
	initialized := !h.eventHandlers.Insert(f)
	if initialized && runExistingState {
		// Already started. Pause everything, and run through the handler.
		h.mu.Lock()
		events := make([]Event[O], 0, len(h.collectionState.outputs))
		for _, o := range h.collectionState.outputs {
			o := o
			events = append(events, Event[O]{

		t.Fatalf("unexpected error waiting for the cache to be ready")
	}

	w, err := cacher.Watch(context.Background(), "pods/ns", storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything})
	if err != nil {
		t.Fatalf("Failed to create watch: %v", err)
	}

	watchClosed := make(chan struct{})
	go func() {
		defer close(watchClosed)
		for event := range w.ResultChan() {

  // description talks about `keys`, `values`, and `entries`; here the `keys` and `values` arrays
  // are replaced by a single `elements` array but everything else works similarly.

  /**
   * The hashtable object. This can be either:
   *
   * <ul>
   *   <li>a byte[], short[], or int[], with size a power of two, created by

		// New pod has been synced. Re-mount all volumes that need it
		// (e.g. DownwardAPI)
		dswp.actualStateOfWorld.MarkRemountRequired(uniquePodName)
		// Remove any stored errors for the pod, everything went well in this processPodVolumes
		dswp.desiredStateOfWorld.PopPodErrors(uniquePodName)
	} else if dswp.podHasBeenSeenOnce(uniquePodName) {
		// For the Pod which has been processed at least once, even though some volumes

        }

        @Override
        public DeserializeMap newDeserializeSession() {
            return (classLoaderDetails, className) -> {
                // Assume everything is loaded into the current classloader
                return Class.forName(className);
            };
        }
    }

				}
				originalRevision = out.(metav1.Object).GetResourceVersion()
			}

			// prepare result and pred
			pred := storage.SelectionPredicate{
				Label: tc.selector,
				Field: fields.Everything(),
				GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
					pod, ok := obj.(*examplev1.Pod)
					if !ok {
						return nil, nil, fmt.Errorf("not a pod")
					}

					//    if using HTTPS ensure that port name is distinct so that we can setup separate RDS
					//    for each server (as each server ends up as a separate http connection manager due to filter chain match)
					// 3. No for everything else.
					if current, exists := plainTextServers[resolvedPort]; exists {
						if !canMergeProtocols(serverProtocol, protocol.Parse(current.Protocol)) && current.Bind == serverPort.Bind {

	return potentialNodes, nodeStatuses
}

func getPodDisruptionBudgets(pdbLister policylisters.PodDisruptionBudgetLister) ([]*policy.PodDisruptionBudget, error) {
	if pdbLister != nil {
		return pdbLister.List(labels.Everything())
	}
	return nil, nil
}

// pickOneNodeForPreemption chooses one node among the given nodes.
// It assumes pods in each map entry are ordered by decreasing priority.