def endpoint1 = queue.newEndpoint()
        def endpoint2 = queue.newEndpoint()

        and:
        queue.empty(endpoint1)
        queue.empty(endpoint2)

        when:
        queue.dispatch(message)
        queue.dispatch(message2)
        def messages1 = []
        endpoint1.take(messages1)
        def messages2 = []
        endpoint2.take(messages2)

        then:

// addEndpoints adds an Endpoint for each unique endpoint.
func (cache *EndpointSliceCache) addEndpoints(svcPortName *ServicePortName, portNum int, endpointSet map[string]Endpoint, endpoints []discovery.Endpoint) map[string]Endpoint {
	if endpointSet == nil {
		endpointSet = map[string]Endpoint{}
	}

	// iterate through endpoints to add them to endpointSet.
	for _, endpoint := range endpoints {
		if len(endpoint.Addresses) == 0 {

		if svcInfo, ok := svcPortMap[epSvcPair.ServicePortName]; ok {
			endpointIP := proxyutil.IPPart(epSvcPair.Endpoint)
			nodePort := svcInfo.NodePort()
			var err error
			if nodePort != 0 {
				err = ct.ClearEntriesForPortNAT(endpointIP, nodePort, v1.ProtocolUDP)
				if err != nil {
					klog.ErrorS(err, "Failed to delete nodeport-related endpoint connections", "servicePortName", epSvcPair.ServicePortName)
				}
			}

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/util/retry"
	"k8s.io/klog/v2"
	endpointsv1 "k8s.io/kubernetes/pkg/api/v1/endpoints"
)

// masterCountEndpointReconciler reconciles endpoints based on a specified expected number of
// masters. masterCountEndpointReconciler implements EndpointReconciler.
type masterCountEndpointReconciler struct {

	// get the URL of the local endpoint
	localAPIEndpoint, err := GetLocalAPIEndpoint(localEndpoint)
	if err != nil {
		return "", err
	}

	// if the controlplane endpoint is defined
	if len(controlPlaneEndpoint) > 0 {
		// parse the controlplane endpoint
		var host, port string
		var err error

	plugin "k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/plugin/v1beta1"
)

// endpoint maps to a single registered device plugin. It is responsible
// for managing gRPC communications with the device plugin and caching
// device states reported by the device plugin.
type endpoint interface {
	getPreferredAllocation(available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error)

				Endpoints:  []discovery.Endpoint{endpoint1, endpoint2},
			},
			expectChanged: false,
		},
		"identical with endpoints out of order": {
			cache: NewEndpointSliceCache("", v1.IPv4Protocol, nil, nil),
			initialSlice: &discovery.EndpointSlice{
				ObjectMeta: objMeta,
				Ports:      []discovery.EndpointPort{port443},
				Endpoints:  []discovery.Endpoint{endpoint1, endpoint2},
			},

	endpointsNN := types.NamespacedName{Name: endpoints.Name, Namespace: endpoints.Namespace}
	r.metricsCache.UpdateEndpointPortCache(endpointsNN, epMetrics)

	return r.finalize(endpoints, slices)
}

// reconcileByPortMapping compares the endpoints found in existing slices with
// the list of desired endpoints and returns lists of slices to create, update,
// and delete.

Daniel Smith <******@****.***> 1651776874 +0000

SataQiu <******@****.***> 1713430345 +0800