Repository: grpc/grpc-go

Issue:
xdsclient: ads stream flow control can block forever, resulting in resource-not-found errors
Here is a sequence of events that can lead to the ADS stream level flow control blocking forever.
- T-1: Listener resource is subscribed to and the request has been sent out
- T0: `recv()` receives the Listener resource from the wire: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L494
- T1: `recv()` sets the `pending` bit of the flow control to `true` and invokes the response handler and passes it the `onDone` callback: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L512
- T2: Response handler runs, and as part of handling the update, it subscribes to an RouteConfiguration resource. This results in `subscribe()` being called on the ADS stream, which queues the request: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L187
- T3: Response handler invokes the `onDone` callback to release flow control. This writes to the `readyCh` to unblock goroutines waiting for flow control. It hasn't yet set the `pending` bit to `false`: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L768
- T4: Meanwhile, the `send()` goroutine gets to run and processes the request for the RouteConfiguration. It calls `sendNew()` to send this request out: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L292
- T5: `sendNew()` checks the `pending` bit of the flow control. This is not yet set to `false` by the `onDone` callback. It will try to buffer this request, but before that happens, it loses CPU: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L321
- T6: Meanwhile `recv()` is in the next iteration of the `for` loop and has gotten unblocked on the call to `fc.wait()`: https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L486
- T7: `recv()` attempts to send out any buffered requests by calling `sendBuffered`, but that method does not find any buffered requests, because `sendNew()` hasn't yet written to the `bufferedRequests` channel. https://github.com/grpc/grpc-go/blob/e35080456c7071b6d55887080ef78809c163640c/internal/xds/clients/xdsclient/ads_stream.go#L371
- T8: `sendNew()` now writes to the `bufferedRequests` channel. 
- Anytime after T5: the `onDone` callback sets the `pending` bit to `false`.


But this request (buffered at T8) never gets sent out, because `recv()` is blocked waiting for some response from the management server, but no response is expected because the ADS stream has not requested any new resource.This will eventually lead to the RDS resource watch timer expiring, and being reported to the watcher as a resource-not-found error.


Discussion:

Solution patch:
diff --git a/balancer/pickfirst/pickfirst.go b/balancer/pickfirst/pickfirst.go
index ea889981..b15c10e4 100644
--- a/balancer/pickfirst/pickfirst.go
+++ b/balancer/pickfirst/pickfirst.go
@@ -169,7 +169,7 @@ func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState
 		addrs = state.ResolverState.Addresses
 		if cfg.ShuffleAddressList {
 			addrs = append([]resolver.Address{}, addrs...)
-			rand.Shuffle(len(addrs), func(i, j int) { addrs[i], addrs[j] = addrs[j], addrs[i] })
+			internal.RandShuffle(len(addrs), func(i, j int) { addrs[i], addrs[j] = addrs[j], addrs[i] })
 		}
 	}
 
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
index 67f315a0..9ffdd28a 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
@@ -283,7 +283,7 @@ func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState
 		newAddrs = state.ResolverState.Addresses
 		if cfg.ShuffleAddressList {
 			newAddrs = append([]resolver.Address{}, newAddrs...)
-			internal.RandShuffle(len(endpoints), func(i, j int) { endpoints[i], endpoints[j] = endpoints[j], endpoints[i] })
+			internal.RandShuffle(len(newAddrs), func(i, j int) { newAddrs[i], newAddrs[j] = newAddrs[j], newAddrs[i] })
 		}
 	}
 
@@ -351,6 +351,13 @@ func (b *pickfirstBalancer) ExitIdle() {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	if b.state == connectivity.Idle {
+		// Move the balancer into CONNECTING state immediately. This is done to
+		// avoid staying in IDLE if a resolver update arrives before the first
+		// SubConn reports CONNECTING.
+		b.updateBalancerState(balancer.State{
+			ConnectivityState: connectivity.Connecting,
+			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
+		})
 		b.startFirstPassLocked()
 	}
 }
