Repository: grpc/grpc-go

Issue:
passthrough scheme not documented
As far as I can tell, the `passthrough:///` scheme does not appear to be documented anywhere. It is needed in order to properly use the bufconn package.

It is mentioned in this comment: https://github.com/grpc/grpc-go/issues/7091#issuecomment-2037663115

It is not mentioned in the language agnostic gRPC docs. https://github.com/grpc/grpc/blob/master/doc/naming.md

Hints from issue comments:
@purnesh42H: @rittneje yes, passthrough is only applicable in grpc-go and is already deprecated https://github.com/grpc/grpc-go/blob/master/resolver/passthrough/passthrough.go#L22. It only exist for backward compatibility right now.@rittneje: @purnesh42H First, the file you linked only mentions the _package_ is deprecated and should not be imported. It doesn't mention anything about the scheme itself. Second, it is the only way to get bufconn to work, so you can't just deprecate it without a proper replacement. Please re-open this issue.@dfinkel: FWIW, my team switched our bufconn uses to `localhost` as the address, since that's guaranteed to resolve with the DNS resolver.@purnesh42H: yeah, its not explicitly documented that passthrough is deprecated apart from source code but that's because it is internal to grpc-go. In release 1.64, we deprecated `grpc.Dial` to use `grpc.NewClient` https://github.com/grpc/grpc-go/releases/tag/v1.64.0. With `grpc.Dial` the default resolution scheme is passthrough which uses passthrough resolver that i linked above. With `grpc.NewClient` the default scheme is `dns`. If you want to use passthrough with `grpc.NewClient`, you need to pass the adderss prefixed with `passthrough:///`.

Could you describe what issue you are facing with bufcon though?

@github-actions: This issue is labeled as requiring an update from the reporter, and no update has been received after 6 days.  If no update is provided in the next 7 days, this issue will be automatically closed.@rittneje: @purnesh42H With bufconn, we need to use `grpc.WithContextDialer` to wire it up correctly. In this case, the target passed to `grpc.NewClient` is immaterial, but we need to prefix it with `passthrough:///` or else it will try to do a DNS resolution anyway and fail.

As `passthrough:///` is not documented (and apparently deprecated with no replacement) the only way to figure this out is to stumble across the GitHub comment I originally mentioned. Instead:
1. the package documentation for google.golang.org/grpc/test/bufconn should directly tell you what to do
2. either `passthrough:///` needs to be un-deprecated, or a proper replacement needs to be added and mentioned in the documentation@purnesh42H: @rittneje i discussed with other maintainers. We will document `passthrough` in the WithCustomDialer functions.

Solution patch:
diff --git a/balancer/weightedroundrobin/weightedroundrobin.go b/balancer/weightedroundrobin/weightedroundrobin.go
index 8741fdad..258cdd5d 100644
--- a/balancer/weightedroundrobin/weightedroundrobin.go
+++ b/balancer/weightedroundrobin/weightedroundrobin.go
@@ -56,6 +56,13 @@ func SetAddrInfo(addr resolver.Address, addrInfo AddrInfo) resolver.Address {
 	return addr
 }
 