@@ -604,7 +611,7 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 		if !b.addressList.seekTo(sd.addr) {
 			// This should not fail as we should have only one SubConn after
 			// entering READY. The SubConn should be present in the addressList.
-			b.logger.Errorf("Address %q not found address list in  %v", sd.addr, b.addressList.addresses)
+			b.logger.Errorf("Address %q not found address list in %v", sd.addr, b.addressList.addresses)
 			return
 		}
 		if !b.healthCheckingEnabled {
diff --git a/examples/features/health/README.md b/examples/features/health/README.md
index 69ee3813..1b8f0e89 100644
--- a/examples/features/health/README.md
+++ b/examples/features/health/README.md
@@ -25,15 +25,14 @@ Clients have two ways to monitor a servers health.
 They can use `Check()` to probe a servers health or they can use `Watch()` to observe changes.
 
 In most cases, clients do not need to directly check backend servers.
-Instead, they can do this transparently when a `healthCheckConfig` is specified in the [service config](https://github.com/grpc/proposal/blob/master/A17-client-side-health-checking.md#service-config-changes).
-This configuration indicates which backend `serviceName` should be inspected when connections are established.
+Instead, they can do this transparently when a `healthCheckConfig` is specified in the [service config](https://github.com/grpc/proposal/blob/master/A17-client-side-health-checking.md#service-config-changes) and the [health package](https://pkg.go.dev/google.golang.org/grpc/health) is imported.
+The `serviceName` in `healthCheckConfig` will be used in the health check when connections are established.
 An empty string (`""`) typically indicates the overall health of a server should be reported.
 
 ```go
 // import grpc/health to enable transparent client side checking
 import _ "google.golang.org/grpc/health"
 
-// set up appropriate service config
 serviceConfig := grpc.WithDefaultServiceConfig(`{
   "loadBalancingPolicy": "round_robin",
   "healthCheckConfig": {
diff --git a/examples/features/health/client/main.go b/examples/features/health/client/main.go
index dc5382a4..f7c97aaa 100644
--- a/examples/features/health/client/main.go
+++ b/examples/features/health/client/main.go
@@ -30,12 +30,12 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	pb "google.golang.org/grpc/examples/features/proto/echo"
-	_ "google.golang.org/grpc/health"
+	_ "google.golang.org/grpc/health" // REQUIRED to enable health checks
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 )
 
-var serviceConfig = `{
+const serviceConfig = `{
 	"loadBalancingPolicy": "round_robin",
 	"healthCheckConfig": {
 		"serviceName": ""
@@ -68,8 +68,8 @@ func main() {
 
 	options := []grpc.DialOption{
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithBlock(),
 		grpc.WithResolvers(r),
+		// google.golang.org/grpc/health must also be imported
 		grpc.WithDefaultServiceConfig(serviceConfig),
 	}
 
diff --git a/examples/features/interceptor/README.md b/examples/features/interceptor/README.md
index 68e6a965..bb9dff71 100644
--- a/examples/features/interceptor/README.md
+++ b/examples/features/interceptor/README.md
@@ -30,11 +30,14 @@ types of interceptors in total.
 
 The type for client-side unary interceptors is
 [`UnaryClientInterceptor`](https://godoc.org/google.golang.org/grpc#UnaryClientInterceptor).
-It is essentially a function type with signature: `func(ctx context.Context,
-method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker,
-opts ...CallOption) error`. Unary interceptor implementations can usually be
-divided into three parts: pre-processing, invoking the RPC method, and
-post-processing.
+It is a function type with signature:
+
+```golang
+func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error
+```
+
+Unary interceptor implementations can usually be divided into three parts:
+pre-processing, invoking the RPC method, and post-processing.
 
 For pre-processing, users can get info about the current RPC call by examining
 the args passed in. The args include the RPC context, method string, request to
@@ -59,22 +62,26 @@ To install a unary interceptor on a ClientConn, configure `Dial` with the
 
 The type for client-side stream interceptors is
 [`StreamClientInterceptor`](https://godoc.org/google.golang.org/grpc#StreamClientInterceptor).
-It is a function type with signature: `func(ctx context.Context, desc
-*StreamDesc, cc *ClientConn, method string, streamer Streamer, opts
-...CallOption) (ClientStream, error)`. An implementation of a stream interceptor
-usually includes pre-processing, and stream operation interception.
+It is a function type with signature:
+
+```golang
+func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error)
+```
+
+An implementation of a stream interceptor usually includes pre-processing, and
+stream operation interception.
 
 The pre-processing is similar to unary interceptors.
 
 However, rather than invoking the RPC method followed by post-processing, stream
 interceptors intercept the users' operations on the stream. The interceptor
 first calls the passed-in `streamer` to get a `ClientStream`, and then wraps the
-`ClientStream` while overloading its methods with the interception logic.
+`ClientStream` while overriding its methods with the interception logic.
 Finally, the interceptor returns the wrapped `ClientStream` to user to operate
 on.
 
 In the example, we define a new struct `wrappedStream`, which embeds a
-`ClientStream`. We then implement (overload) the `SendMsg` and `RecvMsg` methods
+`ClientStream`. We then implement (override) the `SendMsg` and `RecvMsg` methods
 on `wrappedStream` to intercept these two operations on the embedded
 `ClientStream`. In the example, we log the message type info and time info for
 interception purpose.
@@ -92,9 +99,11 @@ different information provided as args.
 
 The type for server-side unary interceptors is
 [`UnaryServerInterceptor`](https://godoc.org/google.golang.org/grpc#UnaryServerInterceptor).
-It is a function type with signature: `func(ctx context.Context, req
-interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err
-error)`.
+It is a function type with signature:
+
+```golang
+func(ctx context.Context, req interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)
+```
 
 Refer to the client-side unary interceptor section for a detailed implementation
 and explanation.
@@ -107,8 +116,11 @@ To install a unary interceptor on a Server, configure `NewServer` with the
 
 The type for server-side stream interceptors is
 [`StreamServerInterceptor`](https://godoc.org/google.golang.org/grpc#StreamServerInterceptor).
-It is a function type with the signature: `func(srv interface{}, ss
-ServerStream, info *StreamServerInfo, handler StreamHandler) error`.
+It is a function type with the signature:
+
+```golang
+func(srv interface{}, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error
+```
 
 Refer to the client-side stream interceptor section for a detailed
 implementation and explanation.
@@ -116,4 +128,3 @@ implementation and explanation.
 To install a stream interceptor on a Server, configure `NewServer` with the
 [`StreamInterceptor`](https://godoc.org/google.golang.org/grpc#StreamInterceptor)
 `ServerOption`.
-
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index 7cb23879..e06771c3 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -1002,6 +1002,9 @@ func (t *http2Client) closeStream(s *ClientStream, err error, rst bool, rstCode
 // accessed anymore.
 func (t *http2Client) Close(err error) {
 	t.conn.SetWriteDeadline(time.Now().Add(time.Second * 10))
+	// For background on the deadline value chosen here, see
+	// https://github.com/grpc/grpc-go/issues/8425#issuecomment-3057938248 .
+	t.conn.SetReadDeadline(time.Now().Add(time.Second))
 	t.mu.Lock()
 	// Make sure we only close once.
 	if t.state == closing {
diff --git a/internal/xds/balancer/clusterimpl/clusterimpl.go b/internal/xds/balancer/clusterimpl/clusterimpl.go
index ffe0a3db..72d143a3 100644
--- a/internal/xds/balancer/clusterimpl/clusterimpl.go
+++ b/internal/xds/balancer/clusterimpl/clusterimpl.go
@@ -451,7 +451,10 @@ func (b *clusterImplBalancer) NewSubConn(addrs []resolver.Address, opts balancer
 		lID := xdsinternal.GetLocalityID(addr)
 		if (lID == clients.Locality{}) {
 			if b.logger.V(2) {
-				b.logger.Infof("Locality ID for %s unexpectedly empty", addr)
+				// TODO: After A74, we should have the entire CDS config,
+				// allowing us to verify if this is indeed a Logical DNS cluster
+				// and avoid logging the message below.
+				b.logger.Infof("No Locality ID found for address %s. This is normal if %q is a Logical DNS cluster.", addr, clusterName)
 			}
 			return
 		}
diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go
index 83f8a5df..3ad62ac1 100644
--- a/internal/xds/clients/xdsclient/ads_stream.go
+++ b/internal/xds/clients/xdsclient/ads_stream.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"google.golang.org/grpc/grpclog"
@@ -103,11 +102,11 @@ type adsStreamImpl struct {
 	requestCh    *buffer.Unbounded   // Subscriptions and unsubscriptions are pushed here.
 	runnerDoneCh chan struct{}       // Notify completion of runner goroutine.
 	cancel       context.CancelFunc  // To cancel the context passed to the runner goroutine.
+	fc           *adsFlowControl     // Flow control for ADS stream.
 
 	// Guards access to the below fields (and to the contents of the map).
 	mu                sync.Mutex
 	resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state.
-	fc                *adsFlowControl                     // Flow control for ADS stream.
 	firstRequest      bool                                // False after the first request is sent out.
 }
 
@@ -135,6 +134,7 @@ func newADSStreamImpl(opts adsStreamOpts) *adsStreamImpl {
 		streamCh:          make(chan clients.Stream, 1),
 		requestCh:         buffer.NewUnbounded(),
 		runnerDoneCh:      make(chan struct{}),
+		fc:                newADSFlowControl(),
 		resourceTypeState: make(map[ResourceType]*resourceTypeState),
 	}
 
@@ -150,6 +150,7 @@ func newADSStreamImpl(opts adsStreamOpts) *adsStreamImpl {
 // Stop blocks until the stream is closed and all spawned goroutines exit.
 func (s *adsStreamImpl) Stop() {
 	s.cancel()
+	s.fc.stop()
 	s.requestCh.Close()
 	<-s.runnerDoneCh
 	s.logger.Infof("Shutdown ADS stream")
@@ -240,9 +241,6 @@ func (s *adsStreamImpl) runner(ctx context.Context) {
 		}
 
 		s.mu.Lock()
-		// Flow control is a property of the underlying streaming RPC call and
-		// needs to be initialized everytime a new one is created.
-		s.fc = newADSFlowControl(s.logger)
 		s.firstRequest = true
 		s.mu.Unlock()
 
@@ -256,7 +254,7 @@ func (s *adsStreamImpl) runner(ctx context.Context) {
 
 		// Backoff state is reset upon successful receipt of at least one
 		// message from the server.
-		if s.recv(ctx, stream) {
+		if s.recv(stream) {
 			return backoff.ErrResetBackoff
 		}
 		return nil
@@ -318,11 +316,13 @@ func (s *adsStreamImpl) sendNew(stream clients.Stream, typ ResourceType) error {
 	// This allows us to batch writes for requests which are generated as part
 	// of local processing of a received response.
 	state := s.resourceTypeState[typ]
-	if s.fc.pending.Load() {
+	bufferRequest := func() {
 		select {
 		case state.bufferedRequests <- struct{}{}:
 		default:
 		}
+	}
+	if s.fc.runIfPending(bufferRequest) {
 		return nil
 	}
 
@@ -458,7 +458,7 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string,
 	if s.logger.V(perRPCVerbosityLevel) {
 		s.logger.Infof("ADS request sent: %v", pretty.ToJSON(req))
 	} else if s.logger.V(2) {
-		s.logger.Warningf("ADS request sent for type %q, resources: %v, version: %q, nonce: %q", url, names, version, nonce)
+		s.logger.Infof("ADS request sent for type %q, resources: %v, version: %q, nonce: %q", url, names, version, nonce)
 	}
 
 	return nil
@@ -477,18 +477,19 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string,
 //
 // It returns a boolean indicating whether at least one message was received
 // from the server.
-func (s *adsStreamImpl) recv(ctx context.Context, stream clients.Stream) bool {
+func (s *adsStreamImpl) recv(stream clients.Stream) bool {
 	msgReceived := false
 	for {
-		// Wait for ADS stream level flow control to be available, and send out
-		// a request if anything was buffered while we were waiting for local
-		// processing of the previous response to complete.
-		if !s.fc.wait(ctx) {
+		// Wait for ADS stream level flow control to be available.
+		if s.fc.wait() {
 			if s.logger.V(2) {
-				s.logger.Infof("ADS stream context canceled")
+				s.logger.Infof("ADS stream stopped while waiting for flow control")
 			}
 			return msgReceived
 		}
+
+		// Send out a request if anything was buffered while we were waiting for
+		// local processing of the previous response to complete.
 		s.sendBuffered(stream)
 
 		resources, url, version, nonce, err := s.recvMessage(stream)
@@ -508,8 +509,8 @@ func (s *adsStreamImpl) recv(ctx context.Context, stream clients.Stream) bool {
 		}
 		var resourceNames []string
 		var nackErr error
-		s.fc.setPending()
-		resourceNames, nackErr = s.eventHandler.onResponse(resp, s.fc.onDone)
+		s.fc.setPending(true)
+		resourceNames, nackErr = s.eventHandler.onResponse(resp, sync.OnceFunc(func() { s.fc.setPending(false) }))
 		if xdsresource.ErrType(nackErr) == xdsresource.ErrorTypeResourceTypeUnsupported {
 			// A general guiding principle is that if the server sends
 			// something the client didn't actually subscribe to, then the
@@ -707,69 +708,84 @@ func resourceNames(m map[string]*xdsresource.ResourceWatchState) []string {
 	return ret
 }
 
-// adsFlowControl implements ADS stream level flow control that enables the
-// transport to block the reading of the next message off of the stream until
-// the previous update is consumed by all watchers.
+// adsFlowControl implements ADS stream level flow control that enables the ADS
+// stream to block the reading of the next message until the previous update is
+// consumed by all watchers.
 //
-// The lifetime of the flow control is tied to the lifetime of the stream.
+// The lifetime of the flow control is tied to the lifetime of the stream. When
+// the stream is closed, it is the responsibility of the caller to stop the flow
+// control. This ensures that any goroutine blocked on the flow control's wait
+// method is unblocked.
 type adsFlowControl struct {
-	logger *igrpclog.PrefixLogger
-
-	// Whether the most recent update is pending consumption by all watchers.
-	pending atomic.Bool
-	// Channel used to notify when all the watchers have consumed the most
-	// recent update. Wait() blocks on reading a value from this channel.
-	readyCh chan struct{}
+	mu sync.Mutex
+	// cond is used to signal when the most recent update has been consumed, or
+	// the flow control has been stopped (in which case, waiters should be
+	// unblocked as well).
+	cond    *sync.Cond
+	pending bool // indicates if the most recent update is pending consumption
+	stopped bool // indicates if the ADS stream has been stopped
 }
 
 // newADSFlowControl returns a new adsFlowControl.
-func newADSFlowControl(logger *igrpclog.PrefixLogger) *adsFlowControl {
-	return &adsFlowControl{
-		logger:  logger,
-		readyCh: make(chan struct{}, 1),
-	}
+func newADSFlowControl() *adsFlowControl {
+	fc := &adsFlowControl{}
+	fc.cond = sync.NewCond(&fc.mu)
+	return fc
 }
 
-// setPending changes the internal state to indicate that there is an update
-// pending consumption by all watchers.
-func (fc *adsFlowControl) setPending() {
-	fc.pending.Store(true)
+// stop marks the flow control as stopped and signals the condition variable to
+// unblock any goroutine waiting on it.
+func (fc *adsFlowControl) stop() {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+
+	fc.stopped = true
+	fc.cond.Broadcast()
 }
 
-// wait blocks until all the watchers have consumed the most recent update and
-// returns true. If the context expires before that, it returns false.
-func (fc *adsFlowControl) wait(ctx context.Context) bool {
-	// If there is no pending update, there is no need to block.
-	if !fc.pending.Load() {
-		// If all watchers finished processing the most recent update before the
-		// `recv` goroutine made the next call to `Wait()`, there would be an
-		// entry in the readyCh channel that needs to be drained to ensure that
-		// the next call to `Wait()` doesn't unblock before it actually should.
-		select {
-		case <-fc.readyCh:
-		default:
-		}
-		return true
+// setPending changes the internal state to indicate whether there is an update
+// pending consumption by all watchers. If there is no longer a pending update,
+// the condition variable is signaled to allow the recv method to proceed.
+func (fc *adsFlowControl) setPending(pending bool) {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+
+	if fc.stopped {
+		return
 	}
 
-	select {
-	case <-ctx.Done():
+	fc.pending = pending
+	if !pending {
+		fc.cond.Broadcast()
+	}
+}
+
+func (fc *adsFlowControl) runIfPending(f func()) bool {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+
+	if fc.stopped {
 		return false
-	case <-fc.readyCh:
-		return true
 	}
+
+	// If there's a pending update, run the function while still holding the
+	// lock. This ensures that the pending state does not change between the
+	// check and the function call.
+	if fc.pending {
+		f()
+	}
+	return fc.pending
 }
 
-// onDone indicates that all watchers have consumed the most recent update.
-func (fc *adsFlowControl) onDone() {
-	select {
-	// Writes to the readyCh channel should not block ideally. The default
-	// branch here is to appease the paranoid mind.
-	case fc.readyCh <- struct{}{}:
-	default:
-		if fc.logger.V(2) {
-			fc.logger.Infof("ADS stream flow control readyCh is full")
-		}
+// wait blocks until all the watchers have consumed the most recent update.
+// Returns true if the flow control was stopped while waiting, false otherwise.
+func (fc *adsFlowControl) wait() bool {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+
+	for fc.pending && !fc.stopped {
+		fc.cond.Wait()
 	}
-	fc.pending.Store(false)
+
+	return fc.stopped
 }
diff --git a/internal/xds/clients/xdsclient/authority.go b/internal/xds/clients/xdsclient/authority.go
index 49a4480c..e4425fa7 100644
--- a/internal/xds/clients/xdsclient/authority.go
+++ b/internal/xds/clients/xdsclient/authority.go
@@ -570,10 +570,6 @@ func (a *authority) handleRevertingToPrimaryOnUpdate(serverConfig *ServerConfig)
 		return true
 	}
 
-	if a.logger.V(2) {
-		a.logger.Infof("Received update from non-active server %q", serverConfig)
-	}
-
 	// If the resource update is not from the current active server, it means
 	// that we have received an update either from:
 	// - a server that has a higher priority than the current active server and
@@ -586,8 +582,14 @@ func (a *authority) handleRevertingToPrimaryOnUpdate(serverConfig *ServerConfig)
 	serverIdx := a.serverIndexForConfig(serverConfig)
 	activeServerIdx := a.serverIndexForConfig(a.activeXDSChannel.serverConfig)
 	if activeServerIdx < serverIdx {
+		if a.logger.V(2) {
+			a.logger.Infof("Ignoring update from lower priority server [%d] %q", serverIdx, serverConfig)
+		}
 		return false
 	}
+	if a.logger.V(2) {
+		a.logger.Infof("Received update from higher priority server [%d] %q", serverIdx, serverConfig)
+	}
 
 	// At this point, we are guaranteed that we have received a response from a
 	// higher priority server compared to the current active server. So, we
@@ -622,7 +624,7 @@ func (a *authority) handleRevertingToPrimaryOnUpdate(serverConfig *ServerConfig)
 		// Release the reference to the channel.
 		if cfg.cleanup != nil {
 			if a.logger.V(2) {
-				a.logger.Infof("Closing lower priority server %q", cfg.serverConfig)
+				a.logger.Infof("Closing lower priority server [%d] %q", i, cfg.serverConfig)
 			}
 			cfg.cleanup()
 			cfg.cleanup = nil
diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go
index cc7d5c4e..b1c69554 100644
--- a/internal/xds/clients/xdsclient/xdsclient.go
+++ b/internal/xds/clients/xdsclient/xdsclient.go
@@ -319,7 +319,7 @@ func (c *XDSClient) releaseChannel(serverConfig *ServerConfig, state *channelSta
 		c.channelsMu.Lock()
 
 		if c.logger.V(2) {
-			c.logger.Infof("Received request to release a reference to an xdsChannel for server config %q", serverConfig)
+			c.logger.Infof("Received request to release a reference to an xdsChannel for server config %+v", serverConfig)
 		}
 		deInitLocked(state)
 
diff --git a/internal/xds/resolver/serviceconfig.go b/internal/xds/resolver/serviceconfig.go
index 2b4e5a05..f2ceabe7 100644
--- a/internal/xds/resolver/serviceconfig.go
+++ b/internal/xds/resolver/serviceconfig.go
@@ -151,8 +151,11 @@ func (cs *erroringConfigSelector) SelectConfig(iresolver.RPCInfo) (*iresolver.RP
 func (cs *erroringConfigSelector) stop() {}
 
 type configSelector struct {
-	r                *xdsResolver
-	xdsNodeID        string
+	channelID            uint64 // Static hash when hash policy is HashPolicyTypeChannelID
+	xdsNodeID            string // xDS node ID, for annotating errors.
+	sendNewServiceConfig func() // Function to send a new service config to gRPC.
+
+	// Configuration received from the xDS management server.
 	virtualHost      virtualHost
 	routes           []route
 	clusters         map[string]*clusterInfo
@@ -214,9 +217,7 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 			if v := atomic.AddInt32(ref, -1); v == 0 {
 				// This entry will be removed from activeClusters when
 				// producing the service config for the empty update.
-				cs.r.serializer.TrySchedule(func(context.Context) {
-					cs.r.onClusterRefDownToZero()
-				})
+				cs.sendNewServiceConfig()
 			}
 		},
 		Interceptor: interceptor,
@@ -281,7 +282,7 @@ func (cs *configSelector) generateHash(rpcInfo iresolver.RPCInfo, hashPolicies [
 			generatedPolicyHash = true
 		case xdsresource.HashPolicyTypeChannelID:
 			// Use the static channel ID as the hash for this policy.
-			policyHash = cs.r.channelID
+			policyHash = cs.channelID
 			generatedHash = true
 			generatedPolicyHash = true
 		}
@@ -358,9 +359,7 @@ func (cs *configSelector) stop() {
 	// selector; we need another update to delete clusters from the config (if
 	// we don't have another update pending already).
 	if needUpdate {
-		cs.r.serializer.TrySchedule(func(context.Context) {
-			cs.r.onClusterRefDownToZero()
-		})
+		cs.sendNewServiceConfig()
 	}
 }
 
diff --git a/internal/xds/resolver/xds_resolver.go b/internal/xds/resolver/xds_resolver.go
index 70291f8c..4ed400c3 100644
--- a/internal/xds/resolver/xds_resolver.go
+++ b/internal/xds/resolver/xds_resolver.go
@@ -330,8 +330,13 @@ func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) bool {
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) newConfigSelector() *configSelector {
 	cs := &configSelector{
-		r:         r,
+		channelID: r.channelID,
 		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
+		sendNewServiceConfig: func() {
+			r.serializer.TrySchedule(func(context.Context) {
+				r.sendNewServiceConfig(r.curConfigSelector)
+			})
+		},
 		virtualHost: virtualHost{
 			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
 			retryConfig:              r.currentVirtualHost.RetryConfig,
@@ -607,8 +612,3 @@ func (r *xdsResolver) onRouteConfigResourceError(name string, err error) {
 	}
 	r.onResourceError(err)
 }
-
-// Only executed in the context of a serializer callback.
-func (r *xdsResolver) onClusterRefDownToZero() {
-	r.sendNewServiceConfig(r.curConfigSelector)
-}
diff --git a/stats/opentelemetry/client_metrics.go b/stats/opentelemetry/client_metrics.go
index ee778080..d2d80494 100644
--- a/stats/opentelemetry/client_metrics.go
+++ b/stats/opentelemetry/client_metrics.go
@@ -72,9 +72,6 @@ func (h *clientMetricsHandler) initializeMetrics() {
 func getOrCreateCallInfo(ctx context.Context, cc *grpc.ClientConn, method string, opts ...grpc.CallOption) (context.Context, *callInfo) {
 	ci := getCallInfo(ctx)
 	if ci == nil {
-		if logger.V(2) {
-			logger.Info("Creating new CallInfo since its not present in context")
-		}
 		ci = &callInfo{
 			target: cc.CanonicalTarget(),
 			method: determineMethod(method, opts...),



Test patch:
diff --git a/balancer/pickfirst/pickfirst_ext_test.go b/balancer/pickfirst/pickfirst_ext_test.go
index 81691fc8..207fc431 100644
--- a/balancer/pickfirst/pickfirst_ext_test.go
+++ b/balancer/pickfirst/pickfirst_ext_test.go
@@ -20,6 +20,7 @@ package pickfirst_test
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -28,11 +29,14 @@ import (
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
+	"google.golang.org/grpc/balancer"
+	pfbalancer "google.golang.org/grpc/balancer/pickfirst"
 	pfinternal "google.golang.org/grpc/balancer/pickfirst/internal"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
@@ -463,6 +467,85 @@ func (s) TestPickFirst_ShuffleAddressList(t *testing.T) {
 	}
 }
 
+// Tests the PF LB policy with shuffling enabled. It explicitly unsets the
+// Endpoints field in the resolver update to test the shuffling of the
+// Addresses.
+func (s) TestPickFirst_ShuffleAddressListNoEndpoints(t *testing.T) {
+	// Install a shuffler that always reverses two entries.
+	origShuf := pfinternal.RandShuffle
+	defer func() { pfinternal.RandShuffle = origShuf }()
+	pfinternal.RandShuffle = func(n int, f func(int, int)) {
+		if n != 2 {
+			t.Errorf("Shuffle called with n=%v; want 2", n)
+			return
+		}
+		f(0, 1) // reverse the two addresses
+	}
+
+	pfBuilder := balancer.Get(pfbalancer.Name)
+	shuffleConfig, err := pfBuilder.(balancer.ConfigParser).ParseConfig(json.RawMessage(`{ "shuffleAddressList": true }`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	noShuffleConfig, err := pfBuilder.(balancer.ConfigParser).ParseConfig(json.RawMessage(`{ "shuffleAddressList": false }`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var activeCfg serviceconfig.LoadBalancingConfig
+
+	bf := stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			bd.ChildBalancer = pfBuilder.Build(bd.ClientConn, bd.BuildOptions)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.ChildBalancer.Close()
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			ccs.BalancerConfig = activeCfg
+			ccs.ResolverState.Endpoints = nil
+			return bd.ChildBalancer.UpdateClientConnState(ccs)
+		},
+	}
+
+	stub.Register(t.Name(), bf)
+	svcCfg := fmt.Sprintf(`{ "loadBalancingConfig": [{%q: {}}] }`, t.Name())
+	// Set up our backends.
+	cc, r, backends := setupPickFirst(t, 2, grpc.WithDefaultServiceConfig(svcCfg))
+	addrs := stubBackendsToResolverAddrs(backends)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Push an update with both addresses and shuffling disabled.  We should
+	// connect to backend 0.
+	activeCfg = noShuffleConfig
+	resolverState := resolver.State{Addresses: addrs}
+	r.UpdateState(resolverState)
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
+		t.Fatal(err)
+	}
+
+	// Send a config with shuffling enabled.  This will reverse the addresses,
+	// but the channel should still be connected to backend 0.
+	activeCfg = shuffleConfig
+	r.UpdateState(resolverState)
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
+		t.Fatal(err)
+	}
+
+	// Send a resolver update with no addresses. This should push the channel
+	// into TransientFailure.
+	r.UpdateState(resolver.State{})
+	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
+
+	// Send the same config as last time with shuffling enabled.  Since we are
+	// not connected to backend 0, we should connect to backend 1.
+	r.UpdateState(resolverState)
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[1]); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // Test config parsing with the env var turned on and off for various scenarios.
 func (s) TestPickFirst_ParseConfig_Success(t *testing.T) {
 	// Install a shuffler that always reverses two entries.
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
index df8c06f4..6b2fabe2 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
@@ -1504,6 +1504,102 @@ func (s) TestPickFirstLeaf_AddressUpdateWithMetadata(t *testing.T) {
 	}
 }
 
+// Tests the scenario where a connection is established and then breaks, leading
+// to a reconnection attempt. While the reconnection is in progress, a resolver
+// update with a new address is received. The test verifies that the balancer
+// creates a new SubConn for the new address and that the ClientConn eventually
+// becomes READY.
+func (s) TestPickFirstLeaf_Reconnection(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	cc := testutils.NewBalancerClientConn(t)
+	bal := balancer.Get(pickfirstleaf.Name).Build(cc, balancer.BuildOptions{})
+	defer bal.Close()
+	ccState := balancer.ClientConnState{
+		ResolverState: resolver.State{
+			Endpoints: []resolver.Endpoint{
+				{Addresses: []resolver.Address{{Addr: "1.1.1.1:1"}}},
+			},
+		},
+	}
+	if err := bal.UpdateClientConnState(ccState); err != nil {
+		t.Fatalf("UpdateClientConnState(%v) returned error: %v", ccState, err)
+	}
+
+	select {
+	case state := <-cc.NewStateCh:
+		if got, want := state, connectivity.Connecting; got != want {
+			t.Fatalf("Received unexpected ClientConn sate: got %v, want %v", got, want)
+		}
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for ClientConn state update.")
+	}
+
+	sc1 := <-cc.NewSubConnCh
+	select {
+	case <-sc1.ConnectCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for Connect() to be called on sc1.")
+	}
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Ready); err != nil {
+		t.Fatalf("Context timed out waiting for ClientConn to become READY.")
+	}
+
+	// Simulate a connection breakage, this should result the channel
+	// transitioning to IDLE.
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Idle); err != nil {
+		t.Fatalf("Context timed out waiting for ClientConn to enter IDLE.")
+	}
+
+	// Calling the idle picker should result in the SubConn being re-connected.
+	picker := <-cc.NewPickerCh
+	if _, err := picker.Pick(balancer.PickInfo{}); err != balancer.ErrNoSubConnAvailable {
+		t.Fatalf("picker.Pick() returned error: %v, want %v", err, balancer.ErrNoSubConnAvailable)
+	}
+
+	select {
+	case <-sc1.ConnectCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for Connect() to be called on sc1.")
+	}
+
+	// Send a resolver update, removing the existing SubConn. Since the balancer
+	// is connecting, it should create a new SubConn for the new backend
+	// address.
+	ccState = balancer.ClientConnState{
+		ResolverState: resolver.State{
+			Endpoints: []resolver.Endpoint{
+				{Addresses: []resolver.Address{{Addr: "2.2.2.2:2"}}},
+			},
+		},
+	}
+	if err := bal.UpdateClientConnState(ccState); err != nil {
+		t.Fatalf("UpdateClientConnState(%v) returned error: %v", ccState, err)
+	}
+
+	var sc2 *testutils.TestSubConn
+	select {
+	case sc2 = <-cc.NewSubConnCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for new SubConn to be created.")
+	}
+
+	select {
+	case <-sc2.ConnectCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for Connect() to be called on sc2.")
+	}
+	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Ready); err != nil {
+		t.Fatalf("Context timed out waiting for ClientConn to become READY.")
+	}
+}
+
 // healthListenerCapturingCCWrapper is used to capture the health listener so
 // that health updates can be mocked for testing.
 type healthListenerCapturingCCWrapper struct {
diff --git a/internal/testutils/xds/e2e/clientresources.go b/internal/testutils/xds/e2e/clientresources.go
index b8ebf8ba..be375d40 100644
--- a/internal/testutils/xds/e2e/clientresources.go
+++ b/internal/testutils/xds/e2e/clientresources.go
@@ -712,11 +712,7 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 					},
 				}
 			}
-			metadata, err := structpb.NewStruct(b.Metadata)
-			if err != nil {
-				panic(err)
-			}
-			lbEndpoints = append(lbEndpoints, &v3endpointpb.LbEndpoint{
+			lbEndpoint := &v3endpointpb.LbEndpoint{
 				HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{Endpoint: &v3endpointpb.Endpoint{
 					Address: &v3corepb.Address{Address: &v3corepb.Address_SocketAddress{
 						SocketAddress: &v3corepb.SocketAddress{
@@ -729,12 +725,20 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 				}},
 				HealthStatus:        b.HealthStatus,
 				LoadBalancingWeight: &wrapperspb.UInt32Value{Value: b.Weight},
-				Metadata: &v3corepb.Metadata{
+			}
+
+			if b.Metadata != nil {
+				metadata, err := structpb.NewStruct(b.Metadata)
+				if err != nil {
+					panic(fmt.Sprintf("failed to marshal metadata: %v", err))
+				}
+				lbEndpoint.Metadata = &v3corepb.Metadata{
 					FilterMetadata: map[string]*structpb.Struct{
 						"envoy.lb": metadata,
 					},
-				},
-			})
+				}
+			}
+			lbEndpoints = append(lbEndpoints, lbEndpoint)
 		}
 
 		l := locality.Locality
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index d0c8a88d..0b0396b0 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -3008,6 +3008,64 @@ func (s) TestClientCloseReturnsEarlyWhenGoAwayWriteHangs(t *testing.T) {
 	ct.Close(errors.New("manually closed by client"))
 }
 
+// deadlineTestConn is a net.Conn wrapper used to assert that deadlines are set
+// during http2Client.Close().
+type deadlineTestConn struct {
+	net.Conn
+	// We use atomic.Bool here since there may be more than one call to
+	// http2Client.Close -- which sets these deadlines -- and not all of them
+	// from the same goroutine as our test. In fact we only care about the first
+	// such invocation, which *does* come from the main goroutine of our test,
+	// but the race detector can't know that and complains (understandably)
+	// about writes from those successive calls when these variables are not
+	// atomic.Bool.
+	//
+	// For more detailed background, see
+	// https://github.com/grpc/grpc-go/pull/8534#discussion_r2297717445 .
+	observedReadDeadline  atomic.Bool
+	observedWriteDeadline atomic.Bool
+}
+
+func (c *deadlineTestConn) SetReadDeadline(t time.Time) error {
+	c.observedReadDeadline.Store(true)
+	return c.Conn.SetReadDeadline(t)
+}
+
+func (c *deadlineTestConn) SetWriteDeadline(t time.Time) error {
+	c.observedWriteDeadline.Store(true)
+	return c.Conn.SetWriteDeadline(t)
+}
+
+// Tests that connection read and write deadlines are set as expected during
+// Close().
+func (s) TestCloseSetsConnectionDeadlines(t *testing.T) {
+	dialer := func(_ context.Context, addr string) (net.Conn, error) {
+		conn, err := net.Dial("tcp", addr)
+		if err != nil {
+			return nil, err
+		}
+		return &deadlineTestConn{Conn: conn}, nil
+	}
+	co := ConnectOptions{
+		Dialer: dialer,
+	}
+	server, client, cancel := setUpWithOptions(t, 0, &ServerConfig{}, normal, co)
+	defer cancel()
+	defer server.stop()
+	dConn := client.conn.(*deadlineTestConn)
+	// Set both to false before invoking Close() in case some other code set a
+	// deadline above.
+	dConn.observedReadDeadline.Store(false)
+	dConn.observedWriteDeadline.Store(false)
+	client.Close(fmt.Errorf("closed manually by test"))
+	if !dConn.observedReadDeadline.Load() {
+		t.Errorf("Connection read deadline was never set")
+	}
+	if !dConn.observedWriteDeadline.Load() {
+		t.Errorf("Connection write deadline was never set")
+	}
+}
+
 // TestReadHeaderMultipleBuffers tests the stream when the gRPC headers are
 // split across multiple buffers. It verifies that the reporting of the
 // number of bytes read for flow control is correct.
diff --git a/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go b/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go
index 581f511b..7e277f3e 100644
--- a/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go
+++ b/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go
@@ -39,9 +39,14 @@ import (
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
+
+	_ "google.golang.org/grpc/internal/xds/httpfilter/router" // Register the router filter.
+	_ "google.golang.org/grpc/internal/xds/resolver"          // Register the xds resolver
 )
 
 // makeAggregateClusterResource returns an aggregate cluster resource with the
@@ -105,13 +110,13 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 		},
 		{
 			name:                  "dns",
-			firstClusterResource:  makeLogicalDNSClusterResource(clusterName, "dns_host", uint32(8080)),
-			secondClusterResource: makeLogicalDNSClusterResource(clusterName, "dns_host_new", uint32(8080)),
+			firstClusterResource:  makeLogicalDNSClusterResource(clusterName, "dns_host", uint32(port)),
+			secondClusterResource: makeLogicalDNSClusterResource(clusterName, "dns_host_new", uint32(port)),
 			wantFirstChildCfg: &clusterresolver.LBConfig{
 				DiscoveryMechanisms: []clusterresolver.DiscoveryMechanism{{
 					Cluster:          clusterName,
 					Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
-					DNSHostname:      "dns_host:8080",
+					DNSHostname:      fmt.Sprintf("dns_host:%d", port),
 					OutlierDetection: json.RawMessage(`{}`),
 					TelemetryLabels:  xdsinternal.UnknownCSMLabels,
 				}},
@@ -121,7 +126,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 				DiscoveryMechanisms: []clusterresolver.DiscoveryMechanism{{
 					Cluster:          clusterName,
 					Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
-					DNSHostname:      "dns_host_new:8080",
+					DNSHostname:      fmt.Sprintf("dns_host_new:%d", port),
 					OutlierDetection: json.RawMessage(`{}`),
 					TelemetryLabels:  xdsinternal.UnknownCSMLabels,
 				}},
@@ -133,13 +138,16 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-			mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+			mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 			// Push the first cluster resource through the management server and
 			// verify the configuration pushed to the child policy.
 			resources := e2e.UpdateOptions{
 				NodeID:         nodeID,
+				Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+				Routes:         []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 				Clusters:       []*v3clusterpb.Cluster{test.firstClusterResource},
+				Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 				SkipValidation: true,
 			}
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -154,6 +162,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 			// Push the second cluster resource through the management server and
 			// verify the configuration pushed to the child policy.
 			resources.Clusters[0] = test.secondClusterResource
+			resources.Endpoints[0] = e2e.DefaultEndpoint(serviceName+"-new", host, []uint32{port})
 			if err := mgmtServer.Update(ctx, resources); err != nil {
 				t.Fatal(err)
 			}
@@ -176,19 +185,21 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 // contains the expected discovery mechanisms.
 func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	// Configure the management server with the aggregate cluster resource
 	// pointing to two child clusters, one EDS and one LogicalDNS. Include the
 	// resource corresponding to the EDS cluster here, but don't include
 	// resource corresponding to the LogicalDNS cluster yet.
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{edsClusterName, dnsClusterName}),
 			e2e.DefaultCluster(edsClusterName, serviceName, e2e.SecurityLevelNone),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -237,14 +248,16 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
 	const dnsClusterNameNew = dnsClusterName + "-new"
 	const dnsHostNameNew = dnsHostName + "-new"
 	resources = e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{edsClusterName, dnsClusterNameNew}),
 			e2e.DefaultCluster(edsClusterName, serviceName, e2e.SecurityLevelNone),
 			makeLogicalDNSClusterResource(dnsClusterName, dnsHostName, dnsPort),
 			makeLogicalDNSClusterResource(dnsClusterNameNew, dnsHostNameNew, dnsPort),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -283,18 +296,20 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
 // policy contains a single discovery mechanism.
 func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	// Configure the management server with the aggregate cluster resource
 	// pointing to two child clusters.
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{edsClusterName, dnsClusterName}),
 			e2e.DefaultCluster(edsClusterName, serviceName, e2e.SecurityLevelNone),
 			makeLogicalDNSClusterResource(dnsClusterName, dnsHostName, dnsPort),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -326,12 +341,14 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
 	}
 
 	resources = e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone),
 			makeLogicalDNSClusterResource(dnsClusterName, dnsHostName, dnsPort),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -358,13 +375,15 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
 // discovery mechanisms.
 func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	// Start off with the requested cluster being a leaf EDS cluster.
 	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		SkipValidation: true,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -388,13 +407,15 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 	// Switch the requested cluster to be an aggregate cluster pointing to two
 	// child clusters.
 	resources = e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{edsClusterName, dnsClusterName}),
 			e2e.DefaultCluster(edsClusterName, serviceName, e2e.SecurityLevelNone),
 			makeLogicalDNSClusterResource(dnsClusterName, dnsHostName, dnsPort),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -424,9 +445,11 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 
 	// Switch the cluster back to a leaf EDS cluster.
 	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		SkipValidation: true,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -452,10 +475,13 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 // longer exceed maximum depth, but be at the maximum allowed depth, and
 // verifies that an RPC can be made successfully.
 func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
+
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{clusterName + "-1"}),
 			makeAggregateClusterResource(clusterName+"-1", []string{clusterName + "-2"}),
@@ -475,7 +501,7 @@ func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
 			makeAggregateClusterResource(clusterName+"-15", []string{clusterName + "-16"}),
 			e2e.DefaultCluster(clusterName+"-16", serviceName, e2e.SecurityLevelNone),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -502,7 +528,9 @@ func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
 	// Update the aggregate cluster resource to no longer exceed max depth, and
 	// be at the maximum depth allowed.
 	resources = e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterName, []string{clusterName + "-1"}),
 			makeAggregateClusterResource(clusterName+"-1", []string{clusterName + "-2"}),
@@ -521,8 +549,7 @@ func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
 			makeAggregateClusterResource(clusterName+"-14", []string{clusterName + "-15"}),
 			e2e.DefaultCluster(clusterName+"-15", serviceName, e2e.SecurityLevelNone),
 		},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, server.Address)})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -540,7 +567,7 @@ func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
 // pushed only after all child clusters are resolved.
 func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	// Configure the management server with an aggregate cluster resource having
 	// a diamond dependency pattern, (A->[B,C]; B->D; C->D). Includes resources
@@ -554,13 +581,15 @@ func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
 		clusterNameD = clusterName + "-D"
 	)
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterNameA, []string{clusterNameB, clusterNameC}),
 			makeAggregateClusterResource(clusterNameB, []string{clusterNameD}),
 			e2e.DefaultCluster(clusterNameD, serviceName, e2e.SecurityLevelNone),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -607,7 +636,7 @@ func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
 // pushed only after all child clusters are resolved.
 func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	// Configure the management server with an aggregate cluster resource that
 	// has duplicates in the graph, (A->[B, C]; B->[C, D]). Include resources
@@ -621,13 +650,15 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 		clusterNameD = clusterName + "-D"
 	)
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterNameA, []string{clusterNameB, clusterNameC}),
 			makeAggregateClusterResource(clusterNameB, []string{clusterNameC, clusterNameD}),
 			e2e.DefaultCluster(clusterNameD, serviceName, e2e.SecurityLevelNone),
 		},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -646,7 +677,8 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 	// Now configure the resource for cluster C in the management server,
 	// thereby completing the cluster graph. This should result in configuration
 	// being pushed down to the child policy.
-	resources.Clusters = append(resources.Clusters, e2e.DefaultCluster(clusterNameC, serviceName, e2e.SecurityLevelNone))
+	resources.Clusters = append(resources.Clusters, e2e.DefaultCluster(clusterNameC, edsClusterName, e2e.SecurityLevelNone))
+	resources.Endpoints = append(resources.Endpoints, e2e.DefaultEndpoint(edsClusterName, "eshita_eds", []uint32{1234}))
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -656,7 +688,7 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 			{
 				Cluster:          clusterNameC,
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
-				EDSServiceName:   serviceName,
+				EDSServiceName:   edsClusterName,
 				OutlierDetection: json.RawMessage(`{}`),
 				TelemetryLabels:  xdsinternal.UnknownCSMLabels,
 			},
@@ -685,7 +717,7 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 // child policy and that an RPC can be successfully made.
 func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	const (
 		clusterNameA = clusterName // cluster name in cds LB policy config
@@ -695,6 +727,8 @@ func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
 	// child is itself.
 	resources := e2e.UpdateOptions{
 		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters:       []*v3clusterpb.Cluster{makeAggregateClusterResource(clusterNameA, []string{clusterNameA})},
 		SkipValidation: true,
 	}
@@ -729,13 +763,14 @@ func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
 
 	// Update the aggregate cluster to point to a leaf EDS cluster.
 	resources = e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterNameA, []string{clusterNameB}),
 			e2e.DefaultCluster(clusterNameB, serviceName, e2e.SecurityLevelNone),
 		},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, server.Address)})},
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
@@ -770,7 +805,7 @@ func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
 // that the aggregate cluster graph has no leaf clusters.
 func (s) TestAggregatedCluster_CycleWithNoLeafNode(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	const (
 		clusterNameA = clusterName // cluster name in cds LB policy config
@@ -779,7 +814,9 @@ func (s) TestAggregatedCluster_CycleWithNoLeafNode(t *testing.T) {
 	// Configure the management server with an aggregate cluster resource graph
 	// that contains a cycle and no leaf clusters.
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterNameA, []string{clusterNameB}),
 			makeAggregateClusterResource(clusterNameB, []string{clusterNameA}),
@@ -818,7 +855,7 @@ func (s) TestAggregatedCluster_CycleWithNoLeafNode(t *testing.T) {
 // child policy and RPCs should get routed to that leaf cluster.
 func (s) TestAggregatedCluster_CycleWithLeafNode(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	// Start a test service backend.
 	server := stubserver.StartTestService(t, nil)
@@ -832,13 +869,15 @@ func (s) TestAggregatedCluster_CycleWithLeafNode(t *testing.T) {
 	// Configure the management server with an aggregate cluster resource graph
 	// that contains a cycle, but also contains a leaf cluster.
 	resources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterNameA, []string{clusterNameB}),
 			makeAggregateClusterResource(clusterNameB, []string{clusterNameA, clusterNameC}),
 			e2e.DefaultCluster(clusterNameC, serviceName, e2e.SecurityLevelNone),
 		},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
+		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, server.Address)})},
 		SkipValidation: true,
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -888,7 +927,7 @@ func (s) TestWatchers(t *testing.T) {
 		}
 		return nil
 	}
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, onStreamReq)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, onStreamReq)
 
 	const (
 		clusterA = clusterName
@@ -899,7 +938,9 @@ func (s) TestWatchers(t *testing.T) {
 
 	// Initial CDS resources: A -> B, C
 	initialResources := e2e.UpdateOptions{
-		NodeID: nodeID,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterA)},
 		Clusters: []*v3clusterpb.Cluster{
 			makeAggregateClusterResource(clusterA, []string{clusterB, clusterC}),
 		},
@@ -914,14 +955,10 @@ func (s) TestWatchers(t *testing.T) {
 	}
 
 	// Update the CDS resources to remove cluster C and add cluster D.
-	updatedResources := e2e.UpdateOptions{
-		NodeID: nodeID,
-		Clusters: []*v3clusterpb.Cluster{
-			makeAggregateClusterResource(clusterA, []string{clusterB, clusterD}),
-		},
-		SkipValidation: true,
+	initialResources.Clusters = []*v3clusterpb.Cluster{
+		makeAggregateClusterResource(clusterA, []string{clusterB, clusterD}),
 	}
-	if err := mgmtServer.Update(ctx, updatedResources); err != nil {
+	if err := mgmtServer.Update(ctx, initialResources); err != nil {
 		t.Fatalf("Update failed: %v", err)
 	}
 	wantNames = []string{clusterA, clusterB, clusterD}
diff --git a/internal/xds/balancer/cdsbalancer/cdsbalancer_security_test.go b/internal/xds/balancer/cdsbalancer/cdsbalancer_security_test.go
index 401a16dc..f26571fd 100644
--- a/internal/xds/balancer/cdsbalancer/cdsbalancer_security_test.go
+++ b/internal/xds/balancer/cdsbalancer/cdsbalancer_security_test.go
@@ -25,110 +25,41 @@ import (
 	"os"
 	"strings"
 	"testing"
-	"unsafe"
 
-	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/attributes"
-	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/credentials/tls/certprovider"
 	"google.golang.org/grpc/credentials/xds"
 	"google.golang.org/grpc/internal"
-	"google.golang.org/grpc/internal/balancer/stub"
-	xdscredsinternal "google.golang.org/grpc/internal/credentials/xds"
 	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/xds/bootstrap"
-	"google.golang.org/grpc/internal/xds/xdsclient"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
-	"google.golang.org/grpc/resolver/manual"
-	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/testdata"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
 
 	_ "google.golang.org/grpc/credentials/tls/certprovider/pemfile" // Register the file watcher certificate provider plugin.
+	_ "google.golang.org/grpc/internal/xds/httpfilter/router"       // Register the router filter.
+	_ "google.golang.org/grpc/internal/xds/resolver"                // Register the xds resolver
 )
 
-// testCCWrapper wraps a balancer.ClientConn and intercepts NewSubConn and
-// returns the xDS handshake info back to the test for inspection.
-type testCCWrapper struct {
-	balancer.ClientConn
-	handshakeInfoCh chan *xdscredsinternal.HandshakeInfo
-}
-
-// NewSubConn forwards the call to the underlying balancer.ClientConn, but
-// before that, it validates the following:
-//   - there is only one address in the addrs slice
-//   - the single address contains xDS handshake information, which is then
-//     pushed onto the handshakeInfoCh channel
-func (tcc *testCCWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
-	if len(addrs) != 1 {
-		return nil, fmt.Errorf("NewSubConn got %d addresses, want 1", len(addrs))
-	}
-	getHI := internal.GetXDSHandshakeInfoForTesting.(func(attr *attributes.Attributes) *unsafe.Pointer)
-	hi := getHI(addrs[0].Attributes)
-	if hi == nil {
-		return nil, fmt.Errorf("NewSubConn got address without xDS handshake info")
-	}
-
-	sc, err := tcc.ClientConn.NewSubConn(addrs, opts)
-	select {
-	case tcc.handshakeInfoCh <- (*xdscredsinternal.HandshakeInfo)(*hi):
-	default:
-	}
-	return sc, err
-}
-
-// Registers a wrapped cds LB policy for the duration of this test that retains
-// all the functionality of the original cds LB policy, but overrides the
-// NewSubConn method passed to the policy and makes the xDS handshake
-// information passed to NewSubConn available to the test.
-//
-// Accepts as argument a channel onto which xDS handshake information passed to
-// NewSubConn is written to.
-func registerWrappedCDSPolicyWithNewSubConnOverride(t *testing.T, ch chan *xdscredsinternal.HandshakeInfo) {
-	cdsBuilder := balancer.Get(cdsName)
-	internal.BalancerUnregister(cdsBuilder.Name())
-	var ccWrapper *testCCWrapper
-	stub.Register(cdsBuilder.Name(), stub.BalancerFuncs{
-		Init: func(bd *stub.BalancerData) {
-			ccWrapper = &testCCWrapper{
-				ClientConn:      bd.ClientConn,
-				handshakeInfoCh: ch,
-			}
-			bd.ChildBalancer = cdsBuilder.Build(ccWrapper, bd.BuildOptions)
-		},
-		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
-			return cdsBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
-		},
-		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
-			return bd.ChildBalancer.UpdateClientConnState(ccs)
-		},
-		Close: func(bd *stub.BalancerData) {
-			bd.ChildBalancer.Close()
-		},
-	})
-	t.Cleanup(func() { balancer.Register(cdsBuilder) })
-}
-
 // Common setup for security tests:
 //   - creates an xDS client with the specified bootstrap configuration
-//   - creates a manual resolver that specifies cds as the top-level LB policy
-//   - creates a channel that uses the passed in client creds and the manual
-//     resolver
+//   - creates a gRPC channel that uses the passed in client creds
 //   - creates a test server that uses the passed in server creds
 //
 // Returns the following:
@@ -137,32 +68,13 @@ func registerWrappedCDSPolicyWithNewSubConnOverride(t *testing.T, ch chan *xdscr
 func setupForSecurityTests(t *testing.T, bootstrapContents []byte, clientCreds, serverCreds credentials.TransportCredentials) (*grpc.ClientConn, string) {
 	t.Helper()
 
-	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
-	if err != nil {
-		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
 	}
-	pool := xdsclient.NewPool(config)
-	xdsClient, xdsClose, err := pool.NewClientForTesting(xdsclient.OptionsForTesting{
-		Name: t.Name(),
-	})
+	r, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bootstrapContents)
 	if err != nil {
-		t.Fatalf("Failed to create xDS client: %v", err)
-	}
-	t.Cleanup(xdsClose)
-
-	// Create a manual resolver that configures the CDS LB policy as the
-	// top-level LB policy on the channel.
-	r := manual.NewBuilderWithScheme("whatever")
-	jsonSC := fmt.Sprintf(`{
-			"loadBalancingConfig":[{
-				"cds_experimental":{
-					"cluster": "%s"
-				}
-			}]
-		}`, clusterName)
-	scpr := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(jsonSC)
-	state := xdsclient.SetClient(resolver.State{ServiceConfig: scpr}, xdsClient)
-	r.InitialState(state)
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
 
 	// Create a ClientConn with the specified transport credentials.
 	cc, err := grpc.NewClient(r.Scheme()+":///test.service", grpc.WithTransportCredentials(clientCreds), grpc.WithResolvers(r))
@@ -259,14 +171,8 @@ func verifySecurityInformationFromPeer(t *testing.T, pr *peer.Peer, wantSecLevel
 
 // Tests the case where xDS credentials are not in use, but the cds LB policy
 // receives a Cluster update with security configuration. Verifies that the
-// security configuration is not parsed by the cds LB policy by looking at the
-// xDS handshake info passed to NewSubConn.
+// connection between client and server is insecure.
 func (s) TestSecurityConfigWithoutXDSCreds(t *testing.T) {
-	// Register a wrapped cds LB policy for the duration of this test that writes
-	// the xDS handshake info passed to NewSubConn onto the given channel.
-	handshakeInfoCh := make(chan *xdscredsinternal.HandshakeInfo, 1)
-	registerWrappedCDSPolicyWithNewSubConnOverride(t, handshakeInfoCh)
-
 	// Spin up an xDS management server.
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
@@ -278,14 +184,15 @@ func (s) TestSecurityConfigWithoutXDSCreds(t *testing.T) {
 	// insecure credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, insecure.NewCredentials(), nil)
 
-	// Configure cluster and endpoints resources in the management server. The
+	// Configure default resources in the management server. The
 	// cluster resource is configured to return security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -293,34 +200,19 @@ func (s) TestSecurityConfigWithoutXDSCreds(t *testing.T) {
 	}
 
 	// Verify that a successful RPC can be made.
+	peer := &peer.Peer{}
 	client := testgrpc.NewTestServiceClient(cc)
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true), grpc.Peer(peer)); err != nil {
 		t.Fatalf("EmptyCall() failed: %v", err)
 	}
 
-	// Ensure that the xDS handshake info passed to NewSubConn is empty.
-	var gotHI *xdscredsinternal.HandshakeInfo
-	select {
-	case gotHI = <-handshakeInfoCh:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting to read handshake info passed to NewSubConn")
-	}
-	wantHI := xdscredsinternal.NewHandshakeInfo(nil, nil, nil, false)
-	if !cmp.Equal(gotHI, wantHI) {
-		t.Fatalf("NewSubConn got handshake info %+v, want %+v", gotHI, wantHI)
-	}
+	verifySecurityInformationFromPeer(t, peer, e2e.SecurityLevelNone)
 }
 
 // Tests the case where xDS credentials are in use, but the cds LB policy
 // receives a Cluster update without security configuration. Verifies that the
-// xDS handshake info passed to NewSubConn specified the use of fallback
-// credentials.
+// connection between client and server is insecure.
 func (s) TestNoSecurityConfigWithXDSCreds(t *testing.T) {
-	// Register a wrapped cds LB policy for the duration of this test that writes
-	// the xDS handshake info passed to NewSubConn onto the given channel.
-	handshakeInfoCh := make(chan *xdscredsinternal.HandshakeInfo, 1)
-	registerWrappedCDSPolicyWithNewSubConnOverride(t, handshakeInfoCh)
-
 	// Spin up an xDS management server.
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
@@ -332,14 +224,14 @@ func (s) TestNoSecurityConfigWithXDSCreds(t *testing.T) {
 	// insecure credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), nil)
 
-	// Configure cluster and endpoints resources in the management server. The
+	// Configure default resources in the management server. The
 	// cluster resource is not configured to return any security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -347,25 +239,12 @@ func (s) TestNoSecurityConfigWithXDSCreds(t *testing.T) {
 	}
 
 	// Verify that a successful RPC can be made.
+	peer := &peer.Peer{}
 	client := testgrpc.NewTestServiceClient(cc)
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true), grpc.Peer(peer)); err != nil {
 		t.Fatalf("EmptyCall() failed: %v", err)
 	}
-
-	// Ensure that the xDS handshake info passed to NewSubConn is empty.
-	var gotHI *xdscredsinternal.HandshakeInfo
-	select {
-	case gotHI = <-handshakeInfoCh:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting to read handshake info passed to NewSubConn")
-	}
-	wantHI := xdscredsinternal.NewHandshakeInfo(nil, nil, nil, false)
-	if !cmp.Equal(gotHI, wantHI) {
-		t.Fatalf("NewSubConn got handshake info %+v, want %+v", gotHI, wantHI)
-	}
-	if !gotHI.UseFallbackCreds() {
-		t.Fatal("NewSubConn got handshake info that does not specify the use of fallback creds")
-	}
+	verifySecurityInformationFromPeer(t, peer, e2e.SecurityLevelNone)
 }
 
 // Tests the case where the security config returned by the management server
@@ -395,11 +274,13 @@ func (s) TestSecurityConfigNotFoundInBootstrap(t *testing.T) {
 
 	// Configure a cluster resource that contains security configuration, in the
 	// management server.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       port,
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -458,15 +339,17 @@ func (s) TestCertproviderStoreError(t *testing.T) {
 	}
 
 	// Create a grpc channel with xDS creds.
-	cc, _ := setupForSecurityTests(t, bootstrapContents, xdsClientCredsWithInsecureFallback(t), nil)
+	cc, serverAddress := setupForSecurityTests(t, bootstrapContents, xdsClientCredsWithInsecureFallback(t), nil)
 
 	// Configure a cluster resource that contains security configuration, in the
 	// management server.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -493,14 +376,15 @@ func (s) TestGoodSecurityConfig(t *testing.T) {
 	// credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), tlsServerCreds(t))
 
-	// Configure cluster and endpoints resources in the management server. The
+	// Configure default resources in the management server. The
 	// cluster resource is configured to return security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -535,8 +419,7 @@ func (s) TestSecurityConfigUpdate_BadToGood(t *testing.T) {
 	// credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), tlsServerCreds(t))
 
-	// Configure cluster and endpoints resources in the management server. The
-	// cluster resource contains security configuration with a certificate
+	// The cluster resource contains security configuration with a certificate
 	// provider instance that is missing in the bootstrap configuration.
 	cluster := e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)
 	cluster.TransportSocket = &v3corepb.TransportSocket{
@@ -554,10 +437,11 @@ func (s) TestSecurityConfigUpdate_BadToGood(t *testing.T) {
 		},
 	}
 	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{cluster},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, serverAddress)})},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -572,8 +456,10 @@ func (s) TestSecurityConfigUpdate_BadToGood(t *testing.T) {
 	// configuration.
 	resources = e2e.UpdateOptions{
 		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterName)},
 		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
+		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, serverAddress)})},
 		SkipValidation: true,
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -607,14 +493,15 @@ func (s) TestSecurityConfigUpdate_GoodToFallback(t *testing.T) {
 	// credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), tlsServerCreds(t))
 
-	// Configure cluster and endpoints resources in the management server. The
+	// Configure default resources in the management server. The
 	// cluster resource is configured to return security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -636,12 +523,13 @@ func (s) TestSecurityConfigUpdate_GoodToFallback(t *testing.T) {
 	// Update the resources in the management server to contain no security
 	// configuration. This should result in the use of fallback credentials,
 	// which is insecure in our case.
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, insecureServer.Address)})},
-		SkipValidation: true,
-	}
+	resources = e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, insecureServer.Address),
+		SecLevel:   e2e.SecurityLevelNone,
+	})
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -686,14 +574,15 @@ func (s) TestSecurityConfigUpdate_GoodToBad(t *testing.T) {
 	// credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), tlsServerCreds(t))
 
-	// Configure cluster and endpoints resources in the management server. The
+	// Configure default resources in the management server. The
 	// cluster resource is configured to return security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelMTLS)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelMTLS,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -708,11 +597,10 @@ func (s) TestSecurityConfigUpdate_GoodToBad(t *testing.T) {
 	}
 	verifySecurityInformationFromPeer(t, peer, e2e.SecurityLevelMTLS)
 
-	// Configure cluster and endpoints resources in the management server. The
-	// cluster resource contains security configuration with a certificate
+	// The cluster resource contains security configuration with a certificate
 	// provider instance that is missing in the bootstrap configuration.
-	cluster := e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)
-	cluster.TransportSocket = &v3corepb.TransportSocket{
+	resources.Clusters[0] = e2e.DefaultCluster(resources.Clusters[0].Name, resources.Endpoints[0].ClusterName, e2e.SecurityLevelNone)
+	resources.Clusters[0].TransportSocket = &v3corepb.TransportSocket{
 		Name: "envoy.transport_sockets.tls",
 		ConfigType: &v3corepb.TransportSocket_TypedConfig{
 			TypedConfig: testutils.MarshalAny(t, &v3tlspb.UpstreamTlsContext{
@@ -726,12 +614,6 @@ func (s) TestSecurityConfigUpdate_GoodToBad(t *testing.T) {
 			}),
 		},
 	}
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{cluster},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -791,14 +673,15 @@ func (s) TestSystemRootCertsSecurityConfig(t *testing.T) {
 	// credentials.
 	cc, serverAddress := setupForSecurityTests(t, bc, xdsClientCredsWithInsecureFallback(t), tlsServerCreds(t))
 
-	// Configure cluster and endpoints resources in the management server. The
-	// cluster resource is configured to return security configuration.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelTLSWithSystemRootCerts)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, serverAddress)})},
-		SkipValidation: true,
-	}
+	// Configure default resources in the management server. The cluster
+	// resource is configured to return security configuration.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, serverAddress),
+		SecLevel:   e2e.SecurityLevelTLSWithSystemRootCerts,
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
diff --git a/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go b/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go
index ca13655d..3bd1a49c 100644
--- a/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go
@@ -55,19 +55,27 @@ import (
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
 
-	_ "google.golang.org/grpc/balancer/ringhash" // Register the ring_hash LB policy
+	_ "google.golang.org/grpc/balancer/ringhash"              // Register the ring_hash LB policy
+	_ "google.golang.org/grpc/internal/xds/httpfilter/router" // Register the router filter.
+	_ "google.golang.org/grpc/internal/xds/resolver"          // Register the xds resolver
 )
 
 const (
+	target                  = "test.service"
+	routeName               = "test_route"
 	clusterName             = "cluster1"
 	edsClusterName          = clusterName + "-eds"
 	dnsClusterName          = clusterName + "-dns"
 	serviceName             = "service1"
 	dnsHostName             = "dns_host"
+	host                    = "localhost"
+	port                    = 8080
 	dnsPort                 = uint32(8080)
 	defaultTestTimeout      = 5 * time.Second
 	defaultTestShortTimeout = 10 * time.Millisecond // For events expected to *not* happen.
@@ -152,38 +160,8 @@ func registerWrappedClusterResolverPolicy(t *testing.T) (chan serviceconfig.Load
 	return lbCfgCh, resolverErrCh, exitIdleCh, closeCh
 }
 
-// Registers a wrapped cds LB policy for the duration of this test that retains
-// all the functionality of the original cds LB policy, but makes the newly
-// built policy available to the test to directly invoke any balancer methods.
-//
-// Returns a channel on which the newly built cds LB policy is written to.
-func registerWrappedCDSPolicy(t *testing.T) chan balancer.Balancer {
-	cdsBuilder := balancer.Get(cdsName)
-	internal.BalancerUnregister(cdsBuilder.Name())
-	cdsBalancerCh := make(chan balancer.Balancer, 1)
-	stub.Register(cdsBuilder.Name(), stub.BalancerFuncs{
-		Init: func(bd *stub.BalancerData) {
-			bal := cdsBuilder.Build(bd.ClientConn, bd.BuildOptions)
-			bd.ChildBalancer = bal
-			cdsBalancerCh <- bal
-		},
-		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
-			return cdsBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
-		},
-		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
-			return bd.ChildBalancer.UpdateClientConnState(ccs)
-		},
-		Close: func(bd *stub.BalancerData) {
-			bd.ChildBalancer.Close()
-		},
-	})
-	t.Cleanup(func() { balancer.Register(cdsBuilder) })
-
-	return cdsBalancerCh
-}
-
 // Performs the following setup required for tests:
-//   - Spins up an xDS management server and and the provided onStreamRequest
+//   - Spins up an xDS management server and the provided onStreamRequest
 //     function is set to be called for every incoming request on the ADS stream.
 //   - Creates an xDS client talking to this management server
 //   - Creates a manual resolver that configures the cds LB policy as the
@@ -196,7 +174,7 @@ func registerWrappedCDSPolicy(t *testing.T) chan balancer.Balancer {
 //   - the grpc channel to the test backend service
 //   - the manual resolver configured on the channel
 //   - the xDS client used the grpc channel
-func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient) {
+func setupWithManagementServerAndManualResolver(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient) {
 	t.Helper()
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
 		Listener:        lis,
@@ -236,7 +214,7 @@ func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest f
 
 	cc, err := grpc.NewClient(r.Scheme()+":///test.service", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
 	if err != nil {
-		t.Fatalf("grpc.NewClient(%q) = %v", lis.Addr().String(), err)
+		t.Fatalf("grpc.NewClient(\"%s:///test.service\") = %v", r.Scheme(), err)
 	}
 	cc.Connect()
 	t.Cleanup(func() { cc.Close() })
@@ -244,6 +222,48 @@ func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest f
 	return mgmtServer, nodeID, cc, r, xdsC
 }
 
+// Performs the following setup required for tests:
+//   - Spins up an xDS management server and the provided onStreamRequest
+//     function is set to be called for every incoming request on the ADS stream.
+//   - Creates an xDS client talking to this management server
+//   - Creates a gRPC channel with grpc.NewClient that create a xds resolver.
+//
+// Returns the following:
+//   - the xDS management server
+//   - the nodeID expected by the management server
+//   - the grpc channel to the test backend service
+//   - the xDS client used the grpc channel
+func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn) {
+	t.Helper()
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener:        lis,
+		OnStreamRequest: onStreamRequest,
+		// Required for aggregate clusters as all resources cannot be requested
+		// at once.
+		AllowResourceSubset: true,
+	})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	r, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", target), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(\"xds:///%s\") = %v", target, err)
+	}
+	cc.Connect()
+	t.Cleanup(func() { cc.Close() })
+
+	return mgmtServer, nodeID, cc
+}
+
 // Helper function to compare the load balancing configuration received on the
 // channel with the expected one. Both configs are marshalled to JSON and then
 // compared.