+// SetAddrInfoInEndpoint returns a copy of endpoint in which the Attributes
+// field is updated with addrInfo.
+func SetAddrInfoInEndpoint(endpoint resolver.Endpoint, addrInfo AddrInfo) resolver.Endpoint {
+	endpoint.Attributes = endpoint.Attributes.WithValue(attributeKey{}, addrInfo)
+	return endpoint
+}
+
 // GetAddrInfo returns the AddrInfo stored in the BalancerAttributes field of
 // addr.
 func GetAddrInfo(addr resolver.Address) AddrInfo {
diff --git a/balancer_wrapper.go b/balancer_wrapper.go
index 905817b5..c2688376 100644
--- a/balancer_wrapper.go
+++ b/balancer_wrapper.go
@@ -34,7 +34,15 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-var setConnectedAddress = internal.SetConnectedAddress.(func(*balancer.SubConnState, resolver.Address))
+var (
+	setConnectedAddress = internal.SetConnectedAddress.(func(*balancer.SubConnState, resolver.Address))
+	// noOpRegisterHealthListenerFn is used when client side health checking is
+	// disabled. It sends a single READY update on the registered listener.
+	noOpRegisterHealthListenerFn = func(_ context.Context, listener func(balancer.SubConnState)) func() {
+		listener(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+		return func() {}
+	}
+)
 
 // ccBalancerWrapper sits between the ClientConn and the Balancer.
 //
@@ -277,10 +285,17 @@ type healthData struct {
 	// to the LB policy. This is stored to avoid sending updates when the
 	// SubConn has already exited connectivity state READY.
 	connectivityState connectivity.State
+	// closeHealthProducer stores function to close the ref counted health
+	// producer. The health producer is automatically closed when the SubConn
+	// state changes.
+	closeHealthProducer func()
 }
 
 func newHealthData(s connectivity.State) *healthData {
-	return &healthData{connectivityState: s}
+	return &healthData{
+		connectivityState:   s,
+		closeHealthProducer: func() {},
+	}
 }
 
 // updateState is invoked by grpc to push a subConn state update to the
@@ -413,6 +428,37 @@ func (acbw *acBalancerWrapper) closeProducers() {
 	}
 }
 
+// healthProducerRegisterFn is a type alias for the health producer's function
+// for registering listeners.
+type healthProducerRegisterFn = func(context.Context, balancer.SubConn, string, func(balancer.SubConnState)) func()
+
+// healthListenerRegFn returns a function to register a listener for health
+// updates. If client side health checks are disabled, the registered listener
+// will get a single READY (raw connectivity state) update.
+//
+// Client side health checking is enabled when all the following
+// conditions are satisfied:
+// 1. Health checking is not disabled using the dial option.
+// 2. The health package is imported.
+// 3. The health check config is present in the service config.
+func (acbw *acBalancerWrapper) healthListenerRegFn() func(context.Context, func(balancer.SubConnState)) func() {
+	if acbw.ccb.cc.dopts.disableHealthCheck {
+		return noOpRegisterHealthListenerFn
+	}
+	regHealthLisFn := internal.RegisterClientHealthCheckListener
+	if regHealthLisFn == nil {
+		// The health package is not imported.
+		return noOpRegisterHealthListenerFn
+	}
+	cfg := acbw.ac.cc.healthCheckConfig()
+	if cfg == nil {
+		return noOpRegisterHealthListenerFn
+	}
+	return func(ctx context.Context, listener func(balancer.SubConnState)) func() {
+		return regHealthLisFn.(healthProducerRegisterFn)(ctx, acbw, cfg.ServiceName, listener)
+	}
+}
+
 // RegisterHealthListener accepts a health listener from the LB policy. It sends
 // updates to the health listener as long as the SubConn's connectivity state
 // doesn't change and a new health listener is not registered. To invalidate
@@ -421,6 +467,7 @@ func (acbw *acBalancerWrapper) closeProducers() {
 func (acbw *acBalancerWrapper) RegisterHealthListener(listener func(balancer.SubConnState)) {
 	acbw.healthMu.Lock()
 	defer acbw.healthMu.Unlock()
+	acbw.healthData.closeHealthProducer()
 	// listeners should not be registered when the connectivity state
 	// isn't Ready. This may happen when the balancer registers a listener
 	// after the connectivityState is updated, but before it is notified
@@ -436,6 +483,7 @@ func (acbw *acBalancerWrapper) RegisterHealthListener(listener func(balancer.Sub
 		return
 	}
 
+	registerFn := acbw.healthListenerRegFn()
 	acbw.ccb.serializer.TrySchedule(func(ctx context.Context) {
 		if ctx.Err() != nil || acbw.ccb.balancer == nil {
 			return
@@ -443,10 +491,25 @@ func (acbw *acBalancerWrapper) RegisterHealthListener(listener func(balancer.Sub
 		// Don't send updates if a new listener is registered.
 		acbw.healthMu.Lock()
 		defer acbw.healthMu.Unlock()
-		curHD := acbw.healthData
-		if curHD != hd {
+		if acbw.healthData != hd {
 			return
 		}
-		listener(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+		// Serialize the health updates from the health producer with
+		// other calls into the LB policy.
+		listenerWrapper := func(scs balancer.SubConnState) {
+			acbw.ccb.serializer.TrySchedule(func(ctx context.Context) {
+				if ctx.Err() != nil || acbw.ccb.balancer == nil {
+					return
+				}
+				acbw.healthMu.Lock()
+				defer acbw.healthMu.Unlock()
+				if acbw.healthData != hd {
+					return
+				}
+				listener(scs)
+			})
+		}
+
+		hd.closeHealthProducer = registerFn(ctx, listenerWrapper)
 	})
 }
diff --git a/dialoptions.go b/dialoptions.go
index 7494ae59..f3a04529 100644
--- a/dialoptions.go
+++ b/dialoptions.go
@@ -428,6 +428,11 @@ func WithTimeout(d time.Duration) DialOption {
 // returned by f, gRPC checks the error's Temporary() method to decide if it
 // should try to reconnect to the network address.
 //
+// Note that gRPC by default performs name resolution on the target passed to
+// NewClient. To bypass name resolution and cause the target string to be
+// passed directly to the dialer here instead, use the "passthrough" resolver
+// by specifying it in the target string, e.g. "passthrough:target".
+//
 // Note: All supported releases of Go (as of December 2023) override the OS
 // defaults for TCP keepalive time and interval to 15s. To enable TCP keepalive
 // with OS defaults for keepalive time and interval, use a net.Dialer that sets
diff --git a/examples/features/csm_observability/server/main.go b/examples/features/csm_observability/server/main.go
index 3cafe233..7a87517c 100644
--- a/examples/features/csm_observability/server/main.go
+++ b/examples/features/csm_observability/server/main.go
@@ -66,7 +66,7 @@ func main() {
 	cleanup := csm.EnableObservability(context.Background(), opentelemetry.Options{MetricsOptions: opentelemetry.MetricsOptions{MeterProvider: provider}})
 	defer cleanup()
 
-	lis, err := net.Listen("tcp", ":"+*port)
+	lis, err := net.Listen("tcp4", "0.0.0.0:"+*port)
 	if err != nil {
 		log.Fatalf("Failed to listen: %v", err)
 	}
diff --git a/examples/features/gracefulstop/client/main.go b/examples/features/gracefulstop/client/main.go
index 61f5874b..772d128e 100644
--- a/examples/features/gracefulstop/client/main.go
+++ b/examples/features/gracefulstop/client/main.go
@@ -65,7 +65,7 @@ func main() {
 		}
 		unaryRequests++
 		time.Sleep(200 * time.Millisecond)
-		log.Printf(r.Message)
+		log.Print(r.Message)
 	}
 	log.Printf("Successful unary requests made by client: %d", unaryRequests)
 
diff --git a/examples/features/xds/README.md b/examples/features/xds/README.md
index 8cf8c441..a6a78b0f 100644
--- a/examples/features/xds/README.md
+++ b/examples/features/xds/README.md
@@ -1,7 +1,7 @@
 # gRPC xDS example
 
 xDS is the protocol initially used by Envoy, that is evolving into a universal
-data plan API for service mesh.
+data plane API for service mesh.
 
 The xDS example is a Hello World client/server capable of being configured with
 the XDS management protocol. Out-of-the-box it behaves the same as [our other
diff --git a/experimental/stats/metrics.go b/experimental/stats/metrics.go
index bf9e7f98..ee142360 100644
--- a/experimental/stats/metrics.go
+++ b/experimental/stats/metrics.go
@@ -19,6 +19,8 @@
 // Package stats contains experimental metrics/stats API's.
 package stats
 
+import "google.golang.org/grpc/stats"
+
 // MetricsRecorder records on metrics derived from metric registry.
 type MetricsRecorder interface {
 	// RecordInt64Count records the measurement alongside labels on the int
@@ -37,3 +39,16 @@ type MetricsRecorder interface {
 	// gauge associated with the provided handle.
 	RecordInt64Gauge(handle *Int64GaugeHandle, incr int64, labels ...string)
 }
+
+// Metrics is an experimental legacy alias of the now-stable stats.MetricSet.
+// Metrics will be deleted in a future release.
+type Metrics = stats.MetricSet
+
+// Metric was replaced by direct usage of strings.
+type Metric = string
+
+// NewMetrics is an experimental legacy alias of the now-stable
+// stats.NewMetricSet.  NewMetrics will be deleted in a future release.
+func NewMetrics(metrics ...Metric) *Metrics {
+	return stats.NewMetricSet(metrics...)
+}
diff --git a/health/producer.go b/health/producer.go
new file mode 100644
index 00000000..f938e579
--- /dev/null
+++ b/health/producer.go
@@ -0,0 +1,106 @@
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package health
+
+import (
+	"context"
+	"sync"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/status"
+)
+
+func init() {
+	producerBuilderSingleton = &producerBuilder{}
+	internal.RegisterClientHealthCheckListener = registerClientSideHealthCheckListener
+}
+
+type producerBuilder struct{}
+
+var producerBuilderSingleton *producerBuilder
+
+// Build constructs and returns a producer and its cleanup function.
+func (*producerBuilder) Build(cci any) (balancer.Producer, func()) {
+	p := &healthServiceProducer{
+		cc:     cci.(grpc.ClientConnInterface),
+		cancel: func() {},
+	}
+	return p, func() {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		p.cancel()
+	}
+}
+
+type healthServiceProducer struct {
+	// The following fields are initialized at build time and read-only after
+	// that and therefore do not need to be guarded by a mutex.
+	cc grpc.ClientConnInterface
+
+	mu     sync.Mutex
+	cancel func()
+}
+
+// registerClientSideHealthCheckListener accepts a listener to provide server
+// health state via the health service.
+func registerClientSideHealthCheckListener(ctx context.Context, sc balancer.SubConn, serviceName string, listener func(balancer.SubConnState)) func() {
+	pr, closeFn := sc.GetOrBuildProducer(producerBuilderSingleton)
+	p := pr.(*healthServiceProducer)
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.cancel()
+	if listener == nil {
+		return closeFn
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	p.cancel = cancel
+
+	go p.startHealthCheck(ctx, sc, serviceName, listener)
+	return closeFn
+}
+
+func (p *healthServiceProducer) startHealthCheck(ctx context.Context, sc balancer.SubConn, serviceName string, listener func(balancer.SubConnState)) {
+	newStream := func(method string) (any, error) {
+		return p.cc.NewStream(ctx, &grpc.StreamDesc{ServerStreams: true}, method)
+	}
+
+	setConnectivityState := func(state connectivity.State, err error) {
+		listener(balancer.SubConnState{
+			ConnectivityState: state,
+			ConnectionError:   err,
+		})
+	}
+
+	// Call the function through the internal variable as tests use it for
+	// mocking.
+	err := internal.HealthCheckFunc(ctx, newStream, setConnectivityState, serviceName)
+	if err == nil {
+		return
+	}
+	if status.Code(err) == codes.Unimplemented {
+		logger.Errorf("Subchannel health check is unimplemented at server side, thus health check is disabled for SubConn %p", sc)
+	} else {
+		logger.Errorf("Health checking failed for SubConn %p: %v", sc, err)
+	}
+}
diff --git a/internal/envconfig/xds.go b/internal/envconfig/xds.go
index 29f234ac..9afeb444 100644
--- a/internal/envconfig/xds.go
+++ b/internal/envconfig/xds.go
@@ -53,4 +53,10 @@ var (
 
 	// C2PResolverTestOnlyTrafficDirectorURI is the TD URI for testing.
 	C2PResolverTestOnlyTrafficDirectorURI = os.Getenv("GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI")
+
+	// XDSDualstackEndpointsEnabled is true if gRPC should read the
+	// "additional addresses" in the xDS endpoint resource.
+	// TODO: https://github.com/grpc/grpc-go/issues/7866 - Control this using
+	// an env variable when all LB policies handle endpoints.
+	XDSDualstackEndpointsEnabled = false
 )
diff --git a/internal/internal.go b/internal/internal.go
index 3afc1813..c17b9819 100644
--- a/internal/internal.go
+++ b/internal/internal.go
@@ -31,6 +31,10 @@ import (
 var (
 	// HealthCheckFunc is used to provide client-side LB channel health checking
 	HealthCheckFunc HealthChecker
+	// RegisterClientHealthCheckListener is used to provide a listener for
+	// updates from the client-side health checking service. It returns a
+	// function that can be called to stop the health producer.
+	RegisterClientHealthCheckListener any // func(ctx context.Context, sc balancer.SubConn, serviceName string, listener func(balancer.SubConnState)) func()
 	// BalancerUnregister is exported by package balancer to unregister a balancer.
 	BalancerUnregister func(name string)
 	// KeepaliveMinPingTime is the minimum ping interval.  This must be 10s by
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 0055fddd..997b0a59 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -564,7 +564,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 			t.logger.Infof("Aborting the stream early: %v", errMsg)
 		}
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     405,
+			httpStatus:     http.StatusMethodNotAllowed,
 			streamID:       streamID,
 			contentSubtype: s.contentSubtype,
 			status:         status.New(codes.Internal, errMsg),
@@ -585,7 +585,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 				stat = status.New(codes.PermissionDenied, err.Error())
 			}
 			t.controlBuf.put(&earlyAbortStream{
-				httpStatus:     200,
+				httpStatus:     http.StatusOK,
 				streamID:       s.id,
 				contentSubtype: s.contentSubtype,
 				status:         stat,
diff --git a/xds/internal/balancer/clusterresolver/clusterresolver.go b/xds/internal/balancer/clusterresolver/clusterresolver.go
index ae2c5fe9..f0a8905d 100644
--- a/xds/internal/balancer/clusterresolver/clusterresolver.go
+++ b/xds/internal/balancer/clusterresolver/clusterresolver.go
@@ -234,7 +234,7 @@ func (b *clusterResolverBalancer) updateChildConfig() {
 		b.child = newChildBalancer(b.priorityBuilder, b.cc, b.bOpts)
 	}
 
-	childCfgBytes, addrs, err := buildPriorityConfigJSON(b.priorities, &b.config.xdsLBPolicy)
+	childCfgBytes, endpoints, err := buildPriorityConfigJSON(b.priorities, &b.config.xdsLBPolicy)
 	if err != nil {
 		b.logger.Warningf("Failed to build child policy config: %v", err)
 		return
@@ -248,15 +248,33 @@ func (b *clusterResolverBalancer) updateChildConfig() {
 		b.logger.Infof("Built child policy config: %s", pretty.ToJSON(childCfg))
 	}
 
-	endpoints := make([]resolver.Endpoint, len(addrs))
-	for i, a := range addrs {
-		endpoints[i].Attributes = a.BalancerAttributes
-		endpoints[i].Addresses = []resolver.Address{a}
+	flattenedAddrs := make([]resolver.Address, len(endpoints))
+	for i := range endpoints {
+		for j := range endpoints[i].Addresses {
+			addr := endpoints[i].Addresses[j]
+			addr.BalancerAttributes = endpoints[i].Attributes
+			// If the endpoint has multiple addresses, only the first is added
+			// to the flattened address list. This ensures that LB policies
+			// that don't support endpoints create only one subchannel to a
+			// backend.
+			if j == 0 {
+				flattenedAddrs[i] = addr
+			}
+			// BalancerAttributes need to be present in endpoint addresses. This
+			// temporary workaround is required to make load reporting work
+			// with the old pickfirst policy which creates SubConns with multiple
+			// addresses. Since the addresses can be from different localities,
+			// an Address.BalancerAttribute is used to identify the locality of the
+			// address used by the transport. This workaround can be removed once
+			// the old pickfirst is removed.
+			// See https://github.com/grpc/grpc-go/issues/7339
+			endpoints[i].Addresses[j] = addr
+		}
 	}
 	if err := b.child.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: resolver.State{
 			Endpoints:     endpoints,
-			Addresses:     addrs,
+			Addresses:     flattenedAddrs,
 			ServiceConfig: b.configRaw,
 			Attributes:    b.attrsWithClient,
 		},
diff --git a/xds/internal/balancer/clusterresolver/configbuilder.go b/xds/internal/balancer/clusterresolver/configbuilder.go
index f62b8e6c..683f973b 100644
--- a/xds/internal/balancer/clusterresolver/configbuilder.go
+++ b/xds/internal/balancer/clusterresolver/configbuilder.go
@@ -48,8 +48,8 @@ type priorityConfig struct {
 	mechanism DiscoveryMechanism
 	// edsResp is set only if type is EDS.
 	edsResp xdsresource.EndpointsUpdate
-	// addresses is set only if type is DNS.
-	addresses []string
+	// endpoints is set only if type is DNS.
+	endpoints []resolver.Endpoint
 	// Each discovery mechanism has a name generator so that the child policies
 	// can reuse names between updates (EDS updates for example).
 	childNameGen *nameGenerator
@@ -71,8 +71,8 @@ type priorityConfig struct {
 //	┌──────▼─────┐  ┌─────▼──────┐
 //	│xDSLBPolicy │  │xDSLBPolicy │ (Locality and Endpoint picking layer)
 //	└────────────┘  └────────────┘
-func buildPriorityConfigJSON(priorities []priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]byte, []resolver.Address, error) {
-	pc, addrs, err := buildPriorityConfig(priorities, xdsLBPolicy)
+func buildPriorityConfigJSON(priorities []priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]byte, []resolver.Endpoint, error) {
+	pc, endpoints, err := buildPriorityConfig(priorities, xdsLBPolicy)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to build priority config: %v", err)
 	}
@@ -80,23 +80,23 @@ func buildPriorityConfigJSON(priorities []priorityConfig, xdsLBPolicy *internals
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to marshal built priority config struct into json: %v", err)
 	}
-	return ret, addrs, nil
+	return ret, endpoints, nil
 }
 
-func buildPriorityConfig(priorities []priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*priority.LBConfig, []resolver.Address, error) {
+func buildPriorityConfig(priorities []priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*priority.LBConfig, []resolver.Endpoint, error) {
 	var (
-		retConfig = &priority.LBConfig{Children: make(map[string]*priority.Child)}
-		retAddrs  []resolver.Address
+		retConfig    = &priority.LBConfig{Children: make(map[string]*priority.Child)}
+		retEndpoints []resolver.Endpoint
 	)
 	for _, p := range priorities {
 		switch p.mechanism.Type {
 		case DiscoveryMechanismTypeEDS:
-			names, configs, addrs, err := buildClusterImplConfigForEDS(p.childNameGen, p.edsResp, p.mechanism, xdsLBPolicy)
+			names, configs, endpoints, err := buildClusterImplConfigForEDS(p.childNameGen, p.edsResp, p.mechanism, xdsLBPolicy)
 			if err != nil {
 				return nil, nil, err
 			}
 			retConfig.Priorities = append(retConfig.Priorities, names...)
-			retAddrs = append(retAddrs, addrs...)
+			retEndpoints = append(retEndpoints, endpoints...)
 			odCfgs := convertClusterImplMapToOutlierDetection(configs, p.mechanism.outlierDetection)
 			for n, c := range odCfgs {
 				retConfig.Children[n] = &priority.Child{
@@ -107,9 +107,9 @@ func buildPriorityConfig(priorities []priorityConfig, xdsLBPolicy *internalservi
 			}
 			continue
 		case DiscoveryMechanismTypeLogicalDNS:
-			name, config, addrs := buildClusterImplConfigForDNS(p.childNameGen, p.addresses, p.mechanism)
+			name, config, endpoints := buildClusterImplConfigForDNS(p.childNameGen, p.endpoints, p.mechanism)
 			retConfig.Priorities = append(retConfig.Priorities, name)
-			retAddrs = append(retAddrs, addrs...)
+			retEndpoints = append(retEndpoints, endpoints...)
 			odCfg := makeClusterImplOutlierDetectionChild(config, p.mechanism.outlierDetection)
 			retConfig.Children[name] = &priority.Child{
 				Config: &internalserviceconfig.BalancerConfig{Name: outlierdetection.Name, Config: odCfg},
@@ -120,7 +120,7 @@ func buildPriorityConfig(priorities []priorityConfig, xdsLBPolicy *internalservi
 			continue
 		}
 	}
-	return retConfig, retAddrs, nil
+	return retConfig, retEndpoints, nil
 }
 
 func convertClusterImplMapToOutlierDetection(ciCfgs map[string]*clusterimpl.LBConfig, odCfg outlierdetection.LBConfig) map[string]*outlierdetection.LBConfig {
@@ -137,19 +137,19 @@ func makeClusterImplOutlierDetectionChild(ciCfg *clusterimpl.LBConfig, odCfg out
 	return &odCfgRet
 }
 
-func buildClusterImplConfigForDNS(g *nameGenerator, addrStrs []string, mechanism DiscoveryMechanism) (string, *clusterimpl.LBConfig, []resolver.Address) {
+func buildClusterImplConfigForDNS(g *nameGenerator, endpoints []resolver.Endpoint, mechanism DiscoveryMechanism) (string, *clusterimpl.LBConfig, []resolver.Endpoint) {
 	// Endpoint picking policy for DNS is hardcoded to pick_first.
 	const childPolicy = "pick_first"
-	retAddrs := make([]resolver.Address, 0, len(addrStrs))
+	retEndpoints := make([]resolver.Endpoint, len(endpoints))
 	pName := fmt.Sprintf("priority-%v", g.prefix)
-	for _, addrStr := range addrStrs {
-		retAddrs = append(retAddrs, hierarchy.Set(resolver.Address{Addr: addrStr}, []string{pName}))
+	for i, e := range endpoints {
+		retEndpoints[i] = hierarchy.SetInEndpoint(e, []string{pName})
 	}
 	return pName, &clusterimpl.LBConfig{
 		Cluster:         mechanism.Cluster,
 		TelemetryLabels: mechanism.TelemetryLabels,
 		ChildPolicy:     &internalserviceconfig.BalancerConfig{Name: childPolicy},
-	}, retAddrs
+	}, retEndpoints
 }
 
 // buildClusterImplConfigForEDS returns a list of cluster_impl configs, one for
@@ -161,7 +161,7 @@ func buildClusterImplConfigForDNS(g *nameGenerator, addrStrs []string, mechanism
 // - map{"p0":p0_config, "p1":p1_config}
 // - [p0_address_0, p0_address_1, p1_address_0, p1_address_1]
 //   - p0 addresses' hierarchy attributes are set to p0
-func buildClusterImplConfigForEDS(g *nameGenerator, edsResp xdsresource.EndpointsUpdate, mechanism DiscoveryMechanism, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]string, map[string]*clusterimpl.LBConfig, []resolver.Address, error) {
+func buildClusterImplConfigForEDS(g *nameGenerator, edsResp xdsresource.EndpointsUpdate, mechanism DiscoveryMechanism, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]string, map[string]*clusterimpl.LBConfig, []resolver.Endpoint, error) {
 	drops := make([]clusterimpl.DropConfig, 0, len(edsResp.Drops))
 	for _, d := range edsResp.Drops {
 		drops = append(drops, clusterimpl.DropConfig{
@@ -183,17 +183,17 @@ func buildClusterImplConfigForEDS(g *nameGenerator, edsResp xdsresource.Endpoint
 	}
 	retNames := g.generate(priorities)
 	retConfigs := make(map[string]*clusterimpl.LBConfig, len(retNames))
-	var retAddrs []resolver.Address
+	var retEndpoints []resolver.Endpoint
 	for i, pName := range retNames {
 		priorityLocalities := priorities[i]
-		cfg, addrs, err := priorityLocalitiesToClusterImpl(priorityLocalities, pName, mechanism, drops, xdsLBPolicy)
+		cfg, endpoints, err := priorityLocalitiesToClusterImpl(priorityLocalities, pName, mechanism, drops, xdsLBPolicy)
 		if err != nil {
 			return nil, nil, nil, err
 		}
 		retConfigs[pName] = cfg
-		retAddrs = append(retAddrs, addrs...)
+		retEndpoints = append(retEndpoints, endpoints...)
 	}
-	return retNames, retConfigs, retAddrs, nil
+	return retNames, retConfigs, retEndpoints, nil
 }
 
 // groupLocalitiesByPriority returns the localities grouped by priority.
@@ -244,8 +244,8 @@ func dedupSortedIntSlice(a []int) []int {
 // priority), and generates a cluster impl policy config, and a list of
 // addresses with their path hierarchy set to [priority-name, locality-name], so
 // priority and the xDS LB Policy know which child policy each address is for.
-func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priorityName string, mechanism DiscoveryMechanism, drops []clusterimpl.DropConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*clusterimpl.LBConfig, []resolver.Address, error) {
-	var addrs []resolver.Address
+func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priorityName string, mechanism DiscoveryMechanism, drops []clusterimpl.DropConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*clusterimpl.LBConfig, []resolver.Endpoint, error) {
+	var retEndpoints []resolver.Endpoint
 	for _, locality := range localities {
 		var lw uint32 = 1
 		if locality.Weight != 0 {
@@ -262,21 +262,24 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority
 			if endpoint.HealthStatus != xdsresource.EndpointHealthStatusHealthy && endpoint.HealthStatus != xdsresource.EndpointHealthStatusUnknown {
 				continue
 			}
-			addr := resolver.Address{Addr: endpoint.Address}
-			addr = hierarchy.Set(addr, []string{priorityName, localityStr})
-			addr = internal.SetLocalityID(addr, locality.ID)
+			resolverEndpoint := resolver.Endpoint{}
+			for _, as := range endpoint.Addresses {
+				resolverEndpoint.Addresses = append(resolverEndpoint.Addresses, resolver.Address{Addr: as})
+			}
+			resolverEndpoint = hierarchy.SetInEndpoint(resolverEndpoint, []string{priorityName, localityStr})
+			resolverEndpoint = internal.SetLocalityIDInEndpoint(resolverEndpoint, locality.ID)
 			// "To provide the xds_wrr_locality load balancer information about
 			// locality weights received from EDS, the cluster resolver will
 			// populate a new locality weight attribute for each address The
 			// attribute will have the weight (as an integer) of the locality
 			// the address is part of." - A52
-			addr = wrrlocality.SetAddrInfo(addr, wrrlocality.AddrInfo{LocalityWeight: lw})
+			resolverEndpoint = wrrlocality.SetAddrInfoInEndpoint(resolverEndpoint, wrrlocality.AddrInfo{LocalityWeight: lw})
 			var ew uint32 = 1
 			if endpoint.Weight != 0 {
 				ew = endpoint.Weight
 			}
-			addr = weightedroundrobin.SetAddrInfo(addr, weightedroundrobin.AddrInfo{Weight: lw * ew})
-			addrs = append(addrs, addr)
+			resolverEndpoint = weightedroundrobin.SetAddrInfoInEndpoint(resolverEndpoint, weightedroundrobin.AddrInfo{Weight: lw * ew})
+			retEndpoints = append(retEndpoints, resolverEndpoint)
 		}
 	}
 	return &clusterimpl.LBConfig{
@@ -287,5 +290,5 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority
 		TelemetryLabels:       mechanism.TelemetryLabels,
 		DropCategories:        drops,
 		ChildPolicy:           xdsLBPolicy,
-	}, addrs, nil
+	}, retEndpoints, nil
 }
diff --git a/xds/internal/balancer/clusterresolver/resource_resolver.go b/xds/internal/balancer/clusterresolver/resource_resolver.go
index 3bcfba87..d9315c3a 100644
--- a/xds/internal/balancer/clusterresolver/resource_resolver.go
+++ b/xds/internal/balancer/clusterresolver/resource_resolver.go
@@ -24,6 +24,7 @@ import (
 
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcsync"
+	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 )
 
@@ -294,8 +295,8 @@ func (rr *resourceResolver) generateLocked(onDone xdsresource.OnDoneFunc) {
 		switch uu := u.(type) {
 		case xdsresource.EndpointsUpdate:
 			ret = append(ret, priorityConfig{mechanism: rDM.dm, edsResp: uu, childNameGen: rDM.childNameGen})
-		case []string:
-			ret = append(ret, priorityConfig{mechanism: rDM.dm, addresses: uu, childNameGen: rDM.childNameGen})
+		case []resolver.Endpoint:
+			ret = append(ret, priorityConfig{mechanism: rDM.dm, endpoints: uu, childNameGen: rDM.childNameGen})
 		}
 	}
 	select {
diff --git a/xds/internal/balancer/clusterresolver/resource_resolver_dns.go b/xds/internal/balancer/clusterresolver/resource_resolver_dns.go
index cfc871d3..5f7a2115 100644
--- a/xds/internal/balancer/clusterresolver/resource_resolver_dns.go
+++ b/xds/internal/balancer/clusterresolver/resource_resolver_dns.go
@@ -47,7 +47,7 @@ type dnsDiscoveryMechanism struct {
 	logger           *grpclog.PrefixLogger
 
 	mu             sync.Mutex
-	addrs          []string
+	endpoints      []resolver.Endpoint
 	updateReceived bool
 }
 
@@ -103,7 +103,7 @@ func (dr *dnsDiscoveryMechanism) lastUpdate() (any, bool) {
 	if !dr.updateReceived {
 		return nil, false
 	}
-	return dr.addrs, true
+	return dr.endpoints, true
 }
 
 func (dr *dnsDiscoveryMechanism) resolveNow() {
@@ -133,23 +133,15 @@ func (dr *dnsDiscoveryMechanism) UpdateState(state resolver.State) error {
 	}
 
 	dr.mu.Lock()
-	var addrs []string
-	if len(state.Endpoints) > 0 {
-		// Assume 1 address per endpoint, which is how DNS is expected to
-		// behave.  The slice will grow as needed, however.
-		addrs = make([]string, 0, len(state.Endpoints))
-		for _, e := range state.Endpoints {
-			for _, a := range e.Addresses {
-				addrs = append(addrs, a.Addr)
-			}
-		}
-	} else {
-		addrs = make([]string, len(state.Addresses))
+	var endpoints = state.Endpoints
+	if len(endpoints) == 0 {
+		endpoints = make([]resolver.Endpoint, len(state.Addresses))
 		for i, a := range state.Addresses {
-			addrs[i] = a.Addr
+			endpoints[i] = resolver.Endpoint{Addresses: []resolver.Address{a}}
+			endpoints[i].Attributes = a.BalancerAttributes
 		}
 	}
-	dr.addrs = addrs
+	dr.endpoints = endpoints
 	dr.updateReceived = true
 	dr.mu.Unlock()
 
@@ -172,7 +164,7 @@ func (dr *dnsDiscoveryMechanism) ReportError(err error) {
 		dr.mu.Unlock()
 		return
 	}
-	dr.addrs = nil
+	dr.endpoints = nil
 	dr.updateReceived = true
 	dr.mu.Unlock()
 
diff --git a/xds/internal/balancer/wrrlocality/balancer.go b/xds/internal/balancer/wrrlocality/balancer.go
index 943ee780..2b289a81 100644
--- a/xds/internal/balancer/wrrlocality/balancer.go
+++ b/xds/internal/balancer/wrrlocality/balancer.go
@@ -120,6 +120,13 @@ func SetAddrInfo(addr resolver.Address, addrInfo AddrInfo) resolver.Address {
 	return addr
 }
 
+// SetAddrInfoInEndpoint returns a copy of endpoint in which the Attributes
+// field is updated with AddrInfo.
+func SetAddrInfoInEndpoint(endpoint resolver.Endpoint, addrInfo AddrInfo) resolver.Endpoint {
+	endpoint.Attributes = endpoint.Attributes.WithValue(attributeKey{}, addrInfo)
+	return endpoint
+}
+
 func (a AddrInfo) String() string {
 	return fmt.Sprintf("Locality Weight: %d", a.LocalityWeight)
 }
diff --git a/xds/internal/internal.go b/xds/internal/internal.go
index 1d8a6b03..74c91952 100644
--- a/xds/internal/internal.go
+++ b/xds/internal/internal.go
@@ -86,6 +86,12 @@ func SetLocalityID(addr resolver.Address, l LocalityID) resolver.Address {
 	return addr
 }
 
+// SetLocalityIDInEndpoint sets locality ID in endpoint to l.
+func SetLocalityIDInEndpoint(endpoint resolver.Endpoint, l LocalityID) resolver.Endpoint {
+	endpoint.Attributes = endpoint.Attributes.WithValue(localityKey, l)
+	return endpoint
+}
+
 // ResourceTypeMapForTesting maps TypeUrl to corresponding ResourceType.
 var ResourceTypeMapForTesting map[string]any
 
diff --git a/xds/internal/xdsclient/authority.go b/xds/internal/xdsclient/authority.go
index 24673a8d..f81685a4 100644
--- a/xds/internal/xdsclient/authority.go
+++ b/xds/internal/xdsclient/authority.go
@@ -639,6 +639,9 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 			if a.logger.V(2) {
 				a.logger.Infof("Resource type %q with resource name %q found in cache: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
 			}
+			// state can only be accessed in the context of an
+			// xdsClientSerializer callback. Hence making a copy of the cached
+			// resource here for watchCallbackSerializer.
 			resource := state.cache
 			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnUpdate(resource, func() {}) })
 		}
@@ -646,9 +649,13 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 		// immediately as well.
 		if state.md.Status == xdsresource.ServiceStatusNACKed {
 			if a.logger.V(2) {
-				a.logger.Infof("Resource type %q with resource name %q was NACKed: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
+				a.logger.Infof("Resource type %q with resource name %q was NACKed", rType.TypeName(), resourceName)
 			}
-			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(state.md.ErrState.Err, func() {}) })
+			// state can only be accessed in the context of an
+			// xdsClientSerializer callback. Hence making a copy of the error
+			// here for watchCallbackSerializer.
+			err := state.md.ErrState.Err
+			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
 		}
 		// If the metadata field is updated to indicate that the management
 		// server does not have this resource, notify the new watcher.
@@ -687,7 +694,7 @@ func (a *authority) unwatchResource(rType xdsresource.Type, resourceName string,
 			delete(state.watchers, watcher)
 			if len(state.watchers) > 0 {
 				if a.logger.V(2) {
-					a.logger.Infof("%d more watchers exist for type %q, resource name %q", rType.TypeName(), resourceName)
+					a.logger.Infof("Other watchers exist for type %q, resource name %q", rType.TypeName(), resourceName)
 				}
 				return
 			}
diff --git a/xds/internal/xdsclient/client_new.go b/xds/internal/xdsclient/client_new.go
index d8f9d6c9..55299c45 100644
--- a/xds/internal/xdsclient/client_new.go
+++ b/xds/internal/xdsclient/client_new.go
@@ -26,7 +26,6 @@ import (
 
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/backoff"
-	"google.golang.org/grpc/internal/cache"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	xdsclientinternal "google.golang.org/grpc/xds/internal/xdsclient/internal"
@@ -61,11 +60,11 @@ func New(name string) (XDSClient, func(), error) {
 	if err != nil {
 		return nil, nil, fmt.Errorf("xds: failed to get xDS bootstrap config: %v", err)
 	}
-	return newRefCounted(name, config, defaultWatchExpiryTimeout, defaultIdleChannelExpiryTimeout, backoff.DefaultExponential.Backoff)
+	return newRefCounted(name, config, defaultWatchExpiryTimeout, backoff.DefaultExponential.Backoff)
 }
 
 // newClientImpl returns a new xdsClient with the given config.
-func newClientImpl(config *bootstrap.Config, watchExpiryTimeout, idleChannelExpiryTimeout time.Duration, streamBackoff func(int) time.Duration) (*clientImpl, error) {
+func newClientImpl(config *bootstrap.Config, watchExpiryTimeout time.Duration, streamBackoff func(int) time.Duration) (*clientImpl, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	c := &clientImpl{
 		done:               grpcsync.NewEvent(),
@@ -78,7 +77,6 @@ func newClientImpl(config *bootstrap.Config, watchExpiryTimeout, idleChannelExpi
 		transportBuilder:   &grpctransport.Builder{},
 		resourceTypes:      newResourceTypeRegistry(),
 		xdsActiveChannels:  make(map[string]*channelState),
-		xdsIdleChannels:    cache.NewTimeoutCache(idleChannelExpiryTimeout),
 	}
 
 	for name, cfg := range config.Authorities() {
@@ -121,10 +119,6 @@ type OptionsForTesting struct {
 	// unspecified, uses the default value used in non-test code.
 	WatchExpiryTimeout time.Duration
 
-	// IdleChannelExpiryTimeout is the timeout before idle xdsChannels are
-	// deleted. If unspecified, uses the default value used in non-test code.
-	IdleChannelExpiryTimeout time.Duration
-
 	// StreamBackoffAfterFailure is the backoff function used to determine the
 	// backoff duration after stream failures.
 	// If unspecified, uses the default value used in non-test code.
@@ -147,9 +141,6 @@ func NewForTesting(opts OptionsForTesting) (XDSClient, func(), error) {
 	if opts.WatchExpiryTimeout == 0 {
 		opts.WatchExpiryTimeout = defaultWatchExpiryTimeout
 	}
-	if opts.IdleChannelExpiryTimeout == 0 {
-		opts.IdleChannelExpiryTimeout = defaultIdleChannelExpiryTimeout
-	}
 	if opts.StreamBackoffAfterFailure == nil {
 		opts.StreamBackoffAfterFailure = defaultStreamBackoffFunc
 	}
@@ -158,7 +149,7 @@ func NewForTesting(opts OptionsForTesting) (XDSClient, func(), error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	return newRefCounted(opts.Name, config, opts.WatchExpiryTimeout, opts.IdleChannelExpiryTimeout, opts.StreamBackoffAfterFailure)
+	return newRefCounted(opts.Name, config, opts.WatchExpiryTimeout, opts.StreamBackoffAfterFailure)
 }
 
 // GetForTesting returns an xDS client created earlier using the given name.
diff --git a/xds/internal/xdsclient/client_refcounted.go b/xds/internal/xdsclient/client_refcounted.go
index 1c105ac4..5a256c1b 100644
--- a/xds/internal/xdsclient/client_refcounted.go
+++ b/xds/internal/xdsclient/client_refcounted.go
@@ -27,10 +27,7 @@ import (
 	"google.golang.org/grpc/internal/xds/bootstrap"
 )
 
-const (
-	defaultWatchExpiryTimeout       = 15 * time.Second
-	defaultIdleChannelExpiryTimeout = 5 * time.Minute
-)
+const defaultWatchExpiryTimeout = 15 * time.Second
 
 var (
 	// The following functions are no-ops in the actual code, but can be
@@ -62,7 +59,7 @@ func clientRefCountedClose(name string) {
 // newRefCounted creates a new reference counted xDS client implementation for
 // name, if one does not exist already. If an xDS client for the given name
 // exists, it gets a reference to it and returns it.
-func newRefCounted(name string, config *bootstrap.Config, watchExpiryTimeout, idleChannelExpiryTimeout time.Duration, streamBackoff func(int) time.Duration) (XDSClient, func(), error) {
+func newRefCounted(name string, config *bootstrap.Config, watchExpiryTimeout time.Duration, streamBackoff func(int) time.Duration) (XDSClient, func(), error) {
 	clientsMu.Lock()
 	defer clientsMu.Unlock()
 
@@ -72,7 +69,7 @@ func newRefCounted(name string, config *bootstrap.Config, watchExpiryTimeout, id
 	}
 
 	// Create the new client implementation.
-	c, err := newClientImpl(config, watchExpiryTimeout, idleChannelExpiryTimeout, streamBackoff)
+	c, err := newClientImpl(config, watchExpiryTimeout, streamBackoff)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/xds/internal/xdsclient/clientimpl.go b/xds/internal/xdsclient/clientimpl.go
index df0949e2..bb8d9040 100644
--- a/xds/internal/xdsclient/clientimpl.go
+++ b/xds/internal/xdsclient/clientimpl.go
@@ -25,7 +25,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"google.golang.org/grpc/internal/cache"
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/xds/bootstrap"
@@ -63,14 +62,9 @@ type clientImpl struct {
 	// these channels, and forwards updates from the channels to each of these
 	// authorities.
 	//
-	// Once all references to a channel are dropped, the channel is moved to the
-	// idle cache where it lives for a configured duration before being closed.
-	// If the channel is required before the idle timeout fires, it is revived
-	// from the idle cache and used.
+	// Once all references to a channel are dropped, the channel is closed.
 	channelsMu        sync.Mutex
 	xdsActiveChannels map[string]*channelState // Map from server config to in-use xdsChannels.
-	xdsIdleChannels   *cache.TimeoutCache      // Map from server config to idle xdsChannels.
-	closeCond         *sync.Cond
 }
 
 // channelState represents the state of an xDS channel. It tracks the number of
@@ -173,21 +167,6 @@ func (c *clientImpl) close() {
 		c.close()
 	}
 
-	// Similarly, closing idle channels cannot be done with the lock held, for
-	// the same reason as described above.  So, we clear the idle cache in a
-	// goroutine and use a condition variable to wait on the condition that the
-	// idle cache has zero entries. The Wait() method on the condition variable
-	// releases the lock and blocks the goroutine until signaled (which happens
-	// when an idle channel is removed from the cache and closed), and grabs the
-	// lock before returning.
-	c.channelsMu.Lock()
-	c.closeCond = sync.NewCond(&c.channelsMu)
-	go c.xdsIdleChannels.Clear(true)
-	for c.xdsIdleChannels.Len() > 0 {
-		c.closeCond.Wait()
-	}
-	c.channelsMu.Unlock()
-
 	c.serializerClose()
 	<-c.serializer.Done()
 
@@ -289,27 +268,15 @@ func (c *clientImpl) getOrCreateChannel(serverConfig *bootstrap.ServerConfig, in
 		c.logger.Infof("Received request for a reference to an xdsChannel for server config %q", serverConfig)
 	}
 
-	// Use an active channel, if one exists for this server config.
+	// Use an existing channel, if one exists for this server config.
 	if state, ok := c.xdsActiveChannels[serverConfig.String()]; ok {
 		if c.logger.V(2) {
-			c.logger.Infof("Reusing an active xdsChannel for server config %q", serverConfig)
+			c.logger.Infof("Reusing an existing xdsChannel for server config %q", serverConfig)
 		}
 		initLocked(state)
 		return state.channel, c.releaseChannel(serverConfig, state, deInitLocked), nil
 	}
 
-	// If an idle channel exists for this server config, remove it from the
-	// idle cache and add it to the map of active channels, and return it.
-	if s, ok := c.xdsIdleChannels.Remove(serverConfig.String()); ok {
-		if c.logger.V(2) {
-			c.logger.Infof("Reviving an xdsChannel from the idle cache for server config %q", serverConfig)
-		}
-		state := s.(*channelState)
-		c.xdsActiveChannels[serverConfig.String()] = state
-		initLocked(state)
-		return state.channel, c.releaseChannel(serverConfig, state, deInitLocked), nil
-	}
-
 	if c.logger.V(2) {
 		c.logger.Infof("Creating a new xdsChannel for server config %q", serverConfig)
 	}
@@ -345,9 +312,7 @@ func (c *clientImpl) getOrCreateChannel(serverConfig *bootstrap.ServerConfig, in
 }
 
 // releaseChannel is a function that is called when a reference to an xdsChannel
-// needs to be released. It handles the logic of moving the channel to an idle
-// cache if there are no other active references, and closing the channel if it
-// remains in the idle cache for the configured duration.
+// needs to be released. It handles closing channels with no active references.
 //
 // The function takes the following parameters:
 // - serverConfig: the server configuration for the xdsChannel
@@ -360,7 +325,6 @@ func (c *clientImpl) getOrCreateChannel(serverConfig *bootstrap.ServerConfig, in
 func (c *clientImpl) releaseChannel(serverConfig *bootstrap.ServerConfig, state *channelState, deInitLocked func(*channelState)) func() {
 	return grpcsync.OnceFunc(func() {
 		c.channelsMu.Lock()
-		defer c.channelsMu.Unlock()
 
 		if c.logger.V(2) {
 			c.logger.Infof("Received request to release a reference to an xdsChannel for server config %q", serverConfig)
@@ -372,40 +336,17 @@ func (c *clientImpl) releaseChannel(serverConfig *bootstrap.ServerConfig, state
 			if c.logger.V(2) {
 				c.logger.Infof("xdsChannel %p has other active references", state.channel)
 			}
+			c.channelsMu.Unlock()
 			return
 		}
 
-		// Move the channel to the idle cache instead of closing
-		// immediately. If the channel remains in the idle cache for
-		// the configured duration, it will get closed.
 		delete(c.xdsActiveChannels, serverConfig.String())
 		if c.logger.V(2) {
-			c.logger.Infof("Moving xdsChannel [%p] for server config %s to the idle cache", state.channel, serverConfig)
+			c.logger.Infof("Closing xdsChannel [%p] for server config %s", state.channel, serverConfig)
 		}
+		channelToClose := state.channel
+		c.channelsMu.Unlock()
 
-		// The idle cache expiry timeout results in the channel getting
-		// closed in another serializer callback.
-		c.xdsIdleChannels.Add(serverConfig.String(), state, grpcsync.OnceFunc(func() {
-			c.channelsMu.Lock()
-			channelToClose := state.channel
-			c.channelsMu.Unlock()
-
-			if c.logger.V(2) {
-				c.logger.Infof("Idle cache expiry timeout fired for xdsChannel [%p] for server config %s", state.channel, serverConfig)
-			}
-			channelToClose.close()
-
-			// If the channel is being closed as a result of the xDS client
-			// being closed, closeCond is non-nil and we need to signal from
-			// here to unblock Close().  Holding the lock is not necessary
-			// to call Signal() on a condition variable. But the field
-			// `c.closeCond` needs to guarded by the lock, which is why we
-			// acquire it here.
-			c.channelsMu.Lock()
-			if c.closeCond != nil {
-				c.closeCond.Signal()
-			}
-			c.channelsMu.Unlock()
-		}))
+		channelToClose.close()
 	})
 }
diff --git a/xds/internal/xdsclient/xdsresource/type_eds.go b/xds/internal/xdsclient/xdsresource/type_eds.go
index 1254d250..f94a17e7 100644
--- a/xds/internal/xdsclient/xdsresource/type_eds.go
+++ b/xds/internal/xdsclient/xdsresource/type_eds.go
@@ -49,7 +49,7 @@ const (
 
 // Endpoint contains information of an endpoint.
 type Endpoint struct {
-	Address      string
+	Addresses    []string
 	HealthStatus EndpointHealthStatus
 	Weight       uint32
 }
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_eds.go b/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
index f65845b7..fd780d66 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
@@ -26,6 +26,7 @@ import (
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3typepb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/pretty"
 	"google.golang.org/grpc/xds/internal"
 	"google.golang.org/protobuf/proto"
@@ -93,14 +94,22 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			}
 			weight = w.GetValue()
 		}
-		addr := parseAddress(lbEndpoint.GetEndpoint().GetAddress().GetSocketAddress())
-		if uniqueEndpointAddrs[addr] {
-			return nil, fmt.Errorf("duplicate endpoint with the same address %s", addr)
+		addrs := []string{parseAddress(lbEndpoint.GetEndpoint().GetAddress().GetSocketAddress())}
+		if envconfig.XDSDualstackEndpointsEnabled {
+			for _, sa := range lbEndpoint.GetEndpoint().GetAdditionalAddresses() {
+				addrs = append(addrs, parseAddress(sa.GetAddress().GetSocketAddress()))
+			}
+		}
+
+		for _, a := range addrs {
+			if uniqueEndpointAddrs[a] {
+				return nil, fmt.Errorf("duplicate endpoint with the same address %s", a)
+			}
+			uniqueEndpointAddrs[a] = true
 		}
-		uniqueEndpointAddrs[addr] = true
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
-			Address:      addr,
+			Addresses:    addrs,
 			Weight:       weight,
 		})
 	}



Test patch:
diff --git a/internal/testutils/channel.go b/internal/testutils/channel.go
index 720d8537..d1bb0458 100644
--- a/internal/testutils/channel.go
+++ b/internal/testutils/channel.go
@@ -70,6 +70,17 @@ func (c *Channel) ReceiveOrFail() (any, bool) {
 	}
 }
 
+// Drain drains the channel by repeatedly reading from it until it is empty.
+func (c *Channel) Drain() {
+	for {
+		select {
+		case <-c.C:
+		default:
+			return
+		}
+	}
+}
+
 // Receive returns the value received on the underlying channel, or the error
 // returned by ctx if it is closed or cancelled.
 func (c *Channel) Receive(ctx context.Context) (any, error) {
diff --git a/internal/testutils/xds/e2e/clientresources.go b/internal/testutils/xds/e2e/clientresources.go
index 526f22ae..bf063a2e 100644
--- a/internal/testutils/xds/e2e/clientresources.go
+++ b/internal/testutils/xds/e2e/clientresources.go
@@ -692,9 +692,9 @@ type LocalityOptions struct {
 // BackendOptions contains options to configure individual backends in a
 // locality.
 type BackendOptions struct {
-	// Port number on which the backend is accepting connections. All backends
+	// Ports on which the backend is accepting connections. All backends
 	// are expected to run on localhost, hence host name is not stored here.
-	Port uint32
+	Ports []uint32
 	// Health status of the backend. Default is UNKNOWN which is treated the
 	// same as HEALTHY.
 	HealthStatus v3corepb.HealthStatus
@@ -722,7 +722,7 @@ type EndpointOptions struct {
 func DefaultEndpoint(clusterName string, host string, ports []uint32) *v3endpointpb.ClusterLoadAssignment {
 	var bOpts []BackendOptions
 	for _, p := range ports {
-		bOpts = append(bOpts, BackendOptions{Port: p, Weight: 1})
+		bOpts = append(bOpts, BackendOptions{Ports: []uint32{p}, Weight: 1})
 	}
 	return EndpointResourceWithOptions(EndpointOptions{
 		ClusterName: clusterName,
@@ -747,15 +747,28 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 			if b.Weight == 0 {
 				b.Weight = 1
 			}
+			additionalAddresses := make([]*v3endpointpb.Endpoint_AdditionalAddress, len(b.Ports)-1)
+			for i, p := range b.Ports[1:] {
+				additionalAddresses[i] = &v3endpointpb.Endpoint_AdditionalAddress{
+					Address: &v3corepb.Address{Address: &v3corepb.Address_SocketAddress{
+						SocketAddress: &v3corepb.SocketAddress{
+							Protocol:      v3corepb.SocketAddress_TCP,
+							Address:       opts.Host,
+							PortSpecifier: &v3corepb.SocketAddress_PortValue{PortValue: p},
+						}},
+					},
+				}
+			}
 			lbEndpoints = append(lbEndpoints, &v3endpointpb.LbEndpoint{
 				HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{Endpoint: &v3endpointpb.Endpoint{
 					Address: &v3corepb.Address{Address: &v3corepb.Address_SocketAddress{
 						SocketAddress: &v3corepb.SocketAddress{
 							Protocol:      v3corepb.SocketAddress_TCP,
 							Address:       opts.Host,
-							PortSpecifier: &v3corepb.SocketAddress_PortValue{PortValue: b.Port},
+							PortSpecifier: &v3corepb.SocketAddress_PortValue{PortValue: b.Ports[0]},
 						},
 					}},
+					AdditionalAddresses: additionalAddresses,
 				}},
 				HealthStatus:        b.HealthStatus,
 				LoadBalancingWeight: &wrapperspb.UInt32Value{Value: b.Weight},
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
index e56c0fe9..ac671e29 100644
--- a/stats/opentelemetry/e2e_test.go
+++ b/stats/opentelemetry/e2e_test.go
@@ -436,7 +436,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 			Host:        "localhost",
 			Localities: []e2e.LocalityOptions{
 				{
-					Backends: []e2e.BackendOptions{{Port: port1}, {Port: port2}},
+					Backends: []e2e.BackendOptions{{Ports: []uint32{port1}}, {Ports: []uint32{port2}}},
 					Weight:   1,
 				},
 			},
@@ -473,7 +473,7 @@ func (s) TestWRRMetrics(t *testing.T) {
 	// scheduler.
 	receivedExpectedMetrics := grpcsync.NewEvent()
 	go func() {
-		for !receivedExpectedMetrics.HasFired() {
+		for !receivedExpectedMetrics.HasFired() && ctx.Err() == nil {
 			client.EmptyCall(ctx, &testpb.Empty{})
 			time.Sleep(2 * time.Millisecond)
 		}
diff --git a/test/balancer_switching_test.go b/test/balancer_switching_test.go
index e5da19d3..2e46fcdb 100644
--- a/test/balancer_switching_test.go
+++ b/test/balancer_switching_test.go
@@ -128,22 +128,24 @@ func (s) TestBalancerSwitch_Basic(t *testing.T) {
 	addrs := stubBackendsToResolverAddrs(backends)
 
 	r := manual.NewBuilderWithScheme("whatever")
-	cc, err := grpc.Dial(r.Scheme()+":///test.server", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
+
+	r.InitialState(resolver.State{Addresses: addrs})
+
+	cc, err := grpc.NewClient(r.Scheme()+":///test.server", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
 	if err != nil {
-		t.Fatalf("grpc.Dial() failed: %v", err)
+		t.Fatalf("grpc.NewClient() failed: %v", err)
 	}
 	defer cc.Close()
 
 	// Push a resolver update without an LB policy in the service config. The
 	// channel should pick the default LB policy, which is pick_first.
-	r.UpdateState(resolver.State{Addresses: addrs})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := pfutil.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
 		t.Fatal(err)
 	}
 
-	// Push a resolver update with the service config specifying "round_robin".
+	// Push a resolver update with a service config specifying "round_robin".
 	r.UpdateState(resolver.State{
 		Addresses:     addrs,
 		ServiceConfig: parseServiceConfig(t, r, rrServiceConfig),
@@ -153,7 +155,7 @@ func (s) TestBalancerSwitch_Basic(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Push a resolver update with the service config specifying "pick_first".
+	// Push another resolver update with a service config specifying "pick_first".
 	r.UpdateState(resolver.State{
 		Addresses:     addrs,
 		ServiceConfig: parseServiceConfig(t, r, pickFirstServiceConfig),
diff --git a/test/healthcheck_test.go b/test/healthcheck_test.go
index 424682d0..fac56524 100644
--- a/test/healthcheck_test.go
+++ b/test/healthcheck_test.go
@@ -28,12 +28,17 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/pickfirst"
+	"google.golang.org/grpc/balancer/pickfirst/pickfirstleaf"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/health"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
 	"google.golang.org/grpc/internal/channelz"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/resolver"
@@ -46,6 +51,47 @@ import (
 	testpb "google.golang.org/grpc/interop/grpc_testing"
 )
 
+const healthCheckingPetiolePolicyName = "health_checking_petiole_policy"
+
+var (
+	// healthCheckTestPolicyName is the LB policy used for testing the health check
+	// service.
+	healthCheckTestPolicyName = "round_robin"
+)
+
+func init() {
+	balancer.Register(&healthCheckingPetiolePolicyBuilder{})
+	// Till dualstack changes are not implemented and round_robin doesn't
+	// delegate to pickfirst, test a fake petiole policy that delegates to
+	// the new pickfirst balancer.
+	// TODO: https://github.com/grpc/grpc-go/issues/7906 - Remove the fake
+	// petiole policy one round robin starts delegating to pickfirst.
+	if envconfig.NewPickFirstEnabled {
+		healthCheckTestPolicyName = healthCheckingPetiolePolicyName
+	}
+}
+
+type healthCheckingPetiolePolicyBuilder struct{}
+
+func (bb *healthCheckingPetiolePolicyBuilder) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
+	return &healthCheckingPetiolePolicy{
+		Balancer: balancer.Get(pickfirstleaf.Name).Build(cc, opts),
+	}
+}
+
+func (bb *healthCheckingPetiolePolicyBuilder) Name() string {
+	return healthCheckingPetiolePolicyName
+}
+
+func (b *healthCheckingPetiolePolicy) UpdateClientConnState(state balancer.ClientConnState) error {
+	state.ResolverState = pickfirstleaf.EnableHealthListener(state.ResolverState)
+	return b.Balancer.UpdateClientConnState(state)
+}
+
+type healthCheckingPetiolePolicy struct {
+	balancer.Balancer
+}
+
 func newTestHealthServer() *testHealthServer {
 	return newTestHealthServerWithWatchFunc(defaultWatchFunc)
 }
@@ -261,12 +307,12 @@ func (s) TestHealthCheckHealthServerNotRegistered(t *testing.T) {
 	cc, r := setupClient(t, nil)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "foo"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -288,12 +334,12 @@ func (s) TestHealthCheckWithGoAway(t *testing.T) {
 	tc := testgrpc.NewTestServiceClient(cc)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "foo"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -366,12 +412,12 @@ func (s) TestHealthCheckWithConnClose(t *testing.T) {
 	tc := testgrpc.NewTestServiceClient(cc)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "foo"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -414,12 +460,12 @@ func (s) TestHealthCheckWithAddrConnDrain(t *testing.T) {
 	hcEnterChan, hcExitChan := setupHealthCheckWrapper(t)
 	cc, r := setupClient(t, &clientConfig{})
 	tc := testgrpc.NewTestServiceClient(cc)
-	sc := parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)
+	sc := parseServiceConfig(t, r, fmt.Sprintf(`{
+		"healthCheckConfig": {
+			"serviceName": "foo"
+		},
+		"loadBalancingConfig": [{"%s":{}}]
+	}`, healthCheckTestPolicyName))
 	r.UpdateState(resolver.State{
 		Addresses:     []resolver.Address{{Addr: lis.Addr().String()}},
 		ServiceConfig: sc,
@@ -496,12 +542,12 @@ func (s) TestHealthCheckWithClientConnClose(t *testing.T) {
 	tc := testgrpc.NewTestServiceClient(cc)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, (fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "foo"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName)))})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -563,12 +609,12 @@ func (s) TestHealthCheckWithoutSetConnectivityStateCalledAddrConnShutDown(t *tes
 	// The serviceName "delay" is specially handled at server side, where response will not be sent
 	// back to client immediately upon receiving the request (client should receive no response until
 	// test ends).
-	sc := parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "delay"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)
+	sc := parseServiceConfig(t, r, fmt.Sprintf(`{
+		"healthCheckConfig": {
+			"serviceName": "delay"
+		},
+		"loadBalancingConfig": [{"%s":{}}]
+	}`, healthCheckTestPolicyName))
 	r.UpdateState(resolver.State{
 		Addresses:     []resolver.Address{{Addr: lis.Addr().String()}},
 		ServiceConfig: sc,
@@ -628,12 +674,12 @@ func (s) TestHealthCheckWithoutSetConnectivityStateCalled(t *testing.T) {
 	// test ends).
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "delay"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "delay"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	select {
 	case <-hcExitChan:
@@ -666,12 +712,12 @@ func testHealthCheckDisableWithDialOption(t *testing.T, addr string) {
 	tc := testgrpc.NewTestServiceClient(cc)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: addr}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "foo"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "foo"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -772,12 +818,12 @@ func (s) TestHealthCheckChannelzCountingCallSuccess(t *testing.T) {
 	_, r := setupClient(t, nil)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "channelzSuccess"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "channelzSuccess"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	if err := verifyResultWithDelay(func() (bool, error) {
 		cm, _ := channelz.GetTopChannels(0, 0)
@@ -821,12 +867,12 @@ func (s) TestHealthCheckChannelzCountingCallFailure(t *testing.T) {
 	_, r := setupClient(t, nil)
 	r.UpdateState(resolver.State{
 		Addresses: []resolver.Address{{Addr: lis.Addr().String()}},
-		ServiceConfig: parseServiceConfig(t, r, `{
-	"healthCheckConfig": {
-		"serviceName": "channelzFailure"
-	},
-	"loadBalancingConfig": [{"round_robin":{}}]
-}`)})
+		ServiceConfig: parseServiceConfig(t, r, fmt.Sprintf(`{
+			"healthCheckConfig": {
+				"serviceName": "channelzFailure"
+			},
+			"loadBalancingConfig": [{"%s":{}}]
+		}`, healthCheckTestPolicyName))})
 
 	if err := verifyResultWithDelay(func() (bool, error) {
 		cm, _ := channelz.GetTopChannels(0, 0)
@@ -935,12 +981,12 @@ func testHealthCheckSuccess(t *testing.T, e env) {
 // TestHealthCheckFailure invokes the unary Check() RPC on the health server
 // with an expired context and expects the RPC to fail.
 func (s) TestHealthCheckFailure(t *testing.T) {
-	for _, e := range listTestEnv() {
-		testHealthCheckFailure(t, e)
+	e := env{
+		name:     "tcp-tls",
+		network:  "tcp",
+		security: "tls",
+		balancer: healthCheckTestPolicyName,
 	}
-}
-
-func testHealthCheckFailure(t *testing.T, e env) {
 	te := newTest(t, e)
 	te.declareLogNoise(
 		"Failed to dial ",
@@ -1166,3 +1212,111 @@ func testHealthCheckServingStatus(t *testing.T, e env) {
 	te.setHealthServingStatus(defaultHealthService, healthpb.HealthCheckResponse_NOT_SERVING)
 	verifyHealthCheckStatus(t, 1*time.Second, cc, defaultHealthService, healthpb.HealthCheckResponse_NOT_SERVING)
 }
+
+// Test verifies that registering a nil health listener closes the health
+// client.
+func (s) TestHealthCheckUnregisterHealthListener(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	hcEnterChan, hcExitChan := setupHealthCheckWrapper(t)
+	scChan := make(chan balancer.SubConn, 1)
+	readyUpdateReceivedCh := make(chan struct{})
+	bf := stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			cc := bd.ClientConn
+			ccw := &subConnStoringCCWrapper{
+				ClientConn: cc,
+				scChan:     scChan,
+				stateListener: func(scs balancer.SubConnState) {
+					if scs.ConnectivityState != connectivity.Ready {
+						return
+					}
+					close(readyUpdateReceivedCh)
+				},
+			}
+			bd.Data = balancer.Get(pickfirst.Name).Build(ccw, bd.BuildOptions)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.Data.(balancer.Balancer).Close()
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
+		},
+	}
+
+	stub.Register(t.Name(), bf)
+	_, lis, ts := setupServer(t, nil)
+	ts.SetServingStatus("foo", healthpb.HealthCheckResponse_SERVING)
+
+	_, r := setupClient(t, nil)
+	svcCfg := fmt.Sprintf(`{
+		"healthCheckConfig": {
+			"serviceName": "foo"
+		},
+		"loadBalancingConfig": [{"%s":{}}]
+	}`, t.Name())
+	r.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: lis.Addr().String()}},
+		ServiceConfig: parseServiceConfig(t, r, svcCfg)})
+
+	var sc balancer.SubConn
+	select {
+	case sc = <-scChan:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for SubConn creation")
+	}
+
+	// Wait for the SubConn to enter READY.
+	select {
+	case <-readyUpdateReceivedCh:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for SubConn to enter READY")
+	}
+
+	// Health check should start only after a health listener is registered.
+	select {
+	case <-hcEnterChan:
+		t.Fatalf("Health service client created prematurely.")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	// Register a health listener and verify it receives updates.
+	healthChan := make(chan balancer.SubConnState, 1)
+	sc.RegisterHealthListener(func(scs balancer.SubConnState) {
+		healthChan <- scs
+	})
+
+	select {
+	case <-hcEnterChan:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for health check to begin.")
+	}
+
+	for readyReceived := false; !readyReceived; {
+		select {
+		case scs := <-healthChan:
+			t.Logf("Received health update: %v", scs)
+			readyReceived = scs.ConnectivityState == connectivity.Ready
+		case <-ctx.Done():
+			t.Fatalf("Context timed out waiting for healthy backend.")
+		}
+	}
+
+	// Registering a nil listener should invalidate the previously registered
+	// listener and close the health service client.
+	sc.RegisterHealthListener(nil)
+	select {
+	case <-hcExitChan:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for the health client to close.")
+	}
+
+	ts.SetServingStatus("foo", healthpb.HealthCheckResponse_NOT_SERVING)
+
+	// No updates should be received on the listener.
+	select {
+	case scs := <-healthChan:
+		t.Fatalf("Received unexpected health update on the listener: %v", scs)
+	case <-time.After(defaultTestShortTimeout):
+	}
+}
diff --git a/test/xds/xds_client_custom_lb_test.go b/test/xds/xds_client_custom_lb_test.go
index 8d87a897..2c881301 100644
--- a/test/xds/xds_client_custom_lb_test.go
+++ b/test/xds/xds_client_custom_lb_test.go
@@ -238,11 +238,11 @@ func (s) TestWrrLocality(t *testing.T) {
 					Host:        "localhost",
 					Localities: []e2e.LocalityOptions{
 						{
-							Backends: []e2e.BackendOptions{{Port: port1}, {Port: port2}},
+							Backends: []e2e.BackendOptions{{Ports: []uint32{port1}}, {Ports: []uint32{port2}}},
 							Weight:   1,
 						},
 						{
-							Backends: []e2e.BackendOptions{{Port: port3}, {Port: port4}, {Port: port5}},
+							Backends: []e2e.BackendOptions{{Ports: []uint32{port3}}, {Ports: []uint32{port4}}, {Ports: []uint32{port5}}},
 							Weight:   2,
 						},
 					},
diff --git a/test/xds/xds_client_priority_locality_test.go b/test/xds/xds_client_priority_locality_test.go
index 0f7c5e21..607e6ea8 100644
--- a/test/xds/xds_client_priority_locality_test.go
+++ b/test/xds/xds_client_priority_locality_test.go
@@ -95,14 +95,14 @@ func (s) TestClientSideXDS_LocalityChangesPriority(t *testing.T) {
 					Name:     "my-locality-1",
 					Weight:   1000000,
 					Priority: 0,
-					Backends: []e2e.BackendOptions{{Port: ports[0]}},
+					Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}},
 					Locality: locality1,
 				},
 				{
 					Name:     "my-locality-2",
 					Weight:   1000000,
 					Priority: 1,
-					Backends: []e2e.BackendOptions{{Port: ports[1]}},
+					Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 					Locality: locality2,
 				},
 			},
@@ -138,14 +138,14 @@ func (s) TestClientSideXDS_LocalityChangesPriority(t *testing.T) {
 				Name:     "my-locality-1",
 				Weight:   500000,
 				Priority: 0,
-				Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, backend0.Address)}},
+				Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, backend0.Address)}}},
 				Locality: locality1,
 			},
 			{
 				Name:     "my-locality-2",
 				Weight:   500000,
 				Priority: 0,
-				Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, backend1.Address)}},
+				Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, backend1.Address)}}},
 				Locality: locality2,
 			},
 		},
@@ -167,14 +167,14 @@ func (s) TestClientSideXDS_LocalityChangesPriority(t *testing.T) {
 				Name:     "my-locality-1",
 				Weight:   499884,
 				Priority: 0,
-				Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, backend0.Address)}},
+				Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, backend0.Address)}}},
 				Locality: locality1,
 			},
 			{
 				Name:     "my-locality-2",
 				Weight:   500115,
 				Priority: 0,
-				Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, backend1.Address)}},
+				Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, backend1.Address)}}},
 				Locality: locality2,
 			},
 		},
@@ -197,7 +197,7 @@ func (s) TestClientSideXDS_LocalityChangesPriority(t *testing.T) {
 				Name:     "my-locality-2",
 				Weight:   1000000,
 				Priority: 0,
-				Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, backend1.Address)}},
+				Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, backend1.Address)}}},
 				Locality: locality2,
 			},
 		},
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
index bd9cd557..9c17cee6 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"strings"
 	"testing"
 	"time"
@@ -202,11 +203,18 @@ func registerWrappedCDSPolicy(t *testing.T) chan balancer.Balancer {
 //   - a channel used to signal that previously requested cluster resources are
 //     no longer requested
 func setupWithManagementServer(t *testing.T) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient, chan []string, chan struct{}) {
+	return setupWithManagementServerAndListener(t, nil)
+}
+
+// Same as setupWithManagementServer, but also allows the caller to specify
+// a listener to be used by the management server.
+func setupWithManagementServerAndListener(t *testing.T, lis net.Listener) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient, chan []string, chan struct{}) {
 	t.Helper()
 
 	cdsResourceRequestedCh := make(chan []string, 1)
 	cdsResourceCanceledCh := make(chan struct{}, 1)
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener: lis,
 		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
 			if req.GetTypeUrl() == version.V3ClusterURL {
 				switch len(req.GetResourceNames()) {
@@ -807,11 +815,20 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 //     TRANSIENT_FAILURE. It is also expected to cancel the CDS watch.
 func (s) TestResolverError(t *testing.T) {
 	_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, r, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServer(t)
+	lis := testutils.NewListenerWrapper(t, nil)
+	mgmtServer, nodeID, cc, r, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServerAndListener(t, lis)
 
-	// Verify that the specified cluster resource is requested.
+	// Grab the wrapped connection from the listener wrapper. This will be used
+	// to verify the connection is closed.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	val, err := lis.NewConnCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Failed to receive new connection from wrapped listener: %v", err)
+	}
+	conn := val.(*testutils.ConnWrapper)
+
+	// Verify that the specified cluster resource is requested.
 	wantNames := []string{clusterName}
 	if err := waitForResourceNames(ctx, cdsResourceRequestedCh, wantNames); err != nil {
 		t.Fatal(err)
@@ -831,7 +848,7 @@ func (s) TestResolverError(t *testing.T) {
 
 	// Ensure that the resolver error is propagated to the RPC caller.
 	client := testgrpc.NewTestServiceClient(cc)
-	_, err := client.EmptyCall(ctx, &testpb.Empty{})
+	_, err = client.EmptyCall(ctx, &testpb.Empty{})
 	if code := status.Code(err); code != codes.Unavailable {
 		t.Fatalf("EmptyCall() failed with code: %v, want %v", code, codes.Unavailable)
 	}
@@ -901,11 +918,14 @@ func (s) TestResolverError(t *testing.T) {
 	resolverErr = xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")
 	r.ReportError(resolverErr)
 
-	// Wait for the CDS resource to be not requested anymore.
+	// Wait for the CDS resource to be not requested anymore, or the connection
+	// to the management server to be closed (which happens as part of the last
+	// resource watch being canceled).
 	select {
 	case <-ctx.Done():
 		t.Fatal("Timeout when waiting for CDS resource to be not requested")
 	case <-cdsResourceCanceledCh:
+	case <-conn.CloseCh.C:
 	}
 
 	// Verify that the resolver error is pushed to the child policy.
@@ -928,12 +948,12 @@ func (s) TestResolverError(t *testing.T) {
 	}
 }
 
-// Tests that closing the cds LB policy results in the cluster resource watch
-// being cancelled and the child policy being closed.
+// Tests that closing the cds LB policy results in the the child policy being
+// closed.
 func (s) TestClose(t *testing.T) {
 	cdsBalancerCh := registerWrappedCDSPolicy(t)
 	_, _, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _, _, cdsResourceCanceledCh := setupWithManagementServer(t)
+	mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
 
 	// Start a test service backend.
 	server := stubserver.StartTestService(t, nil)
@@ -967,12 +987,6 @@ func (s) TestClose(t *testing.T) {
 	}
 	cdsBal.Close()
 
-	// Wait for the CDS resource to be not requested anymore.
-	select {
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for CDS resource to be not requested")
-	case <-cdsResourceCanceledCh:
-	}
 	// Wait for the child policy to be closed.
 	select {
 	case <-ctx.Done():
diff --git a/xds/internal/balancer/clusterimpl/tests/balancer_test.go b/xds/internal/balancer/clusterimpl/tests/balancer_test.go
index f7d91d19..a8e7ac7a 100644
--- a/xds/internal/balancer/clusterimpl/tests/balancer_test.go
+++ b/xds/internal/balancer/clusterimpl/tests/balancer_test.go
@@ -141,7 +141,7 @@ func (s) TestConfigUpdateWithSameLoadReportingServerConfig(t *testing.T) {
 			Host:        "localhost",
 			Localities: []e2e.LocalityOptions{
 				{
-					Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, server.Address)}},
+					Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, server.Address)}}},
 					Weight:   1,
 				},
 			},
@@ -252,13 +252,13 @@ func (s) TestLoadReportingPickFirstMultiLocality(t *testing.T) {
 			Localities: []e2e.LocalityOptions{
 				{
 					Backends: []e2e.BackendOptions{
-						{Port: testutils.ParsePort(t, server1.Address)},
+						{Ports: []uint32{testutils.ParsePort(t, server1.Address)}},
 					},
 					Weight: 1,
 				},
 				{
 					Backends: []e2e.BackendOptions{
-						{Port: testutils.ParsePort(t, server2.Address)},
+						{Ports: []uint32{testutils.ParsePort(t, server2.Address)}},
 					},
 					Weight: 2,
 				},
diff --git a/xds/internal/balancer/clusterresolver/configbuilder_test.go b/xds/internal/balancer/clusterresolver/configbuilder_test.go
index 01055886..e7dc2b78 100644
--- a/xds/internal/balancer/clusterresolver/configbuilder_test.go
+++ b/xds/internal/balancer/clusterresolver/configbuilder_test.go
@@ -51,23 +51,23 @@ const (
 	testDropCategory    = "test-drops"
 	testDropOverMillion = 1
 
-	localityCount      = 5
-	addressPerLocality = 2
+	localityCount       = 5
+	endpointPerLocality = 2
 )
 
 var (
-	testLocalityIDs []internal.LocalityID
-	testAddressStrs [][]string
-	testEndpoints   [][]xdsresource.Endpoint
+	testLocalityIDs       []internal.LocalityID
+	testResolverEndpoints [][]resolver.Endpoint
+	testEndpoints         [][]xdsresource.Endpoint
 
 	testLocalitiesP0, testLocalitiesP1 []xdsresource.Locality
 
-	addrCmpOpts = cmp.Options{
+	endpointCmpOpts = cmp.Options{
 		cmp.AllowUnexported(attributes.Attributes{}),
-		cmp.Transformer("SortAddrs", func(in []resolver.Address) []resolver.Address {
-			out := append([]resolver.Address(nil), in...) // Copy input to avoid mutating it
+		cmp.Transformer("SortEndpoints", func(in []resolver.Endpoint) []resolver.Endpoint {
+			out := append([]resolver.Endpoint(nil), in...) // Copy input to avoid mutating it
 			sort.Slice(out, func(i, j int) bool {
-				return out[i].Addr < out[j].Addr
+				return out[i].Addresses[0].Addr < out[j].Addresses[0].Addr
 			})
 			return out
 		}),
@@ -85,18 +85,22 @@ func init() {
 	for i := 0; i < localityCount; i++ {
 		testLocalityIDs = append(testLocalityIDs, internal.LocalityID{Zone: fmt.Sprintf("test-zone-%d", i)})
 		var (
-			addrs []string
-			ends  []xdsresource.Endpoint
+			endpoints []resolver.Endpoint
+			ends      []xdsresource.Endpoint
 		)
-		for j := 0; j < addressPerLocality; j++ {
+		for j := 0; j < endpointPerLocality; j++ {
 			addr := fmt.Sprintf("addr-%d-%d", i, j)
-			addrs = append(addrs, addr)
+			endpoints = append(endpoints, resolver.Endpoint{Addresses: []resolver.Address{{Addr: addr}}})
 			ends = append(ends, xdsresource.Endpoint{
-				Address:      addr,
 				HealthStatus: xdsresource.EndpointHealthStatusHealthy,
+				Addresses: []string{
+					addr,
+					fmt.Sprintf("addr-%d-%d-additional-1", i, j),
+					fmt.Sprintf("addr-%d-%d-additional-2", i, j),
+				},
 			})
 		}
-		testAddressStrs = append(testAddressStrs, addrs)
+		testResolverEndpoints = append(testResolverEndpoints, endpoints)
 		testEndpoints = append(testEndpoints, ends)
 	}
 
@@ -171,7 +175,7 @@ func TestBuildPriorityConfigJSON(t *testing.T) {
 			mechanism: DiscoveryMechanism{
 				Type: DiscoveryMechanismTypeLogicalDNS,
 			},
-			addresses:    testAddressStrs[4],
+			endpoints:    testResolverEndpoints[4],
 			childNameGen: newNameGenerator(1),
 		},
 	}, nil)
@@ -226,7 +230,7 @@ func TestBuildPriorityConfig(t *testing.T) {
 				Type:             DiscoveryMechanismTypeLogicalDNS,
 				outlierDetection: noopODCfg,
 			},
-			addresses:    testAddressStrs[4],
+			endpoints:    testResolverEndpoints[4],
 			childNameGen: newNameGenerator(1),
 		},
 	}, nil)
@@ -301,7 +305,7 @@ func TestBuildPriorityConfig(t *testing.T) {
 }
 
 func TestBuildClusterImplConfigForDNS(t *testing.T) {
-	gotName, gotConfig, gotAddrs := buildClusterImplConfigForDNS(newNameGenerator(3), testAddressStrs[0], DiscoveryMechanism{Cluster: testClusterName2, Type: DiscoveryMechanismTypeLogicalDNS})
+	gotName, gotConfig, gotEndpoints := buildClusterImplConfigForDNS(newNameGenerator(3), testResolverEndpoints[0], DiscoveryMechanism{Cluster: testClusterName2, Type: DiscoveryMechanismTypeLogicalDNS})
 	wantName := "priority-3"
 	wantConfig := &clusterimpl.LBConfig{
 		Cluster: testClusterName2,
@@ -309,9 +313,11 @@ func TestBuildClusterImplConfigForDNS(t *testing.T) {
 			Name: "pick_first",
 		},
 	}
-	wantAddrs := []resolver.Address{
-		hierarchy.Set(resolver.Address{Addr: testAddressStrs[0][0]}, []string{"priority-3"}),
-		hierarchy.Set(resolver.Address{Addr: testAddressStrs[0][1]}, []string{"priority-3"}),
+	e1 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][0].Addresses[0]}}}
+	e2 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][1].Addresses[0]}}}
+	wantEndpoints := []resolver.Endpoint{
+		hierarchy.SetInEndpoint(e1, []string{"priority-3"}),
+		hierarchy.SetInEndpoint(e2, []string{"priority-3"}),
 	}
 
 	if diff := cmp.Diff(gotName, wantName); diff != "" {
@@ -320,7 +326,7 @@ func TestBuildClusterImplConfigForDNS(t *testing.T) {
 	if diff := cmp.Diff(gotConfig, wantConfig); diff != "" {
 		t.Errorf("buildClusterImplConfigForDNS() diff (-got +want) %v", diff)
 	}
-	if diff := cmp.Diff(gotAddrs, wantAddrs, addrCmpOpts); diff != "" {
+	if diff := cmp.Diff(gotEndpoints, wantEndpoints, endpointCmpOpts); diff != "" {
 		t.Errorf("buildClusterImplConfigForDNS() diff (-got +want) %v", diff)
 	}
 }
@@ -334,7 +340,7 @@ func TestBuildClusterImplConfigForEDS(t *testing.T) {
 		t.Fatalf("Failed to create LRS server config for testing: %v", err)
 	}
 
-	gotNames, gotConfigs, gotAddrs, _ := buildClusterImplConfigForEDS(
+	gotNames, gotConfigs, gotEndpoints, _ := buildClusterImplConfigForEDS(
 		newNameGenerator(2),
 		xdsresource.EndpointsUpdate{
 			Drops: []xdsresource.OverloadDropConfig{
@@ -408,15 +414,15 @@ func TestBuildClusterImplConfigForEDS(t *testing.T) {
 			},
 		},
 	}
-	wantAddrs := []resolver.Address{
-		testAddrWithAttrs(testAddressStrs[0][0], 20, 1, "priority-2-0", &testLocalityIDs[0]),
-		testAddrWithAttrs(testAddressStrs[0][1], 20, 1, "priority-2-0", &testLocalityIDs[0]),
-		testAddrWithAttrs(testAddressStrs[1][0], 80, 1, "priority-2-0", &testLocalityIDs[1]),
-		testAddrWithAttrs(testAddressStrs[1][1], 80, 1, "priority-2-0", &testLocalityIDs[1]),
-		testAddrWithAttrs(testAddressStrs[2][0], 20, 1, "priority-2-1", &testLocalityIDs[2]),
-		testAddrWithAttrs(testAddressStrs[2][1], 20, 1, "priority-2-1", &testLocalityIDs[2]),
-		testAddrWithAttrs(testAddressStrs[3][0], 80, 1, "priority-2-1", &testLocalityIDs[3]),
-		testAddrWithAttrs(testAddressStrs[3][1], 80, 1, "priority-2-1", &testLocalityIDs[3]),
+	wantEndpoints := []resolver.Endpoint{
+		testEndpointWithAttrs(testEndpoints[0][0].Addresses, 20, 1, "priority-2-0", &testLocalityIDs[0]),
+		testEndpointWithAttrs(testEndpoints[0][1].Addresses, 20, 1, "priority-2-0", &testLocalityIDs[0]),
+		testEndpointWithAttrs(testEndpoints[1][0].Addresses, 80, 1, "priority-2-0", &testLocalityIDs[1]),
+		testEndpointWithAttrs(testEndpoints[1][1].Addresses, 80, 1, "priority-2-0", &testLocalityIDs[1]),
+		testEndpointWithAttrs(testEndpoints[2][0].Addresses, 20, 1, "priority-2-1", &testLocalityIDs[2]),
+		testEndpointWithAttrs(testEndpoints[2][1].Addresses, 20, 1, "priority-2-1", &testLocalityIDs[2]),
+		testEndpointWithAttrs(testEndpoints[3][0].Addresses, 80, 1, "priority-2-1", &testLocalityIDs[3]),
+		testEndpointWithAttrs(testEndpoints[3][1].Addresses, 80, 1, "priority-2-1", &testLocalityIDs[3]),
 	}
 
 	if diff := cmp.Diff(gotNames, wantNames); diff != "" {
@@ -425,7 +431,7 @@ func TestBuildClusterImplConfigForEDS(t *testing.T) {
 	if diff := cmp.Diff(gotConfigs, wantConfigs); diff != "" {
 		t.Errorf("buildClusterImplConfigForEDS() diff (-got +want) %v", diff)
 	}
-	if diff := cmp.Diff(gotAddrs, wantAddrs, addrCmpOpts); diff != "" {
+	if diff := cmp.Diff(gotEndpoints, wantEndpoints, endpointCmpOpts); diff != "" {
 		t.Errorf("buildClusterImplConfigForEDS() diff (-got +want) %v", diff)
 	}
 
@@ -526,29 +532,29 @@ func TestDedupSortedIntSlice(t *testing.T) {
 
 func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
 	tests := []struct {
-		name         string
-		localities   []xdsresource.Locality
-		priorityName string
-		mechanism    DiscoveryMechanism
-		childPolicy  *iserviceconfig.BalancerConfig
-		wantConfig   *clusterimpl.LBConfig
-		wantAddrs    []resolver.Address
-		wantErr      bool
+		name          string
+		localities    []xdsresource.Locality
+		priorityName  string
+		mechanism     DiscoveryMechanism
+		childPolicy   *iserviceconfig.BalancerConfig
+		wantConfig    *clusterimpl.LBConfig
+		wantEndpoints []resolver.Endpoint
+		wantErr       bool
 	}{{
 		name: "round robin as child, no LRS",
 		localities: []xdsresource.Locality{
 			{
 				Endpoints: []xdsresource.Endpoint{
-					{Address: "addr-1-1", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
-					{Address: "addr-1-2", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
+					{Addresses: []string{"addr-1-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
+					{Addresses: []string{"addr-1-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
 				},
 				ID:     internal.LocalityID{Zone: "test-zone-1"},
 				Weight: 20,
 			},
 			{
 				Endpoints: []xdsresource.Endpoint{
-					{Address: "addr-2-1", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
-					{Address: "addr-2-2", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
+					{Addresses: []string{"addr-2-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
+					{Addresses: []string{"addr-2-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
 				},
 				ID:     internal.LocalityID{Zone: "test-zone-2"},
 				Weight: 80,
@@ -567,11 +573,11 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
 			EDSServiceName: testEDSService,
 			ChildPolicy:    &iserviceconfig.BalancerConfig{Name: roundrobin.Name},
 		},
-		wantAddrs: []resolver.Address{
-			testAddrWithAttrs("addr-1-1", 20, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
-			testAddrWithAttrs("addr-1-2", 20, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
-			testAddrWithAttrs("addr-2-1", 80, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
-			testAddrWithAttrs("addr-2-2", 80, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
+		wantEndpoints: []resolver.Endpoint{
+			testEndpointWithAttrs([]string{"addr-1-1"}, 20, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
+			testEndpointWithAttrs([]string{"addr-1-2"}, 20, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
+			testEndpointWithAttrs([]string{"addr-2-1"}, 80, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
+			testEndpointWithAttrs([]string{"addr-2-2"}, 80, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
 		},
 	},
 		{
@@ -579,16 +585,16 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
 			localities: []xdsresource.Locality{
 				{
 					Endpoints: []xdsresource.Endpoint{
-						{Address: "addr-1-1", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
-						{Address: "addr-1-2", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
+						{Addresses: []string{"addr-1-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
+						{Addresses: []string{"addr-1-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
 					},
 					ID:     internal.LocalityID{Zone: "test-zone-1"},
 					Weight: 20,
 				},
 				{
 					Endpoints: []xdsresource.Endpoint{
-						{Address: "addr-2-1", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
-						{Address: "addr-2-2", HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
+						{Addresses: []string{"addr-2-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90},
+						{Addresses: []string{"addr-2-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10},
 					},
 					ID:     internal.LocalityID{Zone: "test-zone-2"},
 					Weight: 80,
@@ -603,11 +609,11 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
 					Config: &ringhash.LBConfig{MinRingSize: 1, MaxRingSize: 2},
 				},
 			},
-			wantAddrs: []resolver.Address{
-				testAddrWithAttrs("addr-1-1", 20, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
-				testAddrWithAttrs("addr-1-2", 20, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
-				testAddrWithAttrs("addr-2-1", 80, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
-				testAddrWithAttrs("addr-2-2", 80, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
+			wantEndpoints: []resolver.Endpoint{
+				testEndpointWithAttrs([]string{"addr-1-1"}, 20, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
+				testEndpointWithAttrs([]string{"addr-1-2"}, 20, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-1"}),
+				testEndpointWithAttrs([]string{"addr-2-1"}, 80, 90, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
+				testEndpointWithAttrs([]string{"addr-2-2"}, 80, 10, "test-priority", &internal.LocalityID{Zone: "test-zone-2"}),
 			},
 		},
 	}
@@ -620,7 +626,7 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
 			if diff := cmp.Diff(got, tt.wantConfig); diff != "" {
 				t.Errorf("localitiesToWeightedTarget() diff (-got +want) %v", diff)
 			}
-			if diff := cmp.Diff(got1, tt.wantAddrs, cmp.AllowUnexported(attributes.Attributes{})); diff != "" {
+			if diff := cmp.Diff(got1, tt.wantEndpoints, cmp.AllowUnexported(attributes.Attributes{})); diff != "" {
 				t.Errorf("localitiesToWeightedTarget() diff (-got +want) %v", diff)
 			}
 		})
@@ -635,17 +641,20 @@ func assertString(f func() (string, error)) string {
 	return s
 }
 
-func testAddrWithAttrs(addrStr string, localityWeight, endpointWeight uint32, priority string, lID *internal.LocalityID) resolver.Address {
-	addr := resolver.Address{Addr: addrStr}
+func testEndpointWithAttrs(addrStrs []string, localityWeight, endpointWeight uint32, priority string, lID *internal.LocalityID) resolver.Endpoint {
+	endpoint := resolver.Endpoint{}
+	for _, a := range addrStrs {
+		endpoint.Addresses = append(endpoint.Addresses, resolver.Address{Addr: a})
+	}
 	path := []string{priority}
 	if lID != nil {
 		path = append(path, assertString(lID.ToString))
-		addr = internal.SetLocalityID(addr, *lID)
+		endpoint = internal.SetLocalityIDInEndpoint(endpoint, *lID)
 	}
-	addr = hierarchy.Set(addr, path)
-	addr = wrrlocality.SetAddrInfo(addr, wrrlocality.AddrInfo{LocalityWeight: localityWeight})
-	addr = weightedroundrobin.SetAddrInfo(addr, weightedroundrobin.AddrInfo{Weight: localityWeight * endpointWeight})
-	return addr
+	endpoint = hierarchy.SetInEndpoint(endpoint, path)
+	endpoint = wrrlocality.SetAddrInfoInEndpoint(endpoint, wrrlocality.AddrInfo{LocalityWeight: localityWeight})
+	endpoint = weightedroundrobin.SetAddrInfoInEndpoint(endpoint, weightedroundrobin.AddrInfo{Weight: localityWeight * endpointWeight})
+	return endpoint
 }
 
 func TestConvertClusterImplMapToOutlierDetection(t *testing.T) {
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go b/xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go
index af622458..89760f6f 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
 	"strings"
 	"testing"
 	"time"
@@ -27,9 +28,13 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
@@ -141,7 +146,7 @@ func (s) TestEDS_OneLocality(t *testing.T) {
 	resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: ports[0]}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}},
 	}})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -196,7 +201,7 @@ func (s) TestEDS_OneLocality(t *testing.T) {
 	resources = clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: ports[0]}, {Port: ports[1]}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}, {Ports: []uint32{ports[1]}}},
 	}})
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -210,7 +215,7 @@ func (s) TestEDS_OneLocality(t *testing.T) {
 	resources = clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: ports[1]}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 	}})
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -223,7 +228,7 @@ func (s) TestEDS_OneLocality(t *testing.T) {
 	resources = clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: ports[2]}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{ports[2]}}},
 	}})
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -270,12 +275,12 @@ func (s) TestEDS_MultipleLocalities(t *testing.T) {
 		{
 			Name:     localityName1,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[0]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}},
 		},
 		{
 			Name:     localityName2,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[1]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 		},
 	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -332,17 +337,17 @@ func (s) TestEDS_MultipleLocalities(t *testing.T) {
 		{
 			Name:     localityName1,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[0]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}},
 		},
 		{
 			Name:     localityName2,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[1]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 		},
 		{
 			Name:     localityName3,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[2]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[2]}}},
 		},
 	})
 	if err := managementServer.Update(ctx, resources); err != nil {
@@ -358,12 +363,12 @@ func (s) TestEDS_MultipleLocalities(t *testing.T) {
 		{
 			Name:     localityName2,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[1]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 		},
 		{
 			Name:     localityName3,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[2]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[2]}}},
 		},
 	})
 	if err := managementServer.Update(ctx, resources); err != nil {
@@ -380,12 +385,12 @@ func (s) TestEDS_MultipleLocalities(t *testing.T) {
 		{
 			Name:     localityName2,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[1]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[1]}}},
 		},
 		{
 			Name:     localityName3,
 			Weight:   1,
-			Backends: []e2e.BackendOptions{{Port: ports[2]}, {Port: ports[3]}},
+			Backends: []e2e.BackendOptions{{Ports: []uint32{ports[2]}}, {Ports: []uint32{ports[3]}}},
 		},
 	})
 	if err := managementServer.Update(ctx, resources); err != nil {
@@ -421,24 +426,24 @@ func (s) TestEDS_EndpointsHealth(t *testing.T) {
 			Name:   localityName1,
 			Weight: 1,
 			Backends: []e2e.BackendOptions{
-				{Port: ports[0], HealthStatus: v3corepb.HealthStatus_UNKNOWN},
-				{Port: ports[1], HealthStatus: v3corepb.HealthStatus_HEALTHY},
-				{Port: ports[2], HealthStatus: v3corepb.HealthStatus_UNHEALTHY},
-				{Port: ports[3], HealthStatus: v3corepb.HealthStatus_DRAINING},
-				{Port: ports[4], HealthStatus: v3corepb.HealthStatus_TIMEOUT},
-				{Port: ports[5], HealthStatus: v3corepb.HealthStatus_DEGRADED},
+				{Ports: []uint32{ports[0]}, HealthStatus: v3corepb.HealthStatus_UNKNOWN},
+				{Ports: []uint32{ports[1]}, HealthStatus: v3corepb.HealthStatus_HEALTHY},
+				{Ports: []uint32{ports[2]}, HealthStatus: v3corepb.HealthStatus_UNHEALTHY},
+				{Ports: []uint32{ports[3]}, HealthStatus: v3corepb.HealthStatus_DRAINING},
+				{Ports: []uint32{ports[4]}, HealthStatus: v3corepb.HealthStatus_TIMEOUT},
+				{Ports: []uint32{ports[5]}, HealthStatus: v3corepb.HealthStatus_DEGRADED},
 			},
 		},
 		{
 			Name:   localityName2,
 			Weight: 1,
 			Backends: []e2e.BackendOptions{
-				{Port: ports[6], HealthStatus: v3corepb.HealthStatus_UNKNOWN},
-				{Port: ports[7], HealthStatus: v3corepb.HealthStatus_HEALTHY},
-				{Port: ports[8], HealthStatus: v3corepb.HealthStatus_UNHEALTHY},
-				{Port: ports[9], HealthStatus: v3corepb.HealthStatus_DRAINING},
-				{Port: ports[10], HealthStatus: v3corepb.HealthStatus_TIMEOUT},
-				{Port: ports[11], HealthStatus: v3corepb.HealthStatus_DEGRADED},
+				{Ports: []uint32{ports[6]}, HealthStatus: v3corepb.HealthStatus_UNKNOWN},
+				{Ports: []uint32{ports[7]}, HealthStatus: v3corepb.HealthStatus_HEALTHY},
+				{Ports: []uint32{ports[8]}, HealthStatus: v3corepb.HealthStatus_UNHEALTHY},
+				{Ports: []uint32{ports[9]}, HealthStatus: v3corepb.HealthStatus_DRAINING},
+				{Ports: []uint32{ports[10]}, HealthStatus: v3corepb.HealthStatus_TIMEOUT},
+				{Ports: []uint32{ports[11]}, HealthStatus: v3corepb.HealthStatus_DEGRADED},
 			},
 		},
 	})
@@ -567,7 +572,7 @@ func (s) TestEDS_EmptyUpdate(t *testing.T) {
 	resources = clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: ports[0]}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{ports[0]}}},
 	}})
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -911,7 +916,7 @@ func (s) TestEDS_BadUpdateWithoutPreviousGoodUpdate(t *testing.T) {
 	resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, server.Address)}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, server.Address)}}},
 	}})
 	resources.Endpoints[0].Endpoints[0].LbEndpoints[0].LoadBalancingWeight = &wrapperspb.UInt32Value{Value: 0}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -984,7 +989,7 @@ func (s) TestEDS_BadUpdateWithPreviousGoodUpdate(t *testing.T) {
 	resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
 		Name:     localityName1,
 		Weight:   1,
-		Backends: []e2e.BackendOptions{{Port: testutils.ParsePort(t, server.Address)}},
+		Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, server.Address)}}},
 	}})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -1138,3 +1143,182 @@ func waitForProducedZeroAddressesError(ctx context.Context, t *testing.T, client
 	}
 	return errors.New("timeout when waiting for RPCs to fail with UNAVAILABLE status and produced zero addresses")
 }
+
+// Test runs a server which listens on multiple ports. The test updates xds resouce
+// cache to contain a single endpoint with multiple addresses. The test intercepts
+// the resolver updates sent to the petiole policy and verifies that the
+// additional endpoint addresses are correctly propagated.
+func (s) TestEDS_EndpointWithMultipleAddresses(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start a backend server which listens to multiple ports to simulate a
+	// backend with multiple addresses.
+	server := &stubserver.StubServer{
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
+		UnaryCallF: func(context.Context, *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+			return &testpb.SimpleResponse{}, nil
+		},
+	}
+	lis1, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer lis1.Close()
+	lis2, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer lis2.Close()
+	lis3, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer lis3.Close()
+
+	server.Listener = lis1
+	if err := server.StartServer(); err != nil {
+		t.Fatalf("Failed to start stub server: %v", err)
+	}
+	go server.S.Serve(lis2)
+	go server.S.Serve(lis3)
+
+	t.Logf("Started test service backend at addresses %q, %q, %q", lis1.Addr(), lis2.Addr(), lis3.Addr())
+
+	ports := []uint32{
+		testutils.ParsePort(t, lis1.Addr().String()),
+		testutils.ParsePort(t, lis2.Addr().String()),
+		testutils.ParsePort(t, lis3.Addr().String()),
+	}
+
+	testCases := []struct {
+		name                      string
+		dualstackEndpointsEnabled bool
+		wantEndpointPorts         []uint32
+		wantAddrPorts             []uint32
+	}{
+		{
+			name:                      "flag_enabled",
+			dualstackEndpointsEnabled: true,
+			wantEndpointPorts:         ports,
+			wantAddrPorts:             ports[:1],
+		},
+		{
+			name:              "flag_disabled",
+			wantEndpointPorts: ports[:1],
+			wantAddrPorts:     ports[:1],
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
+			defer func() {
+				envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
+			}()
+			envconfig.XDSDualstackEndpointsEnabled = tc.dualstackEndpointsEnabled
+
+			// Wrap the round robin balancer to intercept resolver updates.
+			originalRRBuilder := balancer.Get(roundrobin.Name)
+			defer func() {
+				balancer.Register(originalRRBuilder)
+			}()
+			resolverUpdateCh := make(chan resolver.State, 1)
+			stub.Register(roundrobin.Name, stub.BalancerFuncs{
+				Init: func(bd *stub.BalancerData) {
+					bd.Data = originalRRBuilder.Build(bd.ClientConn, bd.BuildOptions)
+				},
+				Close: func(bd *stub.BalancerData) {
+					bd.Data.(balancer.Balancer).Close()
+				},
+				UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+					resolverUpdateCh <- ccs.ResolverState
+					return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
+				},
+			})
+
+			// Spin up a management server to receive xDS resources from.
+			mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+			// Create bootstrap configuration pointing to the above management server.
+			nodeID := uuid.New().String()
+			bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+			// Create xDS resources for consumption by the test. We start off with a
+			// single backend in a single EDS locality.
+			resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
+				Name:   localityName1,
+				Weight: 1,
+				Backends: []e2e.BackendOptions{{
+					Ports: ports,
+				}},
+			}})
+			if err := mgmtServer.Update(ctx, resources); err != nil {
+				t.Fatal(err)
+			}
+
+			// Create an xDS client talking to the above management server, configured
+			// with a short watch expiry timeout.
+			xdsClient, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
+				Name:     t.Name(),
+				Contents: bootstrapContents,
+			})
+			if err != nil {
+				t.Fatalf("Failed to create an xDS client: %v", err)
+			}
+			defer close()
+
+			// Create a manual resolver and push a service config specifying the use of
+			// the cluster_resolver LB policy with a single discovery mechanism.
+			r := manual.NewBuilderWithScheme("whatever")
+			jsonSC := fmt.Sprintf(`{
+				"loadBalancingConfig":[{
+					"cluster_resolver_experimental":{
+						"discoveryMechanisms": [{
+							"cluster": "%s",
+							"type": "EDS",
+							"edsServiceName": "%s",
+							"outlierDetection": {}
+						}],
+						"xdsLbPolicy":[{"round_robin":{}}]
+					}
+				}]
+			}`, clusterName, edsServiceName)
+			scpr := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(jsonSC)
+			r.InitialState(xdsclient.SetClient(resolver.State{ServiceConfig: scpr}, xdsClient))
+
+			cc, err := grpc.NewClient(r.Scheme()+":///test.service", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
+			if err != nil {
+				t.Fatalf("failed to create new client for local test server: %v", err)
+			}
+			defer cc.Close()
+			client := testgrpc.NewTestServiceClient(cc)
+			if err := rrutil.CheckRoundRobinRPCs(ctx, client, []resolver.Address{{Addr: lis1.Addr().String()}}); err != nil {
+				t.Fatal(err)
+			}
+
+			var rs resolver.State
+			select {
+			case rs = <-resolverUpdateCh:
+			case <-ctx.Done():
+				t.Fatalf("Context timed out waiting for resolver update.")
+			}
+
+			gotEndpointPorts := []uint32{}
+			for _, a := range rs.Endpoints[0].Addresses {
+				gotEndpointPorts = append(gotEndpointPorts, testutils.ParsePort(t, a.Addr))
+			}
+			if diff := cmp.Diff(gotEndpointPorts, tc.wantEndpointPorts); diff != "" {
+				t.Errorf("Unexpected endpoint address ports in resolver update, diff (-got +want): %v", diff)
+			}
+
+			gotAddrPorts := []uint32{}
+			for _, a := range rs.Addresses {
+				gotAddrPorts = append(gotAddrPorts, testutils.ParsePort(t, a.Addr))
+			}
+			if diff := cmp.Diff(gotAddrPorts, tc.wantAddrPorts); diff != "" {
+				t.Errorf("Unexpected address ports in resolver update, diff (-got +want): %v", diff)
+			}
+		})
+	}
+}
diff --git a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
index 94efa99d..afc60924 100644
--- a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
+++ b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
@@ -182,7 +182,7 @@ func backendOptions(t *testing.T, serverAddrs []string) []e2e.BackendOptions {
 
 	var backendOpts []e2e.BackendOptions
 	for _, addr := range serverAddrs {
-		backendOpts = append(backendOpts, e2e.BackendOptions{Port: testutils.ParsePort(t, addr)})
+		backendOpts = append(backendOpts, e2e.BackendOptions{Ports: []uint32{testutils.ParsePort(t, addr)}})
 	}
 	return backendOpts
 }
@@ -872,9 +872,9 @@ func (s) TestRingHash_EndpointWeights(t *testing.T) {
 
 	const clusterName = "cluster"
 	backendOpts := []e2e.BackendOptions{
-		{Port: testutils.ParsePort(t, backends[0])},
-		{Port: testutils.ParsePort(t, backends[1])},
-		{Port: testutils.ParsePort(t, backends[2]), Weight: 2},
+		{Ports: []uint32{testutils.ParsePort(t, backends[0])}},
+		{Ports: []uint32{testutils.ParsePort(t, backends[1])}},
+		{Ports: []uint32{testutils.ParsePort(t, backends[2])}, Weight: 2},
 	}
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
@@ -1209,14 +1209,14 @@ func (s) TestRingHash_RandomHashingDistributionAccordingToLocalityAndEndpointWei
 		Localities: []e2e.LocalityOptions{
 			{
 				Backends: []e2e.BackendOptions{{
-					Port:   testutils.ParsePort(t, backends[0]),
+					Ports:  []uint32{testutils.ParsePort(t, backends[0])},
 					Weight: endpoint1Weight,
 				}},
 				Weight: locality1Weight,
 			},
 			{
 				Backends: []e2e.BackendOptions{{
-					Port:   testutils.ParsePort(t, backends[1]),
+					Ports:  []uint32{testutils.ParsePort(t, backends[1])},
 					Weight: endpoint2Weight,
 				}},
 				Weight: locality2Weight,
diff --git a/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go b/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go
index feebc1ad..af16b6a5 100644
--- a/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go
+++ b/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go
@@ -337,9 +337,10 @@ func (s) TestADS_NACK_InvalidFirstResponse(t *testing.T) {
 //  1. A resource is requested and a good response is received. The test verifies
 //     that an ACK is sent for this resource.
 //  2. The previously requested resource is no longer requested. The test
-//     verifies that a request with no resource names is sent out.
-//  3. The same resource is requested again. The test verifies that the request
-//     is sent with the previously ACKed version.
+//     verifies that the connection to the management server is closed.
+//  3. The same resource is requested again. The test verifies that a new
+//     request is sent with an empty version string, which corresponds to the
+//     first request on a new connection.
 func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -349,6 +350,7 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 	// the test goroutine to verify ACK version and nonce.
 	streamRequestCh := testutils.NewChannel()
 	streamResponseCh := testutils.NewChannel()
+	streamCloseCh := testutils.NewChannel()
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
 		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
 			streamRequestCh.SendContext(ctx, req)
@@ -357,6 +359,9 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 		OnStreamResponse: func(_ context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, resp *v3discoverypb.DiscoveryResponse) {
 			streamResponseCh.SendContext(ctx, resp)
 		},
+		OnStreamClosed: func(int64, *v3corepb.Node) {
+			streamCloseCh.SendContext(ctx, struct{}{})
+		},
 	})
 
 	// Create a listener resource on the management server.
@@ -425,9 +430,10 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 		t.Fatal("Timeout when waiting for ACK")
 	}
 	gotReq = r.(*v3discoverypb.DiscoveryRequest)
-	wantReq.VersionInfo = gotResp.GetVersionInfo()
-	wantReq.ResponseNonce = gotResp.GetNonce()
-	if diff := cmp.Diff(gotReq, wantReq, protocmp.Transform()); diff != "" {
+	wantACKReq := proto.Clone(wantReq).(*v3discoverypb.DiscoveryRequest)
+	wantACKReq.VersionInfo = gotResp.GetVersionInfo()
+	wantACKReq.ResponseNonce = gotResp.GetNonce()
+	if diff := cmp.Diff(gotReq, wantACKReq, protocmp.Transform()); diff != "" {
 		t.Fatalf("Unexpected diff in received discovery request, diff (-got, +want):\n%s", diff)
 	}
 
@@ -442,39 +448,41 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Cancel the watch on the listener resource. This should result in a
-	// discovery request with no resource names.
+	// Cancel the watch on the listener resource. This should result in the
+	// existing connection to be management server getting closed.
 	ldsCancel()
-
-	// Verify that the discovery request matches expectation.
-	r, err = streamRequestCh.Receive(ctx)
-	if err != nil {
-		t.Fatal("Timeout when waiting for discovery request")
-	}
-	gotReq = r.(*v3discoverypb.DiscoveryRequest)
-	wantReq.ResourceNames = nil
-	if diff := cmp.Diff(gotReq, wantReq, protocmp.Transform()); diff != "" {
-		t.Fatalf("Unexpected diff in received discovery request, diff (-got, +want):\n%s", diff)
+	if _, err := streamCloseCh.Receive(ctx); err != nil {
+		t.Fatalf("Timeout when expecting existing connection to be closed: %v", err)
 	}
 
+	// There is a race between two events when the last watch on an xdsChannel
+	// is canceled:
+	// - an empty discovery request being sent out
+	// - the ADS stream being closed
+	// To handle this race, we drain the request channel here so that if an
+	// empty discovery request was received, it is pulled out of the request
+	// channel and thereby guaranteeing a clean slate for the next watch
+	// registered below.
+	streamRequestCh.Drain()
+
 	// Register a watch for the same listener resource.
 	lw = newListenerWatcher()
 	ldsCancel = xdsresource.WatchListener(client, listenerName, lw)
 	defer ldsCancel()
 
-	// Verify that the discovery request contains the version from the
-	// previously received response.
+	// Verify that the discovery request is identical to the first one sent out
+	// to the management server.
 	r, err = streamRequestCh.Receive(ctx)
 	if err != nil {
 		t.Fatal("Timeout when waiting for discovery request")
 	}
 	gotReq = r.(*v3discoverypb.DiscoveryRequest)
-	wantReq.ResourceNames = []string{listenerName}
 	if diff := cmp.Diff(gotReq, wantReq, protocmp.Transform()); diff != "" {
 		t.Fatalf("Unexpected diff in received discovery request, diff (-got, +want):\n%s", diff)
 	}
 
-	// TODO(https://github.com/envoyproxy/go-control-plane/issues/1002): Once
-	// this bug is fixed, we need to verify that the update is received by the
-	// watcher.
+	// Verify the update received by the watcher.
+	if err := verifyListenerUpdate(ctx, lw.updateCh, wantUpdate); err != nil {
+		t.Fatal(err)
+	}
 }
diff --git a/xds/internal/xdsclient/tests/ads_stream_restart_test.go b/xds/internal/xdsclient/tests/ads_stream_restart_test.go
index f0da932f..7e15218e 100644
--- a/xds/internal/xdsclient/tests/ads_stream_restart_test.go
+++ b/xds/internal/xdsclient/tests/ads_stream_restart_test.go
@@ -124,6 +124,7 @@ func (s) TestADS_ResourcesAreRequestedAfterStreamRestart(t *testing.T) {
 	// Register a watch for a listener resource.
 	lw := newListenerWatcher()
 	ldsCancel := xdsresource.WatchListener(client, listenerName, lw)
+	defer ldsCancel()
 
 	// Verify that an ADS stream is opened and an LDS request with the above
 	// resource name is sent.
@@ -147,52 +148,6 @@ func (s) TestADS_ResourcesAreRequestedAfterStreamRestart(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Cancel the watch for the above listener resource, and verify that an LDS
-	// request with no resource names is sent.
-	ldsCancel()
-	if err := waitForResourceNames(ctx, t, ldsResourcesCh, []string{}); err != nil {
-		t.Fatal(err)
-	}
-
-	// Stop the restartable listener and wait for the stream to close.
-	lis.Stop()
-	select {
-	case <-streamClosed:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for ADS stream to close")
-	}
-
-	// Restart the restartable listener and wait for the stream to open.
-	lis.Restart()
-	select {
-	case <-streamOpened:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for ADS stream to open")
-	}
-
-	// Wait for a short duration and verify that no LDS request is sent, since
-	// there are no resources being watched.
-	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	select {
-	case <-sCtx.Done():
-	case names := <-ldsResourcesCh:
-		t.Fatalf("LDS request sent for resource names %v, when expecting no request", names)
-	}
-
-	// Register another watch for the same listener resource, and verify that an
-	// LDS request with the above resource name is sent.
-	ldsCancel = xdsresource.WatchListener(client, listenerName, lw)
-	if err := waitForResourceNames(ctx, t, ldsResourcesCh, []string{listenerName}); err != nil {
-		t.Fatal(err)
-	}
-	defer ldsCancel()
-
-	// Verify the update received by the watcher.
-	if err := verifyListenerUpdate(ctx, lw.updateCh, wantListenerUpdate); err != nil {
-		t.Fatal(err)
-	}
-
 	// Create a cluster resource on the management server, in addition to the
 	// existing listener resource.
 	const clusterName = "cluster"
@@ -255,7 +210,7 @@ func (s) TestADS_ResourcesAreRequestedAfterStreamRestart(t *testing.T) {
 
 	// Wait for a short duration and verify that no CDS request is sent, since
 	// there are no resources being watched.
-	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout)
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 	defer sCancel()
 	select {
 	case <-sCtx.Done():
diff --git a/xds/internal/xdsclient/tests/authority_test.go b/xds/internal/xdsclient/tests/authority_test.go
index 1947daf6..3570be59 100644
--- a/xds/internal/xdsclient/tests/authority_test.go
+++ b/xds/internal/xdsclient/tests/authority_test.go
@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/google/uuid"
 	"google.golang.org/grpc/internal/testutils"
@@ -65,7 +64,7 @@ var (
 //
 // Returns two listeners used by the default and non-default management servers
 // respectively, and the xDS client and its close function.
-func setupForAuthorityTests(ctx context.Context, t *testing.T, idleTimeout time.Duration) (*testutils.ListenerWrapper, *testutils.ListenerWrapper, xdsclient.XDSClient, func()) {
+func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.ListenerWrapper, *testutils.ListenerWrapper, xdsclient.XDSClient, func()) {
 	// Create listener wrappers which notify on to a channel whenever a new
 	// connection is accepted. We use this to track the number of transports
 	// used by the xDS client.
@@ -102,10 +101,9 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T, idleTimeout time.
 		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
 	client, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
-		Name:                     t.Name(),
-		Contents:                 bootstrapContents,
-		WatchExpiryTimeout:       defaultTestWatchExpiryTimeout,
-		IdleChannelExpiryTimeout: idleTimeout,
+		Name:               t.Name(),
+		Contents:           bootstrapContents,
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	})
 	if err != nil {
 		t.Fatalf("Failed to create an xDS client: %v", err)
@@ -137,7 +135,7 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T, idleTimeout time.
 func (s) TestAuthority_XDSChannelSharing(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	lis, _, client, close := setupForAuthorityTests(ctx, t, time.Duration(0))
+	lis, _, client, close := setupForAuthorityTests(ctx, t)
 	defer close()
 
 	// Verify that no connection is established to the management server at this
@@ -176,13 +174,12 @@ func (s) TestAuthority_XDSChannelSharing(t *testing.T) {
 	}
 }
 
-// Test the xdsChannel idle timeout logic. The test verifies that the xDS client
-// does not close xdsChannels immediately after the last watch is canceled, but
-// waits for the configured idle timeout to expire before closing them.
-func (s) TestAuthority_XDSChannelIdleTimeout(t *testing.T) {
+// Test the xdsChannel close logic. The test verifies that the xDS client
+// closes an xdsChannel immediately after the last watch is canceled.
+func (s) TestAuthority_XDSChannelClose(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	lis, _, client, close := setupForAuthorityTests(ctx, t, defaultTestIdleChannelExpiryTimeout)
+	lis, _, client, close := setupForAuthorityTests(ctx, t)
 	defer close()
 
 	// Request the first resource. Verify that a new transport is created.
@@ -203,110 +200,10 @@ func (s) TestAuthority_XDSChannelIdleTimeout(t *testing.T) {
 	}
 
 	// Cancel both watches, and verify that the connection to the management
-	// server is not closed immediately.
+	// server is closed.
 	cdsCancel1()
 	cdsCancel2()
-	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	if _, err := conn.CloseCh.Receive(sCtx); err != context.DeadlineExceeded {
-		t.Fatal("Connection to management server closed unexpectedly")
-	}
-
-	// Ensure the transport is closed once the idle timeout fires.
-	select {
-	case <-conn.CloseCh.C:
-	case <-time.After(2 * defaultTestIdleChannelExpiryTimeout):
-		t.Fatal("Connection to management server not closed after idle timeout expiry")
-	}
-}
-
-// Tests that xdsChannels in use and in the idle cache are all closed when the
-// xDS client is closed.
-func (s) TestAuthority_XDSChannelCloseOnClientClose(t *testing.T) {
-	// Set the idle timeout to twice the defaultTestTimeout. This will ensure
-	// that idle channels stay in the cache for the duration of this test, until
-	// explicitly closed.
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	lisDefault, lisNonDefault, client, close := setupForAuthorityTests(ctx, t, time.Duration(2*defaultTestTimeout))
-
-	// Request the first resource. Verify that a new transport is created to the
-	// default management server.
-	watcher := noopClusterWatcher{}
-	cdsCancel1 := xdsresource.WatchCluster(client, authorityTestResourceName11, watcher)
-	val, err := lisDefault.NewConnCh.Receive(ctx)
-	if err != nil {
-		t.Fatalf("Timed out when waiting for a new transport to be created to the management server: %v", err)
-	}
-	connDefault := val.(*testutils.ConnWrapper)
-
-	// Request another resource which is served by the non-default authority.
-	// Verify that a new transport is created to the non-default management
-	// server.
-	xdsresource.WatchCluster(client, authorityTestResourceName3, watcher)
-	val, err = lisNonDefault.NewConnCh.Receive(ctx)
-	if err != nil {
-		t.Fatalf("Timed out when waiting for a new transport to be created to the management server: %v", err)
-	}
-	connNonDefault := val.(*testutils.ConnWrapper)
-
-	// Cancel the first watch. This should move the default authority to the
-	// idle cache, but the connection should not be closed yet, because the idle
-	// timeout would not have fired.
-	cdsCancel1()
-	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	if _, err := connDefault.CloseCh.Receive(sCtx); err != context.DeadlineExceeded {
-		t.Fatal("Connection to management server closed unexpectedly")
-	}
-
-	// Closing the xDS client should close the connection to both management
-	// servers, even though we have an open watch to one of them.
-	close()
-	if _, err := connDefault.CloseCh.Receive(ctx); err != nil {
-		t.Fatal("Connection to management server not closed after client close")
-	}
-	if _, err := connNonDefault.CloseCh.Receive(ctx); err != nil {
-		t.Fatal("Connection to management server not closed after client close")
-	}
-}
-
-// Tests that an xdsChannel in the idle cache is revived when a new watch is
-// started on an authority.
-func (s) TestAuthority_XDSChannelRevive(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	lis, _, client, close := setupForAuthorityTests(ctx, t, defaultTestIdleChannelExpiryTimeout)
-	defer close()
-
-	// Request the first resource. Verify that a new transport is created.
-	watcher := noopClusterWatcher{}
-	cdsCancel1 := xdsresource.WatchCluster(client, authorityTestResourceName11, watcher)
-	val, err := lis.NewConnCh.Receive(ctx)
-	if err != nil {
-		t.Fatalf("Timed out when waiting for a new transport to be created to the management server: %v", err)
-	}
-	conn := val.(*testutils.ConnWrapper)
-
-	// Cancel the above watch. This should move the authority to the idle cache.
-	cdsCancel1()
-
-	// Request the second resource. Verify that no new transport is created.
-	// This should move the authority out of the idle cache.
-	cdsCancel2 := xdsresource.WatchCluster(client, authorityTestResourceName12, watcher)
-	defer cdsCancel2()
-	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	if _, err := lis.NewConnCh.Receive(sCtx); err != context.DeadlineExceeded {
-		t.Fatal("Unexpected new transport created to management server")
-	}
-
-	// Wait for double the idle timeout, and the connection to the management
-	// server should not be closed, since it was revived from the idle cache.
-	time.Sleep(2 * defaultTestIdleChannelExpiryTimeout)
-	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	if _, err := conn.CloseCh.Receive(sCtx); err != context.DeadlineExceeded {
-		t.Fatal("Connection to management server closed unexpectedly")
+	if _, err := conn.CloseCh.Receive(ctx); err != nil {
+		t.Fatal("Timeout when waiting for connection to management server to be closed")
 	}
 }
diff --git a/xds/internal/xdsclient/tests/eds_watchers_test.go b/xds/internal/xdsclient/tests/eds_watchers_test.go
index ef27178f..21021b89 100644
--- a/xds/internal/xdsclient/tests/eds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/eds_watchers_test.go
@@ -176,7 +176,7 @@ func (s) TestEDSWatch(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -199,7 +199,7 @@ func (s) TestEDSWatch(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -335,7 +335,7 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -351,7 +351,7 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost2, edsPort2), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost2, edsPort2)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -373,7 +373,7 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -389,7 +389,7 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 				update: xdsresource.EndpointsUpdate{
 					Localities: []xdsresource.Locality{
 						{
-							Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost2, edsPort2), Weight: 1}},
+							Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost2, edsPort2)}, Weight: 1}},
 							ID: internal.LocalityID{
 								Region:  "region-1",
 								Zone:    "zone-1",
@@ -581,7 +581,7 @@ func (s) TestEDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) {
 		update: xdsresource.EndpointsUpdate{
 			Localities: []xdsresource.Locality{
 				{
-					Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+					Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 					ID: internal.LocalityID{
 						Region:  "region-1",
 						Zone:    "zone-1",
@@ -669,7 +669,7 @@ func (s) TestEDSWatch_ResourceCaching(t *testing.T) {
 		update: xdsresource.EndpointsUpdate{
 			Localities: []xdsresource.Locality{
 				{
-					Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+					Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 					ID: internal.LocalityID{
 						Region:  "region-1",
 						Zone:    "zone-1",
@@ -795,7 +795,7 @@ func (s) TestEDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
 		update: xdsresource.EndpointsUpdate{
 			Localities: []xdsresource.Locality{
 				{
-					Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+					Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 					ID: internal.LocalityID{
 						Region:  "region-1",
 						Zone:    "zone-1",
@@ -952,7 +952,7 @@ func (s) TestEDSWatch_PartialValid(t *testing.T) {
 		update: xdsresource.EndpointsUpdate{
 			Localities: []xdsresource.Locality{
 				{
-					Endpoints: []xdsresource.Endpoint{{Address: fmt.Sprintf("%s:%d", edsHost1, edsPort1), Weight: 1}},
+					Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}},
 					ID: internal.LocalityID{
 						Region:  "region-1",
 						Zone:    "zone-1",
diff --git a/xds/internal/xdsclient/tests/fallback_test.go b/xds/internal/xdsclient/tests/fallback_test.go
index 514945f8..18c382db 100644
--- a/xds/internal/xdsclient/tests/fallback_test.go
+++ b/xds/internal/xdsclient/tests/fallback_test.go
@@ -160,13 +160,10 @@ func (s) TestFallback_OnStartup(t *testing.T) {
 		t.Fatalf("Failed to create bootstrap file: %v", err)
 	}
 
-	// Create an xDS client with the above bootstrap configuration and a short
-	// idle channel expiry timeout. This ensures that connections to lower
-	// priority servers get closed quickly, for the test to verify.
+	// Create an xDS client with the above bootstrap configuration.
 	xdsC, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
-		Name:                     t.Name(),
-		Contents:                 bootstrapContents,
-		IdleChannelExpiryTimeout: defaultTestIdleChannelExpiryTimeout,
+		Name:     t.Name(),
+		Contents: bootstrapContents,
 	})
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
@@ -363,13 +360,10 @@ func (s) TestFallback_MidUpdate(t *testing.T) {
 		t.Fatalf("Failed to create bootstrap file: %v", err)
 	}
 
-	// Create an xDS client with the above bootstrap configuration and a short
-	// idle channel expiry timeout. This ensures that connections to lower
-	// priority servers get closed quickly, for the test to verify.
+	// Create an xDS client with the above bootstrap configuration.
 	xdsC, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
-		Name:                     t.Name(),
-		Contents:                 bootstrapContents,
-		IdleChannelExpiryTimeout: defaultTestIdleChannelExpiryTimeout,
+		Name:     t.Name(),
+		Contents: bootstrapContents,
 	})
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
@@ -556,13 +550,10 @@ func (s) TestFallback_MidStartup(t *testing.T) {
 		t.Fatalf("Failed to create bootstrap file: %v", err)
 	}
 
-	// Create an xDS client with the above bootstrap configuration and a short
-	// idle channel expiry timeout. This ensures that connections to lower
-	// priority servers get closed quickly, for the test to verify.
+	// Create an xDS client with the above bootstrap configuration.
 	xdsC, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
-		Name:                     t.Name(),
-		Contents:                 bootstrapContents,
-		IdleChannelExpiryTimeout: defaultTestIdleChannelExpiryTimeout,
+		Name:     t.Name(),
+		Contents: bootstrapContents,
 	})
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
diff --git a/xds/internal/xdsclient/tests/federation_watchers_test.go b/xds/internal/xdsclient/tests/federation_watchers_test.go
index 78f69518..65233c34 100644
--- a/xds/internal/xdsclient/tests/federation_watchers_test.go
+++ b/xds/internal/xdsclient/tests/federation_watchers_test.go
@@ -287,7 +287,7 @@ func (s) TestFederation_EndpointsResourceContextParamOrder(t *testing.T) {
 		update: xdsresource.EndpointsUpdate{
 			Localities: []xdsresource.Locality{
 				{
-					Endpoints: []xdsresource.Endpoint{{Address: "localhost:666", Weight: 1}},
+					Endpoints: []xdsresource.Endpoint{{Addresses: []string{"localhost:666"}, Weight: 1}},
 					Weight:    1,
 					ID: internal.LocalityID{
 						Region:  "region-1",
diff --git a/xds/internal/xdsclient/tests/helpers_test.go b/xds/internal/xdsclient/tests/helpers_test.go
index 5c4175b0..ab060d53 100644
--- a/xds/internal/xdsclient/tests/helpers_test.go
+++ b/xds/internal/xdsclient/tests/helpers_test.go
@@ -36,10 +36,9 @@ func Test(t *testing.T) {
 }
 
 const (
-	defaultTestWatchExpiryTimeout       = 500 * time.Millisecond
-	defaultTestIdleChannelExpiryTimeout = 50 * time.Millisecond
-	defaultTestTimeout                  = 10 * time.Second
-	defaultTestShortTimeout             = 10 * time.Millisecond // For events expected to *not* happen.
+	defaultTestWatchExpiryTimeout = 500 * time.Millisecond
+	defaultTestTimeout            = 10 * time.Second
+	defaultTestShortTimeout       = 10 * time.Millisecond // For events expected to *not* happen.
 
 	ldsName         = "xdsclient-test-lds-resource"
 	rdsName         = "xdsclient-test-rds-resource"
diff --git a/xds/internal/xdsclient/tests/lds_watchers_test.go b/xds/internal/xdsclient/tests/lds_watchers_test.go
index 7b49b9b1..b05b9caf 100644
--- a/xds/internal/xdsclient/tests/lds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/lds_watchers_test.go
@@ -1029,7 +1029,6 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
 	}
 
-	// Verify that the expected error is propagated to the watcher.
 	// Verify that the expected error is propagated to the existing watcher.
 	if err := verifyUnknownListenerError(ctx, lw.updateCh, wantListenerNACKErr); err != nil {
 		t.Fatal(err)
@@ -1039,45 +1038,18 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 	lw2 := newListenerWatcher()
 	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
 	defer ldsCancel2()
-	// Verify that the expected error is propagated to the existing watcher.
 	if err := verifyUnknownListenerError(ctx, lw2.updateCh, wantListenerNACKErr); err != nil {
 		t.Fatal(err)
 	}
 }
 
-// TestLDSWatch_ResourceCaching_WithNACKError covers the case where a watch is
-// registered for a resource which is already present in the cache with an old
-// good update as well as latest NACK error. The test verifies that new watcher
-// receives both good update and error without a new resource request being
-// sent to the management server.
-func TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
-	firstRequestReceived := false
-	firstAckReceived := grpcsync.NewEvent()
-	secondAckReceived := grpcsync.NewEvent()
-	secondRequestReceived := grpcsync.NewEvent()
-
-	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
-		OnStreamRequest: func(id int64, req *v3discoverypb.DiscoveryRequest) error {
-			// The first request has an empty version string.
-			if !firstRequestReceived && req.GetVersionInfo() == "" {
-				firstRequestReceived = true
-				return nil
-			}
-			// The first ack has a non-empty version string.
-			if !firstAckReceived.HasFired() && req.GetVersionInfo() != "" {
-				firstAckReceived.Fire()
-				return nil
-			}
-			// The second ack has a non-empty version string.
-			if !secondAckReceived.HasFired() && req.GetVersionInfo() != "" {
-				secondAckReceived.Fire()
-				return nil
-			}
-			// Any requests after the first request and two acks, are not expected.
-			secondRequestReceived.Fire()
-			return nil
-		},
-	})
+// Tests the scenario where a watch registered for a resource results in a good
+// update followed by a bad update. This results in the resource cache
+// containing both the old good update and the latest NACK error. The test
+// verifies that a when a new watch is registered for the same resource, the new
+// watcher receives the good update followed by the NACK error.
+func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
@@ -1150,16 +1122,6 @@ func TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	if err := verifyUnknownListenerError(ctx, lw2.updateCh, wantListenerNACKErr); err != nil {
 		t.Fatal(err)
 	}
-
-	// No request should get sent out as part of this watch.
-	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	select {
-	case <-sCtx.Done():
-	case <-secondRequestReceived.Done():
-		t.Fatal("xdsClient sent out request instead of using update from cache")
-	default:
-	}
 }
 
 // TestLDSWatch_PartialValid covers the case where a response from the
diff --git a/xds/internal/xdsclient/tests/resource_update_test.go b/xds/internal/xdsclient/tests/resource_update_test.go
index b493c820..0460385d 100644
--- a/xds/internal/xdsclient/tests/resource_update_test.go
+++ b/xds/internal/xdsclient/tests/resource_update_test.go
@@ -1077,13 +1077,13 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) {
 			wantUpdate: xdsresource.EndpointsUpdate{
 				Localities: []xdsresource.Locality{
 					{
-						Endpoints: []xdsresource.Endpoint{{Address: "addr1:314", Weight: 1}},
+						Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr1:314"}, Weight: 1}},
 						ID:        internal.LocalityID{SubZone: "locality-1"},
 						Priority:  1,
 						Weight:    1,
 					},
 					{
-						Endpoints: []xdsresource.Endpoint{{Address: "addr2:159", Weight: 1}},
+						Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr2:159"}, Weight: 1}},
 						ID:        internal.LocalityID{SubZone: "locality-2"},
 						Priority:  0,
 						Weight:    1,
@@ -1111,13 +1111,13 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) {
 			wantUpdate: xdsresource.EndpointsUpdate{
 				Localities: []xdsresource.Locality{
 					{
-						Endpoints: []xdsresource.Endpoint{{Address: "addr1:314", Weight: 1}},
+						Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr1:314"}, Weight: 1}},
 						ID:        internal.LocalityID{SubZone: "locality-1"},
 						Priority:  1,
 						Weight:    1,
 					},
 					{
-						Endpoints: []xdsresource.Endpoint{{Address: "addr2:159", Weight: 1}},
+						Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr2:159"}, Weight: 1}},
 						ID:        internal.LocalityID{SubZone: "locality-2"},
 						Priority:  0,
 						Weight:    1,
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go b/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go
index f2419eac..e8df0c3c 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go
@@ -28,6 +28,8 @@ import (
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	v3typepb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/pretty"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/xds/internal"
@@ -47,8 +49,8 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "missing-priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 0, []string{"addr1:314"}, nil)
-				clab0.addLocality("locality-2", 1, 2, []string{"addr2:159"}, nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil)
+				clab0.addLocality("locality-2", 1, 2, []endpointOpts{{addrWithPort: "addr2:159"}}, nil)
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -58,7 +60,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "missing-locality-ID",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("", 1, 0, []string{"addr1:314"}, nil)
+				clab0.addLocality("", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil)
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -68,7 +70,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "zero-endpoint-weight",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-0", 1, 0, []string{"addr1:314"}, &addLocalityOptions{Weight: []uint32{0}})
+				clab0.addLocality("locality-0", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{Weight: []uint32{0}})
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -78,8 +80,8 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "duplicate-locality-in-the-same-priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-0", 1, 0, []string{"addr1:314"}, nil)
-				clab0.addLocality("locality-0", 1, 0, []string{"addr1:314"}, nil) // Duplicate locality with the same priority.
+				clab0.addLocality("locality-0", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil)
+				clab0.addLocality("locality-0", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil) // Duplicate locality with the same priority.
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -89,10 +91,10 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "missing locality weight",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 0, 1, []string{"addr1:314"}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 0, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY},
 				})
-				clab0.addLocality("locality-2", 0, 0, []string{"addr2:159"}, &addLocalityOptions{
+				clab0.addLocality("locality-2", 0, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY},
 				})
 				return clab0.Build()
@@ -103,9 +105,9 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "max sum of weights at the same priority exceeded",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 0, []string{"addr1:314"}, nil)
-				clab0.addLocality("locality-2", 4294967295, 1, []string{"addr2:159"}, nil)
-				clab0.addLocality("locality-3", 1, 1, []string{"addr2:88"}, nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil)
+				clab0.addLocality("locality-2", 4294967295, 1, []endpointOpts{{addrWithPort: "addr2:159"}}, nil)
+				clab0.addLocality("locality-3", 1, 1, []endpointOpts{{addrWithPort: "addr2:88"}}, nil)
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -115,8 +117,8 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "duplicate endpoint address",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 1, []string{"addr:997"}, nil)
-				clab0.addLocality("locality-2", 1, 0, []string{"addr:997"}, nil)
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr:997"}}, nil)
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr:997"}}, nil)
 				return clab0.Build()
 			}(),
 			want:    EndpointsUpdate{},
@@ -126,11 +128,11 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "good",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 1, []string{"addr1:314"}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 					Weight: []uint32{271},
 				})
-				clab0.addLocality("locality-2", 1, 0, []string{"addr2:159"}, &addLocalityOptions{
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 					Weight: []uint32{828},
 				})
@@ -141,7 +143,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 				Localities: []Locality{
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr1:314",
+							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
 						}},
@@ -151,7 +153,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 					},
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr2:159",
+							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
 						}},
@@ -167,12 +169,12 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "good duplicate locality with different priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 1, []string{"addr1:314"}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 					Weight: []uint32{271},
 				})
 				// Same locality name, but with different priority.
-				clab0.addLocality("locality-1", 1, 0, []string{"addr2:159"}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 					Weight: []uint32{828},
 				})
@@ -183,7 +185,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 				Localities: []Locality{
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr1:314",
+							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
 						}},
@@ -193,7 +195,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 					},
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr2:159",
+							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
 						}},
@@ -213,7 +215,118 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
-			if d := cmp.Diff(got, tt.want); d != "" {
+			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
+				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
+			}
+		})
+	}
+}
+
+func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) {
+	origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
+	defer func() {
+		envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
+	}()
+	envconfig.XDSDualstackEndpointsEnabled = true
+
+	tests := []struct {
+		name    string
+		m       *v3endpointpb.ClusterLoadAssignment
+		want    EndpointsUpdate
+		wantErr bool
+	}{
+		{
+			name: "duplicate primary address in self additional addresses",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr:998", additionalAddrWithPorts: []string{"addr:998"}}}, nil)
+				return clab0.Build()
+			}(),
+			want:    EndpointsUpdate{},
+			wantErr: true,
+		},
+		{
+			name: "duplicate primary address in other locality additional addresses",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr:997"}}, nil)
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr:998", additionalAddrWithPorts: []string{"addr:997"}}}, nil)
+				return clab0.Build()
+			}(),
+			want:    EndpointsUpdate{},
+			wantErr: true,
+		},
+		{
+			name: "duplicate additional address in self additional addresses",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr:998", additionalAddrWithPorts: []string{"addr:999", "addr:999"}}}, nil)
+				return clab0.Build()
+			}(),
+			want:    EndpointsUpdate{},
+			wantErr: true,
+		},
+		{
+			name: "duplicate additional address in other locality additional addresses",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr:997", additionalAddrWithPorts: []string{"addr:1000"}}}, nil)
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr:998", additionalAddrWithPorts: []string{"addr:1000"}}}, nil)
+				return clab0.Build()
+			}(),
+			want:    EndpointsUpdate{},
+			wantErr: true,
+		},
+		{
+			name: "multiple localities",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:997", additionalAddrWithPorts: []string{"addr1:1000"}}}, &addLocalityOptions{
+					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
+					Weight: []uint32{271},
+				})
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:998", additionalAddrWithPorts: []string{"addr2:1000"}}}, &addLocalityOptions{
+					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY},
+					Weight: []uint32{828},
+				})
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Drops: nil,
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:997", "addr1:1000"},
+							HealthStatus: EndpointHealthStatusUnhealthy,
+							Weight:       271,
+						}},
+						ID:       internal.LocalityID{SubZone: "locality-1"},
+						Priority: 1,
+						Weight:   1,
+					},
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr2:998", "addr2:1000"},
+							HealthStatus: EndpointHealthStatusHealthy,
+							Weight:       828,
+						}},
+						ID:       internal.LocalityID{SubZone: "locality-2"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := parseEDSRespProto(tt.m)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
 				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
 			}
 		})
@@ -223,11 +336,11 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 func (s) TestUnmarshalEndpoints(t *testing.T) {
 	var v3EndpointsAny = testutils.MarshalAny(t, func() *v3endpointpb.ClusterLoadAssignment {
 		clab0 := newClaBuilder("test", nil)
-		clab0.addLocality("locality-1", 1, 1, []string{"addr1:314"}, &addLocalityOptions{
+		clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
 			Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 			Weight: []uint32{271},
 		})
-		clab0.addLocality("locality-2", 1, 0, []string{"addr2:159"}, &addLocalityOptions{
+		clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
 			Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 			Weight: []uint32{828},
 		})
@@ -258,8 +371,8 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 			name: "bad endpoints resource",
 			resource: testutils.MarshalAny(t, func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 0, []string{"addr1:314"}, nil)
-				clab0.addLocality("locality-2", 1, 2, []string{"addr2:159"}, nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, nil)
+				clab0.addLocality("locality-2", 1, 2, []endpointOpts{{addrWithPort: "addr2:159"}}, nil)
 				return clab0.Build()
 			}()),
 			wantName: "test",
@@ -274,7 +387,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 				Localities: []Locality{
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr1:314",
+							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
 						}},
@@ -284,7 +397,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 					},
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr2:159",
+							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
 						}},