@@ -308,7 +328,7 @@ func (s) TestConfigurationUpdate_Success(t *testing.T) {
 		}
 		return nil
 	}
-	_, _, _, r, xdsClient := setupWithManagementServer(t, nil, onStreamReq)
+	_, _, _, r, xdsClient := setupWithManagementServerAndManualResolver(t, nil, onStreamReq)
 
 	// Verify that the specified cluster resource is requested.
 	wantNames := []string{clusterName}
@@ -601,14 +621,16 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-			mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+			mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 			defer cancel()
 			if err := mgmtServer.Update(ctx, e2e.UpdateOptions{
-				NodeID:         nodeID,
-				Clusters:       []*v3clusterpb.Cluster{test.clusterResource},
-				SkipValidation: true,
+				NodeID:    nodeID,
+				Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+				Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, test.clusterResource.Name)},
+				Clusters:  []*v3clusterpb.Cluster{test.clusterResource},
+				Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 			}); err != nil {
 				t.Fatal(err)
 			}
@@ -625,7 +647,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 // balancing configuration pushed to the child is as expected.
 func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
 	lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil)
 
 	clusterResource := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
@@ -652,9 +674,11 @@ func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{clusterResource},
-		SkipValidation: true,
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(target, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, target, clusterResource.Name)},
+		Clusters:  []*v3clusterpb.Cluster{clusterResource},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})},
 	}); err != nil {
 		t.Fatal(err)
 	}
@@ -674,7 +698,7 @@ func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
 //     continue using the previous good update.
 func (s) TestClusterUpdate_Failure(t *testing.T) {
 	_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
 	defer cancel()
 	cdsResourceCanceledCh := make(chan struct{}, 1)
 	onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
@@ -688,18 +712,19 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 		}
 		return nil
 	}
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, onStreamReq)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, onStreamReq)
 
 	// Configure the management server to return a cluster resource that
 	// contains a config_source_specifier for the `lrs_server` field which is not
 	// set to `self`, and hence is expected to be NACKed by the client.
-	cluster := e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)
-	cluster.LrsServer = &v3corepb.ConfigSource{ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{}}
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{cluster},
-		SkipValidation: true,
-	}
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       port,
+	})
+	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{}}
+
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -728,13 +753,13 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 	server := stubserver.StartTestService(t, nil)
 	t.Cleanup(server.Stop)
 
-	// Configure cluster and endpoints resources in the management server.
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	// Configure correct cluster and endpoints resources in the management server.
+	resources = e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, server.Address),
+	})
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -745,12 +770,8 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 	}
 
 	// Send the bad cluster resource again.
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{cluster},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{}}
+
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -818,7 +839,7 @@ func (s) TestResolverError(t *testing.T) {
 		}
 		return nil
 	}