@@ -305,7 +418,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 				Localities: []Locality{
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr1:314",
+							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
 						}},
@@ -315,7 +428,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 					},
 					{
 						Endpoints: []Endpoint{{
-							Address:      "addr2:159",
+							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
 						}},
@@ -379,29 +492,49 @@ type addLocalityOptions struct {
 	Weight []uint32
 }
 
+type endpointOpts struct {
+	addrWithPort            string
+	additionalAddrWithPorts []string
+}
+
+func addressFromStr(addrWithPort string) *v3corepb.Address {
+	host, portStr, err := net.SplitHostPort(addrWithPort)
+	if err != nil {
+		panic("failed to split " + addrWithPort)
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		panic("failed to atoi " + portStr)
+	}
+
+	return &v3corepb.Address{
+		Address: &v3corepb.Address_SocketAddress{
+			SocketAddress: &v3corepb.SocketAddress{
+				Protocol:      v3corepb.SocketAddress_TCP,
+				Address:       host,
+				PortSpecifier: &v3corepb.SocketAddress_PortValue{PortValue: uint32(port)},
+			},
+		},
+	}
+}
+
 // addLocality adds a locality to the builder.
-func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint32, addrsWithPort []string, opts *addLocalityOptions) {
+func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint32, endpoints []endpointOpts, opts *addLocalityOptions) {
 	var lbEndPoints []*v3endpointpb.LbEndpoint
-	for i, a := range addrsWithPort {
-		host, portStr, err := net.SplitHostPort(a)
-		if err != nil {
-			panic("failed to split " + a)
-		}
-		port, err := strconv.Atoi(portStr)
-		if err != nil {
-			panic("failed to atoi " + portStr)
+	for i, e := range endpoints {
+		var additionalAddrs []*v3endpointpb.Endpoint_AdditionalAddress
+		for _, a := range e.additionalAddrWithPorts {
+			additionalAddrs = append(additionalAddrs, &v3endpointpb.Endpoint_AdditionalAddress{
+				Address: addressFromStr(a),
+			})
 		}
-
 		lbe := &v3endpointpb.LbEndpoint{
 			HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{
 				Endpoint: &v3endpointpb.Endpoint{
-					Address: &v3corepb.Address{
-						Address: &v3corepb.Address_SocketAddress{
-							SocketAddress: &v3corepb.SocketAddress{
-								Protocol: v3corepb.SocketAddress_TCP,
-								Address:  host,
-								PortSpecifier: &v3corepb.SocketAddress_PortValue{
-									PortValue: uint32(port)}}}}}},
+					Address:             addressFromStr(e.addrWithPort),
+					AdditionalAddresses: additionalAddrs,
+				},
+			},
 		}
 		if opts != nil {
 			if i < len(opts.Health) {