-	mgmtServer, nodeID, cc, r, _ := setupWithManagementServer(t, lis, onStreamReq)
+	mgmtServer, nodeID, cc, r, _ := setupWithManagementServerAndManualResolver(t, lis, onStreamReq)
 
 	// Grab the wrapped connection from the listener wrapper. This will be used
 	// to verify the connection is closed.
@@ -872,7 +893,7 @@ func (s) TestResolverError(t *testing.T) {
 	resources := e2e.UpdateOptions{
 		NodeID:         nodeID,
 		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
+		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{testutils.ParsePort(t, server.Address)})},
 		SkipValidation: true,
 	}
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -950,6 +971,83 @@ func (s) TestResolverError(t *testing.T) {
 	}
 }
 
+// Tests the scenario for resolver errors: when a resolver resource not found
+// error is received when the LDS resource is removed with a previous good
+// update from the management server, the cds LB policy is expected to push the
+// error down the child policy and put the channel in TRANSIENT_FAILURE. It is
+// also expected to cancel the CDS watch.
+func (s) TestResourceNotFoundResolverError(t *testing.T) {
+	_, _, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	cdsResourceCanceledCh := make(chan struct{}, 1)
+	onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
+		if req.GetTypeUrl() == version.V3ClusterURL {
+			if len(req.GetResourceNames()) == 0 {
+				select {
+				case cdsResourceCanceledCh <- struct{}{}:
+				case <-ctx.Done():
+				}
+			}
+		}
+		return nil
+	}
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, onStreamReq)
+
+	// Start a test service backend.
+	server := stubserver.StartTestService(t, nil)
+	t.Cleanup(server.Stop)
+
+	// Configure default resources in the management server.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, server.Address),
+	})
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that a successful RPC can be made.
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+		t.Fatalf("EmptyCall() failed: %v", err)
+	}
+
+	// Push a resource-not-found type error  by removing the LDS resource.
+	resources.Listeners = nil
+	resources.SkipValidation = true
+	mgmtServer.Update(ctx, resources)
+
+	// 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:
+	}
+
+	// Verify that the resolver error is pushed to the child policy.
+	select {
+	case <-childPolicyCloseCh:
+	case <-ctx.Done():
+		t.Fatal("Timeout when waiting for child policy to be closed")
+	}
+
+	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
+
+	// Ensure that the resolver error is propagated to the RPC caller.
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout when waiting for error from RPC")
+	default:
+		_, err := client.EmptyCall(ctx, &testpb.Empty{})
+		if err := verifyRPCError(err, codes.Unavailable, "", nodeID); err == nil {
+			return
+		}
+	}
+}
+
 // Tests scenarios involving removal of a cluster resource from the management
 // server.
 //
@@ -973,19 +1071,19 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 		}
 		return nil
 	}
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, onStreamReq)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, onStreamReq)
 
 	// Start a test service backend.
 	server := stubserver.StartTestService(t, nil)
 	t.Cleanup(server.Stop)
 
-	// Configure cluster and endpoints resources in the management server.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	// Configure default resources in the management server.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, server.Address),
+	})
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -998,7 +1096,9 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 
 	// Remove the cluster resource from the management server, triggering a
 	// resource-not-found error.
+	oldClusters := resources.Clusters
 	resources.Clusters = nil
+	resources.SkipValidation = true
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -1016,19 +1116,14 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 
 	// Ensure RPC fails with Unavailable status code and the error message is
 	// meaningful and contains the xDS node ID.
-	wantErr := fmt.Sprintf("resource %q of type %q has been removed", clusterName, "ClusterResource")
+	wantErr := fmt.Sprintf("resource %q of type %q has been removed", oldClusters[0].Name, "ClusterResource")
 	_, err := client.EmptyCall(ctx, &testpb.Empty{})
 	if err := verifyRPCError(err, codes.Unavailable, wantErr, nodeID); err != nil {
 		t.Fatal(err)
 	}
 
 	// Re-add the cluster resource to the management server.
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	resources.Clusters = oldClusters
 	if err := mgmtServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -1041,22 +1136,21 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 
 // Tests that closing the cds LB policy results in the the child policy being
 // closed.
-func (s) TestClose(t *testing.T) {
-	cdsBalancerCh := registerWrappedCDSPolicy(t)
+func TestClose(t *testing.T) {
 	_, _, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	// Start a test service backend.
 	server := stubserver.StartTestService(t, nil)
 	t.Cleanup(server.Stop)
 
-	// Configure cluster and endpoints resources in the management server.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	// Configure resources in the management server.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, server.Address),
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -1069,14 +1163,8 @@ func (s) TestClose(t *testing.T) {
 		t.Fatalf("EmptyCall() failed: %v", err)
 	}
 
-	// Retrieve the cds LB policy and close it.
-	var cdsBal balancer.Balancer
-	select {
-	case cdsBal = <-cdsBalancerCh:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for cds LB policy to be created")
-	}
-	cdsBal.Close()
+	// Close the gRPC ClientConn, which will close the cds LB policy.
+	cc.Close()
 
 	// Wait for the child policy to be closed.
 	select {
@@ -1089,21 +1177,20 @@ func (s) TestClose(t *testing.T) {
 // Tests that calling ExitIdle on the cds LB policy results in the call being
 // propagated to the child policy.
 func (s) TestExitIdle(t *testing.T) {
-	cdsBalancerCh := registerWrappedCDSPolicy(t)
 	_, _, exitIdleCh, _ := registerWrappedClusterResolverPolicy(t)
-	mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)
+	mgmtServer, nodeID, cc := setupWithManagementServer(t, nil, nil)
 
 	// Start a test service backend.
 	server := stubserver.StartTestService(t, nil)
 	t.Cleanup(server.Stop)
 
-	// Configure cluster and endpoints resources in the management server.
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Clusters:       []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)},
-		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
-		SkipValidation: true,
-	}
+	// Configure resources in the management server.
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: target,
+		NodeID:     nodeID,
+		Host:       host,
+		Port:       testutils.ParsePort(t, server.Address),
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if err := mgmtServer.Update(ctx, resources); err != nil {
@@ -1116,14 +1203,8 @@ func (s) TestExitIdle(t *testing.T) {
 		t.Fatalf("EmptyCall() failed: %v", err)
 	}
 
-	// Retrieve the cds LB policy and call ExitIdle() on it.
-	var cdsBal balancer.Balancer
-	select {
-	case cdsBal = <-cdsBalancerCh:
-	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for cds LB policy to be created")
-	}
-	cdsBal.ExitIdle()
+	// Connect to gRPC channel that will call ExitIdle on the cds LB policy.
+	cc.Connect()
 
 	// Wait for ExitIdle to be called on the child policy.
 	select {
diff --git a/internal/xds/balancer/clusterimpl/balancer_test.go b/internal/xds/balancer/clusterimpl/balancer_test.go
index 53f2debf..b8b22124 100644
--- a/internal/xds/balancer/clusterimpl/balancer_test.go
+++ b/internal/xds/balancer/clusterimpl/balancer_test.go
@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 	"sync"
 	"testing"
@@ -64,7 +63,6 @@ const (
 
 var (
 	testBackendEndpoints = []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1:1"}}}}
-	cmpOpts              = cmp.Options{cmpopts.EquateEmpty(), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{}), sortDataSlice}
 	toleranceCmpOpt      = cmp.Options{cmpopts.EquateApprox(0, 1e-5), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{})}
 )
 
@@ -279,345 +277,6 @@ func init() {
 	NewRandomWRR = testutils.NewTestWRR
 }
 
-var sortDataSlice = cmp.Transformer("SortDataSlice", func(in []*loadData) []*loadData {
-	out := append([]*loadData(nil), in...) // Copy input to avoid mutating it
-	sort.Slice(out,
-		func(i, j int) bool {
-			if out[i].cluster < out[j].cluster {
-				return true
-			}
-			if out[i].cluster == out[j].cluster {
-				return out[i].service < out[j].service
-			}
-			return false
-		},
-	)
-	return out
-})
-
-func verifyLoadStoreData(wantStoreData, gotStoreData *loadData) error {
-	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpOpts); diff != "" {
-		return fmt.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
-	}
-	return nil
-}
-
-// TestDropByCategory verifies that the balancer correctly drops the picks, and
-// that the drops are reported.
-func (s) TestDropByCategory(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	const (
-		dropReason      = "test-dropping-category"
-		dropNumerator   = 1
-		dropDenominator = 2
-	)
-	testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{
-		URI:          "trafficdirector.googleapis.com:443",
-		ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}},
-	})
-	if err != nil {
-		t.Fatalf("Failed to create LRS server config for testing: %v", err)
-	}
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig,
-			DropCategories: []DropConfig{{
-				Category:           dropReason,
-				RequestsPerMillion: million * dropNumerator / dropDenominator,
-			}},
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	got, err := xdsC.WaitForReportLoad(ctx)
-	if err != nil {
-		t.Fatalf("xdsClient.ReportLoad failed with error: %v", err)
-	}
-	if got.Server != testLRSServerConfig {
-		t.Fatalf("xdsClient.ReportLoad called with {%q}: want {%q}", got.Server, testLRSServerConfig)
-	}
-
-	sc1 := <-cc.NewSubConnCh
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	// This should get the connecting picker.
-	if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	// Test pick with one backend.
-
-	testClusterLoadReporter := &testLoadReporter{cluster: testClusterName, service: testServiceName, drops: make(map[string]uint64), localityRPCCount: make(map[clients.Locality]*rpcCountData)}
-
-	const rpcCount = 24
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			// Even RPCs are dropped.
-			if i%2 == 0 {
-				if err == nil || !strings.Contains(err.Error(), "dropped") {
-					return fmt.Errorf("pick.Pick, got %v, %v, want error RPC dropped", gotSCSt, err)
-				}
-				continue
-			}
-			if err != nil || gotSCSt.SubConn != sc1 {
-				return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
-			}
-			if gotSCSt.Done == nil {
-				continue
-			}
-			// Fail 1/4th of the requests that are not dropped.
-			if i%8 == 1 {
-				gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
-			} else {
-				gotSCSt.Done(balancer.DoneInfo{})
-			}
-		}
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	// Dump load data from the store and compare with expected counts.
-	const dropCount = rpcCount * dropNumerator / dropDenominator
-	wantStatsData0 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: dropCount,
-		drops:      map[string]uint64{dropReason: dropCount},
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: (rpcCount - dropCount) * 3 / 4,
-				errored:   (rpcCount - dropCount) / 4,
-				issued:    rpcCount - dropCount,
-			}},
-		},
-	}
-
-	gotStatsData0 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData0, gotStatsData0); err != nil {
-		t.Fatal(err)
-	}
-
-	// Send an update with new drop configs.
-	const (
-		dropReason2      = "test-dropping-category-2"
-		dropNumerator2   = 1
-		dropDenominator2 = 4
-	)
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig,
-			DropCategories: []DropConfig{{
-				Category:           dropReason2,
-				RequestsPerMillion: million * dropNumerator2 / dropDenominator2,
-			}},
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			// Even RPCs are dropped.
-			if i%4 == 0 {
-				if err == nil || !strings.Contains(err.Error(), "dropped") {
-					return fmt.Errorf("pick.Pick, got %v, %v, want error RPC dropped", gotSCSt, err)
-				}
-				continue
-			}
-			if err != nil || gotSCSt.SubConn != sc1 {
-				return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
-			}
-			if gotSCSt.Done != nil {
-				gotSCSt.Done(balancer.DoneInfo{})
-			}
-		}
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	const dropCount2 = rpcCount * dropNumerator2 / dropDenominator2
-	wantStatsData1 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: dropCount2,
-		drops:      map[string]uint64{dropReason2: dropCount2},
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: rpcCount - dropCount2,
-				issued:    rpcCount - dropCount2,
-			}},
-		},
-	}
-
-	gotStatsData1 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData1, gotStatsData1); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// TestDropCircuitBreaking verifies that the balancer correctly drops the picks
-// due to circuit breaking, and that the drops are reported.
-func (s) TestDropCircuitBreaking(t *testing.T) {
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	var maxRequest uint32 = 50
-	testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{
-		URI:          "trafficdirector.googleapis.com:443",
-		ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}},
-	})
-	if err != nil {
-		t.Fatalf("Failed to create LRS server config for testing: %v", err)
-	}
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:               testClusterName,
-			EDSServiceName:        testServiceName,
-			LoadReportingServer:   testLRSServerConfig,
-			MaxConcurrentRequests: &maxRequest,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	got, err := xdsC.WaitForReportLoad(ctx)
-	if err != nil {
-		t.Fatalf("xdsClient.ReportLoad failed with error: %v", err)
-	}
-	if got.Server != testLRSServerConfig {
-		t.Fatalf("xdsClient.ReportLoad called with {%q}: want {%q}", got.Server, testLRSServerConfig)
-	}
-
-	sc1 := <-cc.NewSubConnCh
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	// This should get the connecting picker.
-	if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	// Test pick with one backend.
-	testClusterLoadReporter := &testLoadReporter{cluster: testClusterName, service: testServiceName, drops: make(map[string]uint64), localityRPCCount: make(map[clients.Locality]*rpcCountData)}
-	const rpcCount = 100
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		dones := []func(){}
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			if i < 50 && err != nil {
-				return fmt.Errorf("The first 50%% picks should be non-drops, got error %v", err)
-			} else if i > 50 && err == nil {
-				return fmt.Errorf("The second 50%% picks should be drops, got error <nil>")
-			}
-			dones = append(dones, func() {
-				if gotSCSt.Done != nil {
-					gotSCSt.Done(balancer.DoneInfo{})
-				}
-			})
-		}
-		for _, done := range dones {
-			done()
-		}
-
-		dones = []func(){}
-		// Pick without drops.
-		for i := 0; i < 50; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			if err != nil {
-				t.Errorf("The third 50%% picks should be non-drops, got error %v", err)
-			}
-			dones = append(dones, func() {
-				if gotSCSt.Done != nil {
-					// Fail these requests to test error counts in the load
-					// report.
-					gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
-				}
-			})
-		}
-		for _, done := range dones {
-			done()
-		}
-
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	// Dump load data from the store and compare with expected counts.
-	wantStatsData0 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: uint64(maxRequest),
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: uint64(rpcCount - maxRequest),
-				errored:   50,
-				issued:    uint64(rpcCount - maxRequest + 50),
-			}},
-		},
-	}
-
-	gotStatsData0 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData0, gotStatsData0); err != nil {
-		t.Fatal(err)
-	}
-}
-
 // TestPickerUpdateAfterClose covers the case where a child policy sends a
 // picker update after the cluster_impl policy is closed. Because picker updates
 // are handled in the run() goroutine, which exits before Close() returns, we
diff --git a/internal/xds/balancer/clusterimpl/tests/balancer_test.go b/internal/xds/balancer/clusterimpl/tests/balancer_test.go
index 7cfd3992..2aecce0f 100644
--- a/internal/xds/balancer/clusterimpl/tests/balancer_test.go
+++ b/internal/xds/balancer/clusterimpl/tests/balancer_test.go
@@ -21,12 +21,14 @@ package clusterimpl_test
 import (
 	"context"
 	"fmt"
+	"math"
 	"net"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -41,6 +43,7 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
@@ -85,13 +88,12 @@ func (s) TestConfigUpdateWithSameLoadReportingServerConfig(t *testing.T) {
 	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	var resolverBuilder resolver.Builder
-	var err error
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
-		if err != nil {
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
 	}
 
 	// Start a server backend exposing the test service.
@@ -194,13 +196,12 @@ func (s) TestLoadReportingPickFirstMultiLocality(t *testing.T) {
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	var resolverBuilder resolver.Builder
-	var err error
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
-		if err != nil {
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
 	}
 
 	// Start two server backends exposing the test service.
@@ -360,8 +361,8 @@ func waitForSuccessfulLoadReport(ctx context.Context, lrsServer *fakeserver.Serv
 
 // Tests that circuit breaking limits RPCs E2E.
 func (s) TestCircuitBreaking(t *testing.T) {
-	// Create an xDS management server that serves ADS requests.
-	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+	// Create an xDS management server that serves ADS and LRS requests.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
 
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
@@ -369,20 +370,16 @@ func (s) TestCircuitBreaking(t *testing.T) {
 	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	var resolverBuilder resolver.Builder
-	var err error
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
-		if err != nil {
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
 	}
 
 	// Start a server backend exposing the test service.
 	f := &stubserver.StubServer{
-		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
 		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
 			for {
 				if _, err := stream.Recv(); err != nil {
@@ -394,7 +391,8 @@ func (s) TestCircuitBreaking(t *testing.T) {
 	server := stubserver.StartTestService(t, f)
 	defer server.Stop()
 
-	// Configure the xDS management server with default resources.
+	// Configure xDS resources on the management server with a circuit breaking
+	// policy that limits the maximum number of concurrent requests to 3.
 	const serviceName = "my-test-xds-service"
 	const maxRequests = 3
 	resources := e2e.DefaultClientResources(e2e.ResourceParams{
@@ -411,6 +409,11 @@ func (s) TestCircuitBreaking(t *testing.T) {
 			},
 		},
 	}
+	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{
+		ConfigSourceSpecifier: &v3corepb.ConfigSource_Self{
+			Self: &v3corepb.SelfConfigSource{},
+		},
+	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -436,21 +439,264 @@ func (s) TestCircuitBreaking(t *testing.T) {
 
 	// Since we are at the max, new streams should fail.  It's possible some are
 	// allowed due to inherent raciness in the tracking, however.
-	for i := 0; i < 100; i++ {
-		stream, err := client.FullDuplexCall(ctx)
-		if status.Code(err) == codes.Unavailable {
-			return
+	const droppedRPCCount = 100
+	for i := 0; i < droppedRPCCount; i++ {
+		if ctx.Err() != nil {
+			t.Fatalf("Context error: %v", ctx.Err())
 		}
-		if err == nil {
-			// Terminate the stream (the server immediately exits upon a client
-			// CloseSend) to ensure we never go over the limit.
+		if _, err := client.FullDuplexCall(ctx); status.Code(err) != codes.Unavailable {
+			t.Fatalf("client.FullDuplexCall() returned status %q, want %q", status.Code(err), codes.Unavailable)
+		}
+	}
+
+	if _, err = mgmtServer.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Failure when waiting for an LRS stream to be opened: %v", err)
+	}
+
+	if _, err = mgmtServer.LRSServer.LRSRequestChan.Receive(ctx); err != nil {
+		t.Fatalf("Failure waiting for initial LRS request: %v", err)
+	}
+
+	mgmtServer.LRSServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: durationpb.New(10 * time.Millisecond),
+		},
+	}
+
+	select {
+	case req := <-mgmtServer.LRSServer.LRSRequestChan.C:
+		clusterStats := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+		if l := len(clusterStats); l != 1 {
+			t.Fatalf("Received load for %d clusters, want 1", l)
+		}
+		clusterStats[0].LoadReportInterval = nil
+		wantLoad := &v3endpointpb.ClusterStats{
+			ClusterName:        "cluster-my-test-xds-service",
+			ClusterServiceName: "endpoints-my-test-xds-service",
+			UpstreamLocalityStats: []*v3endpointpb.UpstreamLocalityStats{
+				{
+					Locality:                &v3corepb.Locality{Region: "region-1", Zone: "zone-1", SubZone: "subzone-1"},
+					TotalIssuedRequests:     maxRequests,
+					TotalRequestsInProgress: maxRequests,
+				},
+			},
+			TotalDroppedRequests: droppedRPCCount,
+		}
+		if diff := cmp.Diff(wantLoad, clusterStats[0], protocmp.Transform()); diff != "" {
+			t.Errorf("Failed with unexpected diff (-want +got):\n%s", diff)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timeout while waiting for load report on LRS stream: %v", ctx.Err())
+	}
+}
+
+// ComputeIdealNumRpcs calculates the ideal number of RPCs to send to test
+// probabilistic behaviors.
+//
+// It's based on the idea of a binomial distribution approximated by a normal
+// distribution. The function aims to find a number of RPCs such that
+// the observed probability is within a certain error_tolerance of the expected
+// probability (p).
+func computeIdealNumRpcs(p, errorTolerance float64) uint64 {
+	return uint64(math.Ceil(p * (1 - p) * 5.00 * 5.00 / errorTolerance / errorTolerance))
+}
+
+func verifyDropRateByCategory(ctx context.Context, client testgrpc.TestServiceClient, rpcCount int, lrsServer *fakeserver.Server, wantCluster, wantDropCategory string, wantDropRate, errorTolerance float64) error {
+	for i := 0; i < rpcCount; i++ {
+		if ctx.Err() != nil {
+			return fmt.Errorf("context error: %v", ctx.Err())
+		}
+		stream, err := client.FullDuplexCall(ctx)
+		switch {
+		case err == nil:
+			// Close stream to avoid exceeding limits.
 			stream.CloseSend()
-			stream.Recv()
+		case status.Code(err) == codes.Unavailable && strings.Contains(err.Error(), "RPC is dropped"):
+			continue
+		default:
+			return fmt.Errorf("client.FullDuplexCall(_) failed with unexpected error = %v", err)
 		}
-		time.Sleep(10 * time.Millisecond)
 	}
 
-	t.Fatalf("RPCs unexpectedly allowed beyond circuit breaking maximum")
+	for {
+		if ctx.Err() != nil {
+			return fmt.Errorf("timeout when waiting for load stats")
+		}
+		req, err := lrsServer.LRSRequestChan.Receive(ctx)
+		if err != nil {
+			continue
+		}
+		loadStats := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest).ClusterStats
+		if len(loadStats) != 1 || loadStats[0].ClusterName != wantCluster {
+			continue
+		}
+		if loadStats[0].ClusterName != wantCluster {
+			return fmt.Errorf("Received stats for unexpected cluster got: %q, want: %q", loadStats[0].ClusterName, wantCluster)
+		}
+
+		for _, cs := range loadStats {
+			found := false
+			for _, dr := range cs.DroppedRequests {
+				if dr.Category == wantDropCategory {
+					found = true
+					gotRPCDropRate := float64(dr.DroppedCount) / float64(rpcCount)
+					if (math.Trunc(math.Abs(gotRPCDropRate-wantDropRate)*100) / 100) > errorTolerance {
+						return fmt.Errorf("Drop rate goes out of errortolerance got: %v, want: %v, totalDroppedRequest: %v, totalIssuesRequest: %v", (math.Trunc(math.Abs(gotRPCDropRate-wantDropRate)*100) / 100), errorTolerance, cs.TotalDroppedRequests, cs.UpstreamLocalityStats[0].TotalIssuedRequests)
+					}
+				}
+			}
+			if !found {
+				continue
+			}
+		}
+		break
+	}
+	return nil
+}
+
+// TestDropByCategory verifies that the balancer correctly drops the picks, and
+// that the drops are reported.
+func (s) TestDropByCategory(t *testing.T) {
+	// Create an xDS management server that serves ADS and LRS requests.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+	testutils.CreateBootstrapFileForTesting(t, bc)
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+
+	// Start a server backend exposing the test service.
+	f := &stubserver.StubServer{
+		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+			for {
+				if _, err := stream.Recv(); err != nil {
+					return err
+				}
+			}
+		},
+	}
+	server := stubserver.StartTestService(t, f)
+	defer server.Stop()
+
+	// Configure xDS resources on the management server with drops
+	// configuration that drops one RPC for every 50 RPCs made.
+	const (
+		dropReason      = "test-dropping-category"
+		dropNumerator   = 1
+		dropDenominator = 50
+		serviceName     = "my-test-xds-service"
+		errorTolerance  = 0.05
+	)
+	wantRPCDropRate := float64(dropNumerator) / float64(dropDenominator)
+	rpcCount := computeIdealNumRpcs(wantRPCDropRate, errorTolerance)
+	t.Logf("About to send %d RPCs to test drop rate of %v", rpcCount, wantRPCDropRate)
+
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: serviceName,
+		NodeID:     nodeID,
+		Host:       "localhost",
+		Port:       testutils.ParsePort(t, server.Address),
+	})
+	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{
+		ConfigSourceSpecifier: &v3corepb.ConfigSource_Self{
+			Self: &v3corepb.SelfConfigSource{},
+		},
+	}
+	resources.Endpoints = []*v3endpointpb.ClusterLoadAssignment{
+		e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+			ClusterName: "endpoints-" + serviceName,
+			Host:        "localhost",
+			Localities: []e2e.LocalityOptions{
+				{
+					Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, server.Address)}}},
+					Weight:   1,
+				},
+			},
+			DropPercents: map[string]int{
+				dropReason: dropNumerator * 100 / dropDenominator,
+			},
+		})}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a ClientConn and make a successful RPC.
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolverBuilder))
+	if err != nil {
+		t.Fatalf("failed to dial local test server: %v", err)
+	}
+	defer cc.Close()
+
+	// Ensure the gRPC channel is READY before issuing RPCs to get accurate
+	// drop count.
+	cc.Connect()
+	client := testgrpc.NewTestServiceClient(cc)
+	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
+
+	if _, err = mgmtServer.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Failure when waiting for an LRS stream to be opened: %v", err)
+	}
+
+	resp := fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			Clusters:              []string{resources.Clusters[0].Name},
+			LoadReportingInterval: durationpb.New(50 * time.Millisecond),
+		},
+	}
+	mgmtServer.LRSServer.LRSResponseChan <- &resp
+
+	if err := verifyDropRateByCategory(ctx, client, int(rpcCount), mgmtServer.LRSServer, resources.Clusters[0].Name, dropReason, wantRPCDropRate, errorTolerance); err != nil {
+		t.Fatal(err)
+	}
+
+	// Update the drop configuration to drop 1 out of every 40 RPCs,
+	// and verify that the observed drop rate matches this new config.
+	const (
+		dropReason2      = "test-dropping-category-2"
+		dropNumerator2   = 1
+		dropDenominator2 = 40
+	)
+
+	resources.Endpoints = []*v3endpointpb.ClusterLoadAssignment{
+		e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+			ClusterName: "endpoints-" + serviceName,
+			Host:        "localhost",
+			Localities: []e2e.LocalityOptions{
+				{
+					Backends: []e2e.BackendOptions{{Ports: []uint32{testutils.ParsePort(t, server.Address)}}},
+					Weight:   1,
+				},
+			},
+			DropPercents: map[string]int{
+				dropReason2: dropNumerator2 * 100 / dropDenominator2, // drops one RPC for every 40 RPCs made
+			},
+		}),
+	}
+
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	wantRPCDropRate = float64(dropNumerator2) / float64(dropDenominator2)
+	rpcCount = computeIdealNumRpcs(wantRPCDropRate, errorTolerance)
+	t.Logf("About to send %d RPCs to test drop rate of %v", rpcCount, wantRPCDropRate)
+
+	if err := verifyDropRateByCategory(ctx, client, int(rpcCount), mgmtServer.LRSServer, resources.Clusters[0].Name, dropReason2, wantRPCDropRate, errorTolerance); err != nil {
+		t.Fatal(err)
+	}
 }
 
 // Tests that circuit breaking limits RPCs in LOGICAL_DNS clusters E2E.
@@ -464,13 +710,12 @@ func (s) TestCircuitBreakingLogicalDNS(t *testing.T) {
 	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	var resolverBuilder resolver.Builder
-	var err error
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
-		if err != nil {
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
 	}
 
 	// Start a server backend exposing the test service.
@@ -540,7 +785,7 @@ func (s) TestCircuitBreakingLogicalDNS(t *testing.T) {
 		}
 	}
 
-	// Since we are at the max, new streams should fail.  It's possible some are
+	// Since we are at the max, new streams should fail. It's possible some are
 	// allowed due to inherent raciness in the tracking, however.
 	for i := 0; i < 100; i++ {
 		stream, err := client.FullDuplexCall(ctx)
@@ -584,13 +829,12 @@ func (s) TestLRSLogicalDNS(t *testing.T) {
 	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	var resolverBuilder resolver.Builder
-	var err error
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
-		if err != nil {
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
 	}
 
 	// Start a server backend exposing the test service.
diff --git a/internal/xds/clients/xdsclient/test/ads_stream_flow_control_test.go b/internal/xds/clients/xdsclient/test/ads_stream_flow_control_test.go
index 9c417a14..83155a99 100644
--- a/internal/xds/clients/xdsclient/test/ads_stream_flow_control_test.go
+++ b/internal/xds/clients/xdsclient/test/ads_stream_flow_control_test.go
@@ -20,7 +20,6 @@ package xdsclient_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"slices"
 	"sort"
@@ -125,7 +124,6 @@ func (t *transport) NewStream(ctx context.Context, method string) (clients.Strea
 	stream := &stream{
 		stream: s,
 		recvCh: make(chan struct{}, 1),
-		doneCh: make(chan struct{}),
 	}
 	t.adsStreamCh <- stream
 
@@ -138,9 +136,7 @@ func (t *transport) Close() {
 
 type stream struct {
 	stream grpc.ClientStream
-
 	recvCh chan struct{}
-	doneCh <-chan struct{}
 }
 
 func (s *stream) Send(msg []byte) error {
@@ -150,8 +146,8 @@ func (s *stream) Send(msg []byte) error {
 func (s *stream) Recv() ([]byte, error) {
 	select {
 	case s.recvCh <- struct{}{}:
-	case <-s.doneCh:
-		return nil, errors.New("Recv() called after the test has finished")
+	case <-s.stream.Context().Done():
+		// Unblock the recv() once the stream is done.
 	}
 
 	var typedRes []byte
diff --git a/internal/xds/resolver/serviceconfig_test.go b/internal/xds/resolver/serviceconfig_test.go
index 3570ff2e..f05a7d91 100644
--- a/internal/xds/resolver/serviceconfig_test.go
+++ b/internal/xds/resolver/serviceconfig_test.go
@@ -29,7 +29,6 @@ import (
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/grpcutil"
 	iresolver "google.golang.org/grpc/internal/resolver"
-	"google.golang.org/grpc/internal/testutils"
 	_ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer" // To parse LB config
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
@@ -64,12 +63,8 @@ func (s) TestPruneActiveClusters(t *testing.T) {
 
 func (s) TestGenerateRequestHash(t *testing.T) {
 	const channelID = 12378921
-	cs := &configSelector{
-		r: &xdsResolver{
-			cc:        &testutils.ResolverClientConn{Logger: t},
-			channelID: channelID,
-		},
-	}
+	cs := &configSelector{channelID: channelID}
+
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	tests := []struct {
diff --git a/internal/xds/xdsclient/tests/dump_test.go b/internal/xds/xdsclient/tests/dump_test.go
index a476c29b..11a43c12 100644
--- a/internal/xds/xdsclient/tests/dump_test.go
+++ b/internal/xds/xdsclient/tests/dump_test.go
@@ -63,7 +63,7 @@ func makeGenericXdsConfig(typeURL, name, version string, status v3adminpb.Client
 }
 
 func checkResourceDump(ctx context.Context, want *v3statuspb.ClientStatusResponse, pool *xdsclient.Pool) error {
-	var cmpOpts = cmp.Options{
+	cmpOpts := cmp.Options{
 		protocmp.Transform(),
 		protocmp.IgnoreFields((*v3statuspb.ClientConfig_GenericXdsConfig)(nil), "last_updated"),
 		protocmp.IgnoreFields((*v3adminpb.UpdateFailureState)(nil), "last_update_attempt", "details"),
@@ -89,7 +89,7 @@ func checkResourceDump(ctx context.Context, want *v3statuspb.ClientStatusRespons
 		if diff == "" {
 			return nil
 		}
-		lastErr = fmt.Errorf("received unexpected resource dump, diff (-got, +want):\n%s, got: %s\n want:%s", diff, pretty.ToJSON(got), pretty.ToJSON(want))
+		lastErr = fmt.Errorf("received unexpected resource dump, diff (-want, +got):\n%s, got: %s\n want:%s", diff, pretty.ToJSON(got), pretty.ToJSON(want))
 	}
 	return fmt.Errorf("timeout when waiting for resource dump to reach expected state: %v", lastErr)
 }
diff --git a/internal/xds/xdsclient/tests/fallback/fallback_test.go b/internal/xds/xdsclient/tests/fallback/fallback_test.go
index b45ee7b0..5c96ea2e 100644
--- a/internal/xds/xdsclient/tests/fallback/fallback_test.go
+++ b/internal/xds/xdsclient/tests/fallback/fallback_test.go
@@ -742,14 +742,13 @@ func (s) TestFallback_OnStartup_RPCSuccess(t *testing.T) {
 	}
 }
 
-// TestXDSFallback_ThreeServerPromotion verifies that when the primary
-// management server is unavailable, the system attempts to connect to the
-// first fallback server, and if that is also down, to the second fallback
-// server. It also ensures that the system switches back to the first fallback
-// server once it becomes available again, and eventually returns to the
-// primary server when it comes back online, closing connections to the
-// fallback servers accordingly.
-func (s) TestXDSFallback_ThreeServerPromotion(t *testing.T) {
+// Test verifies that when the primary management server is unavailable, the
+// system attempts to connect to the first fallback server, and if that is also
+// down, to the second fallback server. It also ensures that the system switches
+// back to the first fallback server once it becomes available again, and
+// eventually returns to the primary server when it comes back online, closing
+// connections to the fallback servers accordingly.
+func (s) TestFallback_ThreeServerPromotion(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultFallbackTestTimeout)
 	defer cancel()
 
@@ -770,34 +769,49 @@ func (s) TestXDSFallback_ThreeServerPromotion(t *testing.T) {
 
 	// Start three test service backends.
 	backend1 := stubserver.StartTestService(t, nil)
+	backend1Port := testutils.ParsePort(t, backend1.Address)
 	defer backend1.Stop()
 	backend2 := stubserver.StartTestService(t, nil)
+	backend2Port := testutils.ParsePort(t, backend2.Address)
 	defer backend2.Stop()
 	backend3 := stubserver.StartTestService(t, nil)
+	backend3Port := testutils.ParsePort(t, backend3.Address)
 	defer backend3.Stop()
 
 	nodeID := uuid.New().String()
-	const serviceName = "my-service-fallback-xds"
+	const (
+		serviceName              = "my-service-fallback-xds"
+		primaryRouteConfigName   = "primary-route-" + serviceName
+		secondaryRouteConfigName = "secondary-route-" + serviceName
+		tertiaryRouteConfigName  = "tertiary-route-" + serviceName
+		primaryClusterName       = "primary-cluster-" + serviceName
+		secondaryClusterName     = "secondary-cluster-" + serviceName
+		tertiaryClusterName      = "tertiary-cluster-" + serviceName
+		primaryEndpointsName     = "primary-endpoints-" + serviceName
+		secondaryEndpointsName   = "secondary-endpoints-" + serviceName
+		tertiaryEndpointsName    = "tertiary-endpoints-" + serviceName
+	)
 
 	// Configure partial resources on the primary and secondary
 	// management servers.
 	primaryManagementServer.Update(ctx, e2e.UpdateOptions{
 		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, "route-p")},
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, primaryRouteConfigName)},
 	})
 	secondaryManagementServer.Update(ctx, e2e.UpdateOptions{
 		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, "route-s1")},
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, secondaryRouteConfigName)},
 	})
 
 	// Configure full resources on tertiary management server.
-	tertiaryManagementServer.Update(ctx, e2e.DefaultClientResources(e2e.ResourceParams{
-		DialTarget: serviceName,
-		NodeID:     nodeID,
-		Host:       "localhost",
-		Port:       testutils.ParsePort(t, backend3.Address),
-		SecLevel:   e2e.SecurityLevelNone,
-	}))
+	updateOpts := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, tertiaryRouteConfigName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(tertiaryRouteConfigName, serviceName, tertiaryClusterName)},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(tertiaryClusterName, tertiaryEndpointsName, e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(tertiaryEndpointsName, "localhost", []uint32{backend3Port})},
+	}
+	tertiaryManagementServer.Update(ctx, updateOpts)
 
 	// Create bootstrap configuration for all three management servers.
 	bootstrapContents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
@@ -874,15 +888,14 @@ func (s) TestXDSFallback_ThreeServerPromotion(t *testing.T) {
 
 	// Secondary1 becomes available, RPCs go to backend2.
 	secondaryLis.Restart()
-	secondaryManagementServer.Update(ctx, e2e.UpdateOptions{
+	updateOpts = e2e.UpdateOptions{
 		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, "route-s1")},
-		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig("route-s1", serviceName, "cluster-s1")},
-		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster("cluster-s1", "endpoints-s1", e2e.SecurityLevelNone)},
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{
-			e2e.DefaultEndpoint("endpoints-s1", "localhost", []uint32{testutils.ParsePort(t, backend2.Address)}),
-		},
-	})
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, secondaryRouteConfigName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(secondaryRouteConfigName, serviceName, secondaryClusterName)},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(secondaryClusterName, secondaryEndpointsName, e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(secondaryEndpointsName, "localhost", []uint32{backend2Port})},
+	}
+	secondaryManagementServer.Update(ctx, updateOpts)
 
 	secondaryConn, err := secondaryWrappedLis.NewConnCh.Receive(ctx)
 	if err != nil {
@@ -897,13 +910,14 @@ func (s) TestXDSFallback_ThreeServerPromotion(t *testing.T) {
 
 	// Primary becomes available, RPCs go to backend1.
 	primaryLis.Restart()
-	primaryManagementServer.Update(ctx, e2e.DefaultClientResources(e2e.ResourceParams{
-		DialTarget: serviceName,
-		NodeID:     nodeID,
-		Host:       "localhost",
-		Port:       testutils.ParsePort(t, backend1.Address),
-		SecLevel:   e2e.SecurityLevelNone,
-	}))
+	updateOpts = e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, primaryRouteConfigName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(primaryRouteConfigName, serviceName, primaryClusterName)},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(primaryClusterName, primaryEndpointsName, e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(primaryEndpointsName, "localhost", []uint32{backend1Port})},
+	}
+	primaryManagementServer.Update(ctx, updateOpts)
 
 	if _, err := primaryWrappedLis.NewConnCh.Receive(ctx); err != nil {
 		t.Fatalf("Timeout when waiting for new connection to primary: %v", err)