Repository: grpc/grpc-go

Issue:
Node ID should be available in RPC errors
1. Include node ID with all xdsclient errors possible

2. Audit all possible ways an xds-enabled gRPC channel can fail as a result of xdsclient problems, and ensure the xdsclient error is propagated to RPC errors


Discussion:
@dfawley: @easwars can you help scope this, please?@easwars: I suggest making the following changes:

- [x] The watch resource API call on the xDS client can fail for a bunch of reasons. We need to include the node ID in all these errors:
  - [x] A resource type implementation was already registered.
  - [x] Authority in the resource name is not found in the bootstrap configuration.
  - [x] Creation of transport to the management server fails.
- [x] A resource is NACKed:
  - [x] The `OnError` callback needs to include the xDS node ID in the returned error.
  - [x] The error is also stored in the cache to return to future watchers. This should include the xDS node ID as well.  
- [x] Reporting connectivity failures from xDS client to watchers. xDS node id should be part of these errors.
- [x] CDS lb policy needs to include the xDS node id in errors
  - [x] Requested resource is not found
  - [x] Sends a resolver error to child policy or puts the channel in TF
- [x] xDS resolver needs to include the xDS node id in the following cases
  - [x] Error reported to the channel (when requested resource is not found)
  - [x] Error reported from the config selector
- [x] xDS server needs to include the xDS node id when it puts the server in non-serving mode
- [x] All resource watchers' `OnResourceDoesNotExist` callback should include the xDS node id if they are throwing a log. Also, these logs should not be `Info` protected with a verbosity check, but should be `Warning` logs since resource not found is usually something that is worth sharing with the user by default. 

Solution patch:
diff --git a/internal/proto/grpc_lookup_v1/rls_config.pb.go b/internal/proto/grpc_lookup_v1/rls_config.pb.go
index 1eb0ace1..89fdb7fc 100644
--- a/internal/proto/grpc_lookup_v1/rls_config.pb.go
+++ b/internal/proto/grpc_lookup_v1/rls_config.pb.go
@@ -362,13 +362,15 @@ type RouteLookupConfig struct {
 	LookupServiceTimeout *durationpb.Duration `protobuf:"bytes,4,opt,name=lookup_service_timeout,json=lookupServiceTimeout,proto3" json:"lookup_service_timeout,omitempty"`
 	// How long are responses valid for (like HTTP Cache-Control).
 	// If omitted or zero, the longest valid cache time is used.
-	// This value is clamped to 5 minutes to avoid unflushable bad responses.
+	// This value is clamped to 5 minutes to avoid unflushable bad responses,
+	// unless stale_age is specified.
 	MaxAge *durationpb.Duration `protobuf:"bytes,5,opt,name=max_age,json=maxAge,proto3" json:"max_age,omitempty"`
 	// After a response has been in the client cache for this amount of time
 	// and is re-requested, start an asynchronous RPC to re-validate it.
 	// This value should be less than max_age by at least the length of a
 	// typical RTT to the Route Lookup Service to fully mask the RTT latency.
 	// If omitted, keys are only re-requested after they have expired.
+	// This value is clamped to 5 minutes.
 	StaleAge *durationpb.Duration `protobuf:"bytes,6,opt,name=stale_age,json=staleAge,proto3" json:"stale_age,omitempty"`
 	// Rough indicator of amount of memory to use for the client cache.  Some of
 	// the data structure overhead is not accounted for, so actual memory consumed
diff --git a/internal/stubserver/stubserver.go b/internal/stubserver/stubserver.go
index f290d97e..9974d214 100644
--- a/internal/stubserver/stubserver.go
+++ b/internal/stubserver/stubserver.go
@@ -276,7 +276,7 @@ func (ss *StubServer) Stop() {
 }
 
 func parseCfg(r *manual.Resolver, s string) *serviceconfig.ParseResult {
-	g := r.CC.ParseServiceConfig(s)
+	g := r.CC().ParseServiceConfig(s)
 	if g.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", s, g.Err))
 	}
diff --git a/interop/alts/client/client.go b/interop/alts/client/client.go
index d4628028..52f465ae 100644
--- a/interop/alts/client/client.go
+++ b/interop/alts/client/client.go
@@ -47,7 +47,6 @@ func main() {
 		opts.HandshakerServiceAddress = *hsAddr
 	}
 	altsTC := alts.NewClientCreds(opts)
-	// Block until the server is ready.
 	conn, err := grpc.NewClient(*serverAddr, grpc.WithTransportCredentials(altsTC))
 	if err != nil {
 		logger.Fatalf("grpc.NewClient(%q) = %v", *serverAddr, err)
@@ -58,7 +57,8 @@ func main() {
 	// Call the EmptyCall API.
 	ctx := context.Background()
 	request := &testpb.Empty{}
-	if _, err := grpcClient.EmptyCall(ctx, request); err != nil {
+	// Block until the server is ready.
+	if _, err := grpcClient.EmptyCall(ctx, request, grpc.WaitForReady(true)); err != nil {
 		logger.Fatalf("grpc Client: EmptyCall(_, %v) failed: %v", request, err)
 	}
 	logger.Info("grpc Client: empty call succeeded")
diff --git a/resolver/manual/manual.go b/resolver/manual/manual.go
index 8400b5f2..82fcc2e3 100644
--- a/resolver/manual/manual.go
+++ b/resolver/manual/manual.go
@@ -62,7 +62,7 @@ type Resolver struct {
 	// Fields actually belong to the resolver.
 	// Guards access to below fields.
 	mu sync.Mutex
-	CC resolver.ClientConn
+	cc resolver.ClientConn
 	// Storing the most recent state update makes this resolver resilient to
 	// restarts, which is possible with channel idleness.
 	lastSeenState *resolver.State
@@ -78,12 +78,12 @@ func (r *Resolver) InitialState(s resolver.State) {
 func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	// Call BuildCallback after locking to avoid a race when UpdateState
-	// or ReportError is called before Build returns.
+	// Call BuildCallback after locking to avoid a race when UpdateState or CC
+	// is called before Build returns.
 	r.BuildCallback(target, cc, opts)
-	r.CC = cc
+	r.cc = cc
 	if r.lastSeenState != nil {
-		err := r.CC.UpdateState(*r.lastSeenState)
+		err := r.cc.UpdateState(*r.lastSeenState)
 		go r.UpdateStateCallback(err)
 	}
 	return r, nil
@@ -104,25 +104,27 @@ func (r *Resolver) Close() {
 	r.CloseCallback()
 }
 
-// UpdateState calls CC.UpdateState.
+// UpdateState calls UpdateState(s) on the channel.  If the resolver has not
+// been Built before, this instead sets the initial state of the resolver, like
+// InitialState.
 func (r *Resolver) UpdateState(s resolver.State) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	var err error
-	if r.CC == nil {
-		panic("cannot update state as channel has not exited IDLE state")
-	}
-	err = r.CC.UpdateState(s)
 	r.lastSeenState = &s
+	if r.cc == nil {
+		return
+	}
+	err := r.cc.UpdateState(s)
 	r.UpdateStateCallback(err)
 }
 
-// ReportError calls CC.ReportError.
-func (r *Resolver) ReportError(err error) {
+// CC returns r's ClientConn when r was last Built.  Panics if the resolver has
+// not been Built before.
+func (r *Resolver) CC() resolver.ClientConn {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	if r.CC == nil {
-		panic("cannot report error as channel has not exited IDLE state")
+	if r.cc == nil {
+		panic("Manual resolver instance has not yet been built.")
 	}
-	r.CC.ReportError(err)
+	return r.cc
 }
diff --git a/resolver_wrapper.go b/resolver_wrapper.go
index 945e24ff..80e16a32 100644
--- a/resolver_wrapper.go
+++ b/resolver_wrapper.go
@@ -134,12 +134,7 @@ func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
 		return nil
 	}
 	if s.Endpoints == nil {
-		s.Endpoints = make([]resolver.Endpoint, 0, len(s.Addresses))
-		for _, a := range s.Addresses {
-			ep := resolver.Endpoint{Addresses: []resolver.Address{a}, Attributes: a.BalancerAttributes}
-			ep.Addresses[0].BalancerAttributes = nil
-			s.Endpoints = append(s.Endpoints, ep)
-		}
+		s.Endpoints = addressesToEndpoints(s.Addresses)
 	}
 	ccr.addChannelzTraceEvent(s)
 	ccr.curState = s
@@ -172,7 +167,11 @@ func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
 		ccr.cc.mu.Unlock()
 		return
 	}
-	s := resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig}
+	s := resolver.State{
+		Addresses:     addrs,
+		ServiceConfig: ccr.curState.ServiceConfig,
+		Endpoints:     addressesToEndpoints(addrs),
+	}
 	ccr.addChannelzTraceEvent(s)
 	ccr.curState = s
 	ccr.mu.Unlock()
@@ -210,3 +209,13 @@ func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
 	}
 	channelz.Infof(logger, ccr.cc.channelz, "Resolver state updated: %s (%v)", pretty.ToJSON(s), strings.Join(updates, "; "))
 }
+
+func addressesToEndpoints(addrs []resolver.Address) []resolver.Endpoint {
+	endpoints := make([]resolver.Endpoint, 0, len(addrs))
+	for _, a := range addrs {
+		ep := resolver.Endpoint{Addresses: []resolver.Address{a}, Attributes: a.BalancerAttributes}
+		ep.Addresses[0].BalancerAttributes = nil
+		endpoints = append(endpoints, ep)
+	}
+	return endpoints
+}
diff --git a/scripts/vet.sh b/scripts/vet.sh
index d0f533c2..e3b77dd0 100755
--- a/scripts/vet.sh
+++ b/scripts/vet.sh
@@ -174,6 +174,7 @@ CredsBundle is deprecated:
 GetMetadata is deprecated:
 internal.Logger is deprecated:
 Metadata is deprecated: use Attributes instead.
+NewAddress is deprecated:
 NewSubConn is deprecated:
 OverrideServerName is deprecated:
 RemoveSubConn is deprecated:
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer.go b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
index 3f055a29..f282b43c 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
@@ -430,6 +430,21 @@ func (b *cdsBalancer) ExitIdle() {
 	})
 }
 
+// Node ID needs to be manually added to errors generated in the following
+// scenarios:
+//   - resource-does-not-exist: since the xDS watch API uses a separate callback
+//     instead of returning an error value. TODO(gRFC A88): Once A88 is
+//     implemented, the xDS client will be able to add the node ID to
+//     resource-does-not-exist errors as well, and we can get rid of this
+//     special handling.
+//   - received a good update from the xDS client, but the update either contains
+//     an invalid security configuration or contains invalid aggragate cluster
+//     config.
+func (b *cdsBalancer) annotateErrorWithNodeID(err error) error {
+	nodeID := b.xdsClient.BootstrapConfig().Node().GetId()
+	return fmt.Errorf("[xDS node id: %v]: %w", nodeID, err)
+}
+
 // Handles a good Cluster update from the xDS client. Kicks off the discovery
 // mechanism generation process from the top-level cluster and if the cluster
 // graph is resolved, generates child policy config and pushes it down.
@@ -459,7 +474,7 @@ func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpd
 			// If the security config is invalid, for example, if the provider
 			// instance is not found in the bootstrap config, we need to put the
 			// channel in transient failure.
-			b.onClusterError(name, fmt.Errorf("received Cluster resource contains invalid security config: %v", err))
+			b.onClusterError(name, b.annotateErrorWithNodeID(fmt.Errorf("received Cluster resource contains invalid security config: %v", err)))
 			return
 		}
 	}
@@ -467,12 +482,12 @@ func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpd
 	clustersSeen := make(map[string]bool)
 	dms, ok, err := b.generateDMsForCluster(b.lbCfg.ClusterName, 0, nil, clustersSeen)
 	if err != nil {
-		b.onClusterError(b.lbCfg.ClusterName, fmt.Errorf("failed to generate discovery mechanisms: %v", err))
+		b.onClusterError(b.lbCfg.ClusterName, b.annotateErrorWithNodeID(fmt.Errorf("failed to generate discovery mechanisms: %v", err)))
 		return
 	}
 	if ok {
 		if len(dms) == 0 {
-			b.onClusterError(b.lbCfg.ClusterName, fmt.Errorf("aggregate cluster graph has no leaf clusters"))
+			b.onClusterError(b.lbCfg.ClusterName, b.annotateErrorWithNodeID(fmt.Errorf("aggregate cluster graph has no leaf clusters")))
 			return
 		}
 		// Child policy is built the first time we resolve the cluster graph.
@@ -557,7 +572,7 @@ func (b *cdsBalancer) onClusterError(name string, err error) {
 //
 // Only executed in the context of a serializer callback.
 func (b *cdsBalancer) onClusterResourceNotFound(name string) {
-	err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "cluster %q not found", name)
+	err := b.annotateErrorWithNodeID(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "cluster %q not found", name))
 	b.closeChildPolicyAndReportTF(err)
 }
 
diff --git a/xds/internal/balancer/priority/balancer.go b/xds/internal/balancer/priority/balancer.go
index 761f6eea..194e0319 100644
--- a/xds/internal/balancer/priority/balancer.go
+++ b/xds/internal/balancer/priority/balancer.go
@@ -210,6 +210,9 @@ func (b *priorityBalancer) UpdateClientConnState(s balancer.ClientConnState) err
 }
 
 func (b *priorityBalancer) ResolverError(err error) {
+	if b.logger.V(2) {
+		b.logger.Infof("Received error from the resolver: %v", err)
+	}
 	b.bg.ResolverError(err)
 }
 
diff --git a/xds/internal/clients/config.go b/xds/internal/clients/config.go
index 5640a1e0..a2220f99 100644
--- a/xds/internal/clients/config.go
+++ b/xds/internal/clients/config.go
@@ -16,11 +16,10 @@
  *
  */
 
-// Package clients provides implementations of the xDS and LRS clients,
-// enabling applications to communicate with xDS management servers and report
-// load.
+// Package clients provides implementations of the clients to interact with
+// xDS and LRS servers.
 //
-// xDS Client
+// # xDS Client
 //
 // The xDS client allows applications to:
 //   - Create client instances with in-memory configurations.
@@ -41,100 +40,30 @@
 //
 // NOTICE: This package is EXPERIMENTAL and may be changed or removed
 // in a later release.
-//
-// See [README](https://github.com/grpc/grpc-go/tree/master/xds/clients/README.md).
 package clients
 
-import (
-	"fmt"
-	"slices"
-	"strings"
-
-	"google.golang.org/protobuf/proto"
-	"google.golang.org/protobuf/types/known/structpb"
-
-	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-)
-
-// ServerConfig holds settings for connecting to an xDS management server.
-type ServerConfig struct {
-	// ServerURI is the target URI of the xDS management server.
+// ServerIdentifier holds identifying information for connecting to an xDS
+// management or LRS server.
+type ServerIdentifier struct {
+	// ServerURI is the target URI of the server.
 	ServerURI string
 
-	// IgnoreResourceDeletion is a server feature which if set to true,
-	// indicates that resource deletion errors can be ignored and cached
-	// resource data can be used.
-	//
-	// This will be removed in the future once we implement gRFC A88
-	// and two new fields FailOnDataErrors and
-	// ResourceTimerIsTransientError will be introduced.
-	IgnoreResourceDeletion bool
-
 	// Extensions can be populated with arbitrary data to be passed to the
-	// [TransportBuilder] and/or xDS Client's ResourceType implementations.
+	// TransportBuilder and/or xDS Client's ResourceType implementations.
 	// This field can be used to provide additional configuration or context
 	// specific to the user's needs.
 	//
 	// The xDS and LRS clients do not interpret the contents of this field.
-	// It is the responsibility of the user's custom [TransportBuilder] and/or
+	// It is the responsibility of the user's custom TransportBuilder and/or
 	// ResourceType implementations to handle and interpret these extensions.
 	//
-	// For example, a custom [TransportBuilder] might use this field to
+	// For example, a custom TransportBuilder might use this field to
 	// configure a specific security credentials.
-	//
-	// Note: For custom types used in Extensions, ensure an Equal(any) bool
-	// method is implemented for equality checks on ServerConfig.
 	Extensions any
 }
 
-// equal returns true if sc and other are considered equal.
-func (sc *ServerConfig) equal(other *ServerConfig) bool {
-	switch {
-	case sc == nil && other == nil:
-		return true
-	case (sc != nil) != (other != nil):
-		return false
-	case sc.ServerURI != other.ServerURI:
-		return false
-	case sc.IgnoreResourceDeletion != other.IgnoreResourceDeletion:
-		return false
-	}
-	if sc.Extensions == nil && other.Extensions == nil {
-		return true
-	}
-	if ex, ok := sc.Extensions.(interface{ Equal(any) bool }); ok && ex.Equal(other.Extensions) {
-		return true
-	}
-	return false
-}
-
-// String returns a string representation of the [ServerConfig].
-//
-// WARNING: This method is primarily intended for logging and testing
-// purposes. The output returned by this method is not guaranteed to be stable
-// and may change at any time. Do not rely on it for production use.
-func (sc *ServerConfig) String() string {
-	return strings.Join([]string{sc.ServerURI, fmt.Sprintf("%v", sc.IgnoreResourceDeletion)}, "-")
-}
-
-// Authority contains configuration for an xDS control plane authority.
-type Authority struct {
-	// XDSServers contains the list of server configurations for this authority.
-	XDSServers []ServerConfig
-
-	// Extensions can be populated with arbitrary data to be passed to the xDS
-	// Client's user specific implementations. This field can be used to
-	// provide additional configuration or context specific to the user's
-	// needs.
-	//
-	// The xDS and LRS clients do not interpret the contents of this field. It
-	// is the responsibility of the user's implementations to handle and
-	// interpret these extensions.
-	Extensions any
-}
-
-// Node represents the identity of the xDS client, allowing
-// management servers to identify the source of xDS requests.
+// Node represents the identity of the xDS client, allowing xDS and LRS servers
+// to identify the source of xDS requests.
 type Node struct {
 	// ID is a string identifier of the application.
 	ID string
@@ -150,40 +79,6 @@ type Node struct {
 	UserAgentName string
 	// UserAgentVersion is the user agent version of application.
 	UserAgentVersion string
-	// ClientFeatures is a list of xDS features supported by this client.
-	// These features are set within the xDS client, but may be overridden only
-	// for testing purposes.
-	clientFeatures []string
-}
-
-// toProto converts an instance of [Node] to its protobuf representation.
-func (n Node) toProto() *v3corepb.Node {
-	return &v3corepb.Node{
-		Id:      n.ID,
-		Cluster: n.Cluster,
-		Locality: func() *v3corepb.Locality {
-			if n.Locality.isEmpty() {
-				return nil
-			}
-			return &v3corepb.Locality{
-				Region:  n.Locality.Region,
-				Zone:    n.Locality.Zone,
-				SubZone: n.Locality.SubZone,
-			}
-		}(),
-		Metadata: func() *structpb.Struct {
-			if n.Metadata == nil {
-				return nil
-			}
-			if md, ok := n.Metadata.(*structpb.Struct); ok {
-				return proto.Clone(md).(*structpb.Struct)
-			}
-			return nil
-		}(),
-		UserAgentName:        n.UserAgentName,
-		UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: n.UserAgentVersion},
-		ClientFeatures:       slices.Clone(n.clientFeatures),
-	}
 }
 
 // Locality represents the location of the xDS client application.
@@ -195,13 +90,3 @@ type Locality struct {
 	// SubZone is the further subdivision within a zone.
 	SubZone string
 }
-
-// isEmpty reports whether l is considered empty.
-func (l Locality) isEmpty() bool {
-	return l.equal(Locality{})
-}
-
-// equal returns true if l and other are considered equal.
-func (l Locality) equal(other Locality) bool {
-	return l.Region == other.Region && l.Zone == other.Zone && l.SubZone == other.SubZone
-}
diff --git a/xds/internal/clients/grpctransport/grpc_transport.go b/xds/internal/clients/grpctransport/grpc_transport.go
index c5c1f996..9040be3c 100644
--- a/xds/internal/clients/grpctransport/grpc_transport.go
+++ b/xds/internal/clients/grpctransport/grpc_transport.go
@@ -31,38 +31,38 @@ import (
 	"google.golang.org/grpc/xds/internal/clients"
 )
 
-// ServerConfigExtension holds settings for connecting to a gRPC server,
+// ServerIdentifierExtension holds settings for connecting to a gRPC server,
 // such as an xDS management or an LRS server.
-type ServerConfigExtension struct {
+type ServerIdentifierExtension struct {
 	// Credentials will be used for all gRPC transports. If it is unset,
 	// transport creation will fail.
 	Credentials credentials.Bundle
 }
 
-// Builder creates gRPC-based Transports. It must be paired with ServerConfigs
-// that contain an Extension field of type ServerConfigExtension.
+// Builder creates gRPC-based Transports. It must be paired with ServerIdentifiers
+// that contain an Extension field of type ServerIdentifierExtension.
 type Builder struct{}
 
 // Build returns a gRPC-based clients.Transport.
 //
-// The Extension field of the ServerConfig must be a ServerConfigExtension.
-func (b *Builder) Build(sc clients.ServerConfig) (clients.Transport, error) {
-	if sc.ServerURI == "" {
-		return nil, fmt.Errorf("grpctransport: ServerURI is not set in ServerConfig")
+// The Extension field of the ServerIdentifier must be a ServerIdentifierExtension.
+func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error) {
+	if si.ServerURI == "" {
+		return nil, fmt.Errorf("grpctransport: ServerURI is not set in ServerIdentifier")
 	}
-	if sc.Extensions == nil {
-		return nil, fmt.Errorf("grpctransport: Extensions is not set in ServerConfig")
+	if si.Extensions == nil {
+		return nil, fmt.Errorf("grpctransport: Extensions is not set in ServerIdentifier")
 	}
-	sce, ok := sc.Extensions.(ServerConfigExtension)
+	sce, ok := si.Extensions.(ServerIdentifierExtension)
 	if !ok {
-		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerConfig", sc.Extensions, ServerConfigExtension{})
+		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerIdentifier", si.Extensions, ServerIdentifierExtension{})
 	}
 	if sce.Credentials == nil {
-		return nil, fmt.Errorf("grptransport: Credentials field is not set in ServerConfigExtension")
+		return nil, fmt.Errorf("grptransport: Credentials field is not set in ServerIdentifierExtension")
 	}
 
 	// TODO: Incorporate reference count map for existing transports and
-	// deduplicate transports based on the provided ServerConfig so that
+	// deduplicate transports based on the provided ServerIdentifier so that
 	// transport channel to same server can be shared between xDS and LRS
 	// client.
 
@@ -74,9 +74,9 @@ func (b *Builder) Build(sc clients.ServerConfig) (clients.Transport, error) {
 		Time:    5 * time.Minute,
 		Timeout: 20 * time.Second,
 	})
-	cc, err := grpc.NewClient(sc.ServerURI, kpCfg, grpc.WithCredentialsBundle(sce.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
+	cc, err := grpc.NewClient(si.ServerURI, kpCfg, grpc.WithCredentialsBundle(sce.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
 	if err != nil {
-		return nil, fmt.Errorf("grpctransport: failed to create transport to server %q: %v", sc.ServerURI, err)
+		return nil, fmt.Errorf("grpctransport: failed to create transport to server %q: %v", si.ServerURI, err)
 	}
 
 	return &grpcTransport{cc: cc}, nil
diff --git a/xds/internal/clients/internal/internal.go b/xds/internal/clients/internal/internal.go
new file mode 100644
index 00000000..d7121298
--- /dev/null
+++ b/xds/internal/clients/internal/internal.go
@@ -0,0 +1,84 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// Package internal contains helpers for xDS and LRS clients.
+package internal
+
+import (
+	"fmt"
+	"strings"
+
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+)
+
+// ServerIdentifierString returns a string representation of the
+// clients.ServerIdentifier si.
+//
+// WARNING: This method is primarily intended for logging and testing
+// purposes. The output returned by this method is not guaranteed to be stable
+// and may change at any time. Do not rely on it for production use.
+func ServerIdentifierString(si clients.ServerIdentifier) string {
+	return strings.Join([]string{si.ServerURI, fmt.Sprintf("%v", si.Extensions)}, "-")
+}
+
+// NodeProto returns a protobuf representation of clients.Node n.
+//
+// This function is intended to be used by the client implementation to convert
+// the user-provided Node configuration to its protobuf representation.
+func NodeProto(n clients.Node) *v3corepb.Node {
+	return &v3corepb.Node{
+		Id:      n.ID,
+		Cluster: n.Cluster,
+		Locality: func() *v3corepb.Locality {
+			if isLocalityEmpty(n.Locality) {
+				return nil
+			}
+			return &v3corepb.Locality{
+				Region:  n.Locality.Region,
+				Zone:    n.Locality.Zone,
+				SubZone: n.Locality.SubZone,
+			}
+		}(),
+		Metadata: func() *structpb.Struct {
+			if n.Metadata == nil {
+				return nil
+			}
+			if md, ok := n.Metadata.(*structpb.Struct); ok {
+				return proto.Clone(md).(*structpb.Struct)
+			}
+			return nil
+		}(),
+		UserAgentName:        n.UserAgentName,
+		UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: n.UserAgentVersion},
+	}
+}
+
+// isLocalityEqual reports whether clients.Locality l is considered empty.
+func isLocalityEmpty(l clients.Locality) bool {
+	return isLocalityEqual(l, clients.Locality{})
+}
+
+// isLocalityEqual returns true if clients.Locality l1 and l2 are considered
+// equal.
+func isLocalityEqual(l1, l2 clients.Locality) bool {
+	return l1.Region == l2.Region && l1.Zone == l2.Zone && l1.SubZone == l2.SubZone
+}
diff --git a/xds/internal/clients/lrsclient/load_store.go b/xds/internal/clients/lrsclient/load_store.go
new file mode 100644
index 00000000..d52db0c7
--- /dev/null
+++ b/xds/internal/clients/lrsclient/load_store.go
@@ -0,0 +1,80 @@
+//revive:disable:unused-parameter
+
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package lrsclient
+
+import "context"
+
+// A LoadStore aggregates loads for multiple clusters and services that are
+// intended to be reported via LRS.
+//
+// LoadStore stores loads reported to a single LRS server. Use multiple stores
+// for multiple servers.
+//
+// It is safe for concurrent use.
+type LoadStore struct {
+}
+
+// Stop stops the LRS stream associated with this LoadStore.
+//
+// If this LoadStore is the only one using the underlying LRS stream, the
+// stream will be closed. If other LoadStores are also using the same stream,
+// the reference count to the stream is decremented, and the stream remains
+// open until all LoadStores have called Stop().
+//
+// If this is the last LoadStore for the stream, this method makes a last
+// attempt to flush any unreported load data to the LRS server. It will either
+// wait for this attempt to complete, or for the provided context to be done
+// before canceling the LRS stream.
+func (ls *LoadStore) Stop(ctx context.Context) error {
+	panic("unimplemented")
+}
+
+// ReporterForCluster returns the PerClusterReporter for the given cluster and
+// service.
+func (ls *LoadStore) ReporterForCluster(clusterName, serviceName string) PerClusterReporter {
+	panic("unimplemented")
+}
+
+// PerClusterReporter records load data pertaining to a single cluster. It
+// provides methods to record call starts, finishes, server-reported loads,
+// and dropped calls.
+type PerClusterReporter struct {
+}
+
+// CallStarted records a call started in the LoadStore.
+func (p *PerClusterReporter) CallStarted(locality string) {
+	panic("unimplemented")
+}
+
+// CallFinished records a call finished in the LoadStore.
+func (p *PerClusterReporter) CallFinished(locality string, err error) {
+	panic("unimplemented")
+}
+
+// CallServerLoad records the server load in the LoadStore.
+func (p *PerClusterReporter) CallServerLoad(locality, name string, val float64) {
+	panic("unimplemented")
+}
+
+// CallDropped records a call dropped in the LoadStore.
+func (p *PerClusterReporter) CallDropped(category string) {
+	panic("unimplemented")
+}
diff --git a/xds/internal/clients/lrsclient/lrsclient.go b/xds/internal/clients/lrsclient/lrsclient.go
new file mode 100644
index 00000000..5bd8aa60
--- /dev/null
+++ b/xds/internal/clients/lrsclient/lrsclient.go
@@ -0,0 +1,39 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// Package lrsclient provides an LRS (Load Reporting Service) client.
+//
+// See: https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/load_stats/v3/lrs.proto
+package lrsclient
+
+import "google.golang.org/grpc/xds/internal/clients"
+
+// LRSClient is an LRS (Load Reporting Service) client.
+type LRSClient struct {
+}
+
+// New returns a new LRS Client configured with the provided config.
+func New(_ Config) (*LRSClient, error) {
+	panic("unimplemented")
+}
+
+// ReportLoad creates a new load reporting stream for the provided server. It
+// creates and returns a LoadStore for the caller to report loads.
+func (*LRSClient) ReportLoad(_ clients.ServerIdentifier) *LoadStore {
+	panic("unimplemented")
+}
diff --git a/xds/internal/clients/lrsclient/lrsconfig.go b/xds/internal/clients/lrsclient/lrsconfig.go
new file mode 100644
index 00000000..c4862ff7
--- /dev/null
+++ b/xds/internal/clients/lrsclient/lrsconfig.go
@@ -0,0 +1,35 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package lrsclient
+
+import (
+	"google.golang.org/grpc/xds/internal/clients"
+)
+
+// Config is used to configure an LRS client. After one has been passed to the
+// LRS client's New function, no part of it may modified. A Config may be
+// reused; the lrsclient package will also not modify it.
+type Config struct {
+	// Node is the identity of the client application reporting load to the
+	// LRS server.
+	Node clients.Node
+
+	// TransportBuilder is used to connect to the LRS server.
+	TransportBuilder clients.TransportBuilder
+}
diff --git a/xds/internal/clients/transport_builder.go b/xds/internal/clients/transport_builder.go
index 5b0eb10c..10a25fca 100644
--- a/xds/internal/clients/transport_builder.go
+++ b/xds/internal/clients/transport_builder.go
@@ -23,19 +23,19 @@ import (
 )
 
 // TransportBuilder provides the functionality to create a communication
-// channel to an xDS management server.
+// channel to an xDS or LRS server.
 type TransportBuilder interface {
-	// Build creates a new [Transport] instance to the xDS server based on the
-	// provided ServerConfig.
-	Build(ServerConfig ServerConfig) (Transport, error)
+	// Build creates a new Transport instance to the server based on the
+	// provided ServerIdentifier.
+	Build(serverIdentifier ServerIdentifier) (Transport, error)
 }
 
-// Transport provides the functionality to communicate with an xDS server using
-// streaming calls.
+// Transport provides the functionality to communicate with an xDS or LRS
+// server using streaming calls.
 type Transport interface {
-	// NewStream creates a new streaming call to the xDS server for the
-	// specified RPC method name. The returned Stream interface can be used
-	// to send and receive messages on the stream.
+	// NewStream creates a new streaming call to the server for the specific
+	// RPC method name. The returned Stream interface can be used to send and
+	// receive messages on the stream.
 	NewStream(context.Context, string) (Stream, error)
 
 	// Close closes the Transport.
@@ -43,7 +43,7 @@ type Transport interface {
 }
 
 // Stream provides methods to send and receive messages on a stream. Messages
-// are represented as a byte slice ([]byte).
+// are represented as a byte slice.
 type Stream interface {
 	// Send sends the provided message on the stream.
 	Send([]byte) error
diff --git a/xds/internal/clients/xdsclient/resource_type.go b/xds/internal/clients/xdsclient/resource_type.go
new file mode 100644
index 00000000..8ca466ed
--- /dev/null
+++ b/xds/internal/clients/xdsclient/resource_type.go
@@ -0,0 +1,92 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xdsclient
+
+// ResourceType wraps all resource-type specific functionality. Each supported
+// resource type needs to provide an implementation of the Decoder.
+type ResourceType struct {
+	// TypeURL is the xDS type URL of this resource type for the v3 xDS
+	// protocol. This URL is used as the key to look up the corresponding
+	// ResourceType implementation in the ResourceTypes map provided in the
+	// Config.
+	TypeURL string
+
+	// TODO: Revisit if we need TypeURL to be part of the struct because it is
+	// already a key in config's ResouceTypes map.
+
+	// TypeName is a shorter representation of the TypeURL to identify the
+	// resource type. It is used for logging/debugging purposes.
+	TypeName string
+
+	// AllResourcesRequiredInSotW indicates whether this resource type requires
+	// that all resources be present in every SotW response from the server. If
+	// true, a response that does not include a previously seen resource will
+	// be interpreted as a deletion of that resource.
+	AllResourcesRequiredInSotW bool
+
+	// Decoder is used to deserialize and validate an xDS resource received
+	// from the xDS management server.
+	Decoder Decoder
+}
+
+// Decoder wraps the resource-type specific functionality for validation
+// and deserialization.
+type Decoder interface {
+	// Decode deserializes and validates an xDS resource as received from the
+	// xDS management server.
+	//
+	// If deserialization fails or resource validation fails, it returns a
+	// non-nil error. Otherwise, returns a fully populated DecodeResult.
+	Decode(resource []byte, options DecodeOptions) (*DecodeResult, error)
+}
+
+// DecodeOptions wraps the options required by ResourceType implementations for
+// decoding configuration received from the xDS management server.
+type DecodeOptions struct {
+	// Config contains the complete configuration passed to the xDS client.
+	// This contains useful data for resource validation.
+	Config *Config
+
+	// ServerConfig contains the configuration of the xDS server that provided
+	// the current resource being decoded.
+	ServerConfig *ServerConfig
+}
+
+// DecodeResult is the result of a decode operation.
+type DecodeResult struct {
+	// Name is the name of the decoded resource.
+	Name string
+
+	// Resource contains the configuration associated with the decoded
+	// resource.
+	Resource ResourceData
+}
+
+// ResourceData contains the configuration data sent by the xDS management
+// server, associated with the resource being watched. Every resource type must
+// provide an implementation of this interface to represent the configuration
+// received from the xDS management server.
+type ResourceData interface {
+	// Equal returns true if the passed in resource data is equal to that of
+	// the receiver.
+	Equal(other ResourceData) bool
+
+	// Bytes returns the underlying raw bytes of the resource proto.
+	Bytes() []byte
+}
diff --git a/xds/internal/clients/xdsclient/resource_watcher.go b/xds/internal/clients/xdsclient/resource_watcher.go
new file mode 100644
index 00000000..37d01bc7
--- /dev/null
+++ b/xds/internal/clients/xdsclient/resource_watcher.go
@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xdsclient
+
+// ResourceWatcher is notified of the resource updates and errors that are
+// received by the xDS client from the management server.
+//
+// All methods contain a done parameter which should be called when processing
+// of the update has completed.  For example, if processing a resource requires
+// watching new resources, those watches should be completed before done is
+// called, which can happen after the ResourceWatcher method has returned.
+// Failure to call done will prevent the xDS client from providing future
+// ResourceWatcher notifications.
+type ResourceWatcher interface {
+	// ResourceChanged indicates a new version of the resource is available.
+	ResourceChanged(resourceData ResourceData, done func())
+
+	// ResourceError indicates an error occurred while trying to fetch or
+	// decode the associated resource. The previous version of the resource
+	// should be considered invalid.
+	ResourceError(err error, done func())
+
+	// AmbientError indicates an error occurred after a resource has been
+	// received that should not modify the use of that resource but may provide
+	// useful information about the state of the XDSClient for debugging
+	// purposes. The previous version of the resource should still be
+	// considered valid.
+	AmbientError(err error, done func())
+}
diff --git a/xds/internal/clients/xdsclient/xdsclient.go b/xds/internal/clients/xdsclient/xdsclient.go
new file mode 100644
index 00000000..f893e8c9
--- /dev/null
+++ b/xds/internal/clients/xdsclient/xdsclient.go
@@ -0,0 +1,67 @@
+//revive:disable:unused-parameter
+
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// Package xdsclient provides an xDS (* Discovery Service) client.
+//
+// It allows applications to:
+//   - Create xDS client instances with in-memory configurations.
+//   - Register watches for named resources.
+//   - Receive resources via an ADS (Aggregated Discovery Service) stream.
+//   - Register watches for named resources (e.g. listeners, routes, or
+//     clusters).
+//
+// This enables applications to dynamically discover and configure resources
+// such as listeners, routes, clusters, and endpoints from an xDS management
+// server.
+package xdsclient
+
+// XDSClient is a client which queries a set of discovery APIs (collectively
+// termed as xDS) on a remote management server, to discover
+// various dynamic resources.
+type XDSClient struct {
+}
+
+// New returns a new xDS Client configured with the provided config.
+func New(config Config) (*XDSClient, error) {
+	panic("unimplemented")
+}
+
+// WatchResource starts watching the specified resource.
+//
+// typeURL specifies the resource type implementation to use. The watch fails
+// if there is no resource type implementation for the given typeURL. See the
+// ResourceTypes field in the Config struct used to create the XDSClient.
+//
+// The returned function cancels the watch and prevents future calls to the
+// watcher.
+func (c *XDSClient) WatchResource(typeURL, name string, watcher ResourceWatcher) (cancel func()) {
+	panic("unimplemented")
+}
+
+// Close closes the xDS client.
+func (c *XDSClient) Close() error {
+	panic("unimplemented")
+}
+
+// DumpResources returns the status and contents of all xDS resources being
+// watched by the xDS client.
+func (c *XDSClient) DumpResources() []byte {
+	panic("unimplemented")
+}
diff --git a/xds/internal/clients/xdsclient/xdsconfig.go b/xds/internal/clients/xdsclient/xdsconfig.go
new file mode 100644
index 00000000..bfbe4167
--- /dev/null
+++ b/xds/internal/clients/xdsclient/xdsconfig.go
@@ -0,0 +1,85 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xdsclient
+
+import (
+	"google.golang.org/grpc/xds/internal/clients"
+)
+
+// Config is used to configure an xDS client. After one has been passed to the
+// xDS client's New function, no part of it may be modified. A Config may be
+// reused; the xdsclient package will also not modify it.
+type Config struct {
+	// Servers specifies a list of xDS management servers to connect to. The
+	// order of the servers in this list reflects the order of preference of
+	// the data returned by those servers. The xDS client uses the first
+	// available server from the list.
+	//
+	// See gRFC A71 for more details on fallback behavior when the primary
+	// xDS server is unavailable.
+	//
+	// gRFC A71: https://github.com/grpc/proposal/blob/master/A71-xds-fallback.md
+	Servers []ServerConfig
+
+	// Authorities defines the configuration for each xDS authority.  Federated resources
+	// will be fetched from the servers specified by the corresponding Authority.
+	Authorities map[string]Authority
+
+	// Node is the identity of the xDS client connecting to the xDS
+	// management server.
+	Node clients.Node
+
+	// TransportBuilder is used to create connections to xDS management servers.
+	TransportBuilder clients.TransportBuilder
+
+	// ResourceTypes is a map from resource type URLs to resource type
+	// implementations. Each resource type URL uniquely identifies a specific
+	// kind of xDS resource, and the corresponding resource type implementation
+	// provides logic for parsing, validating, and processing resources of that
+	// type.
+	//
+	// For example: "type.googleapis.com/envoy.config.listener.v3.Listener"
+	ResourceTypes map[string]ResourceType
+}
+
+// ServerConfig contains configuration for an xDS management server.
+type ServerConfig struct {
+	ServerIdentifier clients.ServerIdentifier
+
+	// IgnoreResourceDeletion is a server feature which if set to true,
+	// indicates that resource deletion errors from xDS management servers can
+	// be ignored and cached resource data can be used.
+	//
+	// This will be removed in the future once we implement gRFC A88
+	// and two new fields FailOnDataErrors and
+	// ResourceTimerIsTransientError will be introduced.
+	IgnoreResourceDeletion bool
+
+	// TODO: Link to gRFC A88
+}
+
+// Authority contains configuration for an xDS control plane authority.
+//
+// See: https://www.envoyproxy.io/docs/envoy/latest/xds/core/v3/resource_locator.proto#xds-core-v3-resourcelocator
+type Authority struct {
+	// XDSServers contains the list of server configurations for this authority.
+	//
+	// See Config.Servers for more details.
+	XDSServers []ServerConfig
+}
diff --git a/xds/internal/resolver/serviceconfig.go b/xds/internal/resolver/serviceconfig.go
index ca9f2c01..02e6a73e 100644
--- a/xds/internal/resolver/serviceconfig.go
+++ b/xds/internal/resolver/serviceconfig.go
@@ -125,8 +125,34 @@ func (r route) String() string {
 	return fmt.Sprintf("%s -> { clusters: %v, maxStreamDuration: %v }", r.m.String(), r.clusters, r.maxStreamDuration)
 }
 
+// stoppableConfigSelector extends the iresolver.ConfigSelector interface with a
+// stop() method. This makes it possible to swap the current config selector
+// with an erroring config selector when the LDS or RDS resource is not found on
+// the management server.
+type stoppableConfigSelector interface {
+	iresolver.ConfigSelector
+	stop()
+}
+
+// erroringConfigSelector always returns an error, with the xDS node ID included
+// in the error message. It is used to swap out the current config selector
+// when the LDS or RDS resource is not found on the management server.
+type erroringConfigSelector struct {
+	err error
+}
+
+func newErroringConfigSelector(xdsNodeID string) *erroringConfigSelector {
+	return &erroringConfigSelector{err: annotateErrorWithNodeID(status.Errorf(codes.Unavailable, "no valid clusters"), xdsNodeID)}
+}
+
+func (cs *erroringConfigSelector) SelectConfig(iresolver.RPCInfo) (*iresolver.RPCConfig, error) {
+	return nil, cs.err
+}
+func (cs *erroringConfigSelector) stop() {}
+
 type configSelector struct {
 	r                *xdsResolver
+	xdsNodeID        string
 	virtualHost      virtualHost
 	routes           []route
 	clusters         map[string]*clusterInfo
@@ -136,10 +162,14 @@ type configSelector struct {
 var errNoMatchedRouteFound = status.Errorf(codes.Unavailable, "no matched route was found")
 var errUnsupportedClientRouteAction = status.Errorf(codes.Unavailable, "matched route does not have a supported route action type")
 
+// annotateErrorWithNodeID annotates the given error with the provided xDS node
+// ID. This is used by the real config selector when it runs into errors, and
+// also by the erroring config selector.
+func annotateErrorWithNodeID(err error, nodeID string) error {
+	return fmt.Errorf("[xDS node id: %s]: %w", nodeID, err)
+}
+
 func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RPCConfig, error) {
-	if cs == nil {
-		return nil, status.Errorf(codes.Unavailable, "no valid clusters")
-	}
 	var rt *route
 	// Loop through routes in order and select first match.
 	for _, r := range cs.routes {
@@ -150,16 +180,16 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 	}
 
 	if rt == nil || rt.clusters == nil {
-		return nil, errNoMatchedRouteFound
+		return nil, annotateErrorWithNodeID(errNoMatchedRouteFound, cs.xdsNodeID)
 	}
 
 	if rt.actionType != xdsresource.RouteActionRoute {
-		return nil, errUnsupportedClientRouteAction
+		return nil, annotateErrorWithNodeID(errUnsupportedClientRouteAction, cs.xdsNodeID)
 	}
 
 	cluster, ok := rt.clusters.Next().(*routeCluster)
 	if !ok {
-		return nil, status.Errorf(codes.Internal, "error retrieving cluster for match: %v (%T)", cluster, cluster)
+		return nil, annotateErrorWithNodeID(status.Errorf(codes.Internal, "error retrieving cluster for match: %v (%T)", cluster, cluster), cs.xdsNodeID)
 	}
 
 	// Add a ref to the selected cluster, as this RPC needs this cluster until
@@ -169,7 +199,7 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 
 	interceptor, err := cs.newInterceptor(rt, cluster)
 	if err != nil {
-		return nil, err
+		return nil, annotateErrorWithNodeID(err, cs.xdsNodeID)
 	}
 
 	lbCtx := clustermanager.SetPickedCluster(rpcInfo.Context, cluster.name)
diff --git a/xds/internal/resolver/xds_resolver.go b/xds/internal/resolver/xds_resolver.go
index c333447e..5378371b 100644
--- a/xds/internal/resolver/xds_resolver.go
+++ b/xds/internal/resolver/xds_resolver.go
@@ -249,7 +249,7 @@ type xdsResolver struct {
 	// cluster that includes a ref count and load balancing configuration.
 	activeClusters map[string]*clusterInfo
 
-	curConfigSelector *configSelector
+	curConfigSelector stoppableConfigSelector
 }
 
 // ResolveNow is a no-op at this point.
@@ -284,22 +284,26 @@ func (r *xdsResolver) Close() {
 // false if an error occurs while sending an update to the channel.
 //
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
+func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) bool {
 	// Delete entries from r.activeClusters with zero references;
 	// otherwise serviceConfigJSON will generate a config including
 	// them.
 	r.pruneActiveClusters()
 
-	if cs == nil && len(r.activeClusters) == 0 {
+	errCS, ok := cs.(*erroringConfigSelector)
+	if ok && len(r.activeClusters) == 0 {
 		// There are no clusters and we are sending a failing configSelector.
 		// Send an empty config, which picks pick-first, with no address, and
 		// puts the ClientConn into transient failure.
-		if err := r.cc.UpdateState(resolver.State{ServiceConfig: r.cc.ParseServiceConfig("{}")}); err != nil {
-			if r.logger.V(2) {
-				r.logger.Infof("Channel rejected new state (with empty service config) with error: %v", err)
-			}
-			return false
-		}
+		//
+		// This call to UpdateState is expected to return ErrBadResolverState
+		// since pick_first doesn't like an update with no addresses.
+		r.cc.UpdateState(resolver.State{ServiceConfig: r.cc.ParseServiceConfig("{}")})
+
+		// Send a resolver error to pick_first so that RPCs will fail with a
+		// more meaningful error, as opposed to one that says that pick_first
+		// received no addresses.
+		r.cc.ReportError(errCS.err)
 		return true
 	}
 
@@ -326,7 +330,8 @@ func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) newConfigSelector() *configSelector {
 	cs := &configSelector{
-		r: r,
+		r:         r,
+		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
 		virtualHost: virtualHost{
 			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
 			retryConfig:              r.currentVirtualHost.RetryConfig,
@@ -438,14 +443,16 @@ func (r *xdsResolver) onResolutionComplete() {
 
 	cs := r.newConfigSelector()
 	if !r.sendNewServiceConfig(cs) {
-		// JSON error creating the service config (unexpected); erase
+		// Channel didn't like the update we provided (unexpected); erase
 		// this config selector and ignore this update, continuing with
 		// the previous config selector.
 		cs.stop()
 		return
 	}
 
-	r.curConfigSelector.stop()
+	if r.curConfigSelector != nil {
+		r.curConfigSelector.stop()
+	}
 	r.curConfigSelector = cs
 }
 
@@ -477,18 +484,21 @@ func (r *xdsResolver) onError(err error) {
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) onResourceNotFound() {
 	// We cannot remove clusters from the service config that have ongoing RPCs.
-	// Instead, what we can do is to send an erroring (nil) config selector
+	// Instead, what we can do is to send an erroring config selector
 	// along with normal service config. This will ensure that new RPCs will
 	// fail, and once the active RPCs complete, the reference counts on the
 	// clusters will come down to zero. At that point, we will send an empty
 	// service config with no addresses. This results in the pick-first
 	// LB policy being configured on the channel, and since there are no
 	// address, pick-first will put the channel in TRANSIENT_FAILURE.
-	r.sendNewServiceConfig(nil)
+	cs := newErroringConfigSelector(r.xdsClient.BootstrapConfig().Node().GetId())
+	r.sendNewServiceConfig(cs)
 
 	// Stop and dereference the active config selector, if one exists.
-	r.curConfigSelector.stop()
-	r.curConfigSelector = nil
+	if r.curConfigSelector != nil {
+		r.curConfigSelector.stop()
+	}
+	r.curConfigSelector = cs
 }
 
 // Only executed in the context of a serializer callback.
diff --git a/xds/internal/server/listener_wrapper.go b/xds/internal/server/listener_wrapper.go
index 09d32001..6fd1e766 100644
--- a/xds/internal/server/listener_wrapper.go
+++ b/xds/internal/server/listener_wrapper.go
@@ -84,6 +84,7 @@ func NewListenerWrapper(params ListenerWrapperParams) net.Listener {
 		Listener:          params.Listener,
 		name:              params.ListenerResourceName,
 		xdsC:              params.XDSClient,
+		xdsNodeID:         params.XDSClient.BootstrapConfig().Node().GetId(),
 		modeCallback:      params.ModeCallback,
 		isUnspecifiedAddr: params.Listener.Addr().(*net.TCPAddr).IP.IsUnspecified(),
 		conns:             make(map[*connWrapper]bool),
@@ -116,6 +117,7 @@ type listenerWrapper struct {
 
 	name         string
 	xdsC         XDSClient
+	xdsNodeID    string
 	cancelWatch  func()
 	modeCallback ServingModeCallback
 
@@ -173,7 +175,8 @@ func (l *listenerWrapper) handleLDSUpdate(update xdsresource.ListenerUpdate) {
 	// what we have decided to do.
 	if ilc.Address != l.addr || ilc.Port != l.port {
 		l.mu.Lock()
-		l.switchModeLocked(connectivity.ServingModeNotServing, fmt.Errorf("address (%s:%s) in Listener update does not match listening address: (%s:%s)", ilc.Address, ilc.Port, l.addr, l.port))
+		err := l.annotateErrorWithNodeID(fmt.Errorf("address (%s:%s) in Listener update does not match listening address: (%s:%s)", ilc.Address, ilc.Port, l.addr, l.port))
+		l.switchModeLocked(connectivity.ServingModeNotServing, err)
 		l.mu.Unlock()
 		return
 	}
@@ -228,12 +231,14 @@ func (l *listenerWrapper) handleRDSUpdate(routeName string, rcu rdsWatcherUpdate
 				continue
 			}
 			if rcu.err != nil && rcu.data == nil { // Either NACK before update, or resource not found triggers this conditional.
-				fc.UsableRouteConfiguration.Store(&xdsresource.UsableRouteConfiguration{
-					Err: rcu.err,
-				})
+				urc := &xdsresource.UsableRouteConfiguration{Err: rcu.err}
+				urc.NodeID = l.xdsNodeID
+				fc.UsableRouteConfiguration.Store(urc)
 				continue
 			}
-			fc.UsableRouteConfiguration.Store(fc.ConstructUsableRouteConfiguration(*rcu.data))
+			urc := fc.ConstructUsableRouteConfiguration(*rcu.data)
+			urc.NodeID = l.xdsNodeID
+			fc.UsableRouteConfiguration.Store(urc)
 		}
 	}
 	if l.rdsHandler.determineRouteConfigurationReady() {
@@ -248,15 +253,21 @@ func (l *listenerWrapper) handleRDSUpdate(routeName string, rcu rdsWatcherUpdate
 func (l *listenerWrapper) instantiateFilterChainRoutingConfigurationsLocked() {
 	for _, fc := range l.activeFilterChainManager.FilterChains() {
 		if fc.InlineRouteConfig != nil {
-			fc.UsableRouteConfiguration.Store(fc.ConstructUsableRouteConfiguration(*fc.InlineRouteConfig)) // Can't race with an RPC coming in but no harm making atomic.
+			urc := fc.ConstructUsableRouteConfiguration(*fc.InlineRouteConfig)
+			urc.NodeID = l.xdsNodeID
+			fc.UsableRouteConfiguration.Store(urc) // Can't race with an RPC coming in but no harm making atomic.
 			continue
 		} // Inline configuration constructed once here, will remain for lifetime of filter chain.
 		rcu := l.rdsHandler.updates[fc.RouteConfigName]
 		if rcu.err != nil && rcu.data == nil {
-			fc.UsableRouteConfiguration.Store(&xdsresource.UsableRouteConfiguration{Err: rcu.err})
+			urc := &xdsresource.UsableRouteConfiguration{Err: rcu.err}
+			urc.NodeID = l.xdsNodeID
+			fc.UsableRouteConfiguration.Store(urc)
 			continue
 		}
-		fc.UsableRouteConfiguration.Store(fc.ConstructUsableRouteConfiguration(*rcu.data)) // Can't race with an RPC coming in but no harm making atomic.
+		urc := fc.ConstructUsableRouteConfiguration(*rcu.data)
+		urc.NodeID = l.xdsNodeID
+		fc.UsableRouteConfiguration.Store(urc) // Can't race with an RPC coming in but no harm making atomic.
 	}
 }
 
@@ -397,10 +408,14 @@ func (l *listenerWrapper) switchModeLocked(newMode connectivity.ServingMode, err
 	}
 }
 
+func (l *listenerWrapper) annotateErrorWithNodeID(err error) error {
+	return fmt.Errorf("[xDS node id: %v]: %w", l.xdsNodeID, err)
+}
+
 func (l *listenerWrapper) onLDSResourceDoesNotExist(err error) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
-	l.switchModeLocked(connectivity.ServingModeNotServing, err)
+	l.switchModeLocked(connectivity.ServingModeNotServing, l.annotateErrorWithNodeID(err))
 	l.activeFilterChainManager = nil
 	l.pendingFilterChainManager = nil
 	l.rdsHandler.updateRouteNamesToWatch(make(map[string]bool))
diff --git a/xds/internal/server/rds_handler.go b/xds/internal/server/rds_handler.go
index bcd3938e..b4d536ff 100644
--- a/xds/internal/server/rds_handler.go
+++ b/xds/internal/server/rds_handler.go
@@ -19,6 +19,7 @@
 package server
 
 import (
+	"fmt"
 	"sync"
 
 	igrpclog "google.golang.org/grpc/internal/grpclog"
@@ -30,8 +31,9 @@ import (
 // updates for later use and also determines whether all the rdsWatcher updates
 // needed have been received or not.
 type rdsHandler struct {
-	xdsC   XDSClient
-	logger *igrpclog.PrefixLogger
+	xdsC      XDSClient
+	xdsNodeID string
+	logger    *igrpclog.PrefixLogger
 
 	callback func(string, rdsWatcherUpdate)
 
@@ -50,13 +52,15 @@ type rdsHandler struct {
 // resources. listenerWrapper updates the list of route names to watch by
 // calling updateRouteNamesToWatch() upon receipt of new Listener configuration.
 func newRDSHandler(cb func(string, rdsWatcherUpdate), xdsC XDSClient, logger *igrpclog.PrefixLogger) *rdsHandler {
-	return &rdsHandler{
+	r := &rdsHandler{
 		xdsC:     xdsC,
 		logger:   logger,
 		callback: cb,
 		updates:  make(map[string]rdsWatcherUpdate),
 		cancels:  make(map[string]func()),
 	}
+	r.xdsNodeID = xdsC.BootstrapConfig().Node().GetId()
+	return r
 }
 
 // updateRouteNamesToWatch handles a list of route names to watch for a given
@@ -184,8 +188,9 @@ func (rw *rdsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
 	}
 	rw.mu.Unlock()
 	if rw.logger.V(2) {
-		rw.logger.Infof("RDS watch for resource %q reported resource-does-not-exist error: %v", rw.routeName)
+		rw.logger.Infof("RDS watch for resource %q reported resource-does-not-exist error", rw.routeName)
 	}
 	err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "resource name %q of type RouteConfiguration not found in received response", rw.routeName)
+	err = fmt.Errorf("[xDS node id: %v]: %w", rw.parent.xdsNodeID, err)
 	rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{err: err})
 }
diff --git a/xds/internal/xdsclient/xdsresource/errors.go b/xds/internal/xdsclient/xdsresource/errors.go
index 06909fbd..f90d30b3 100644
--- a/xds/internal/xdsclient/xdsresource/errors.go
+++ b/xds/internal/xdsclient/xdsresource/errors.go
@@ -70,9 +70,9 @@ func NewError(t ErrorType, message string) error {
 }
 
 // ErrType returns the error's type.
-func ErrType(e error) ErrorType {
+func ErrType(err error) ErrorType {
 	var xe *xdsClientError
-	if ok := errors.As(e, &xe); ok {
+	if errors.As(err, &xe) {
 		return xe.t
 	}
 	return ErrorTypeUnknown
diff --git a/xds/internal/xdsclient/xdsresource/filter_chain.go b/xds/internal/xdsclient/xdsresource/filter_chain.go
index 56a94dea..39451991 100644
--- a/xds/internal/xdsclient/xdsresource/filter_chain.go
+++ b/xds/internal/xdsclient/xdsresource/filter_chain.go
@@ -23,7 +23,9 @@ import (
 	"net"
 	"sync/atomic"
 
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/internal/resolver"
+	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/xds/internal/httpfilter"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 	"google.golang.org/protobuf/proto"
@@ -99,8 +101,15 @@ type RouteWithInterceptors struct {
 // UsableRouteConfiguration contains a matchable route configuration, with
 // instantiated HTTP Filters per route.
 type UsableRouteConfiguration struct {
-	VHS []VirtualHostWithInterceptors
-	Err error
+	VHS    []VirtualHostWithInterceptors
+	Err    error
+	NodeID string // For logging purposes. Populated by the listener wrapper.
+}
+
+// StatusErrWithNodeID returns an error produced by the status package with the
+// specified code and message, and includes the xDS node ID.
+func (rc *UsableRouteConfiguration) StatusErrWithNodeID(c codes.Code, msg string, args ...any) error {
+	return status.Error(c, fmt.Sprintf("[xDS node id: %v]: %s", rc.NodeID, fmt.Sprintf(msg, args...)))
 }
 
 // ConstructUsableRouteConfiguration takes Route Configuration and converts it
diff --git a/xds/server.go b/xds/server.go
index aa93130d..5baf91de 100644
--- a/xds/server.go
+++ b/xds/server.go
@@ -256,7 +256,7 @@ func routeAndProcess(ctx context.Context) error {
 		if logger.V(2) {
 			logger.Infof("RPC on connection with xDS Configuration error: %v", rc.Err)
 		}
-		return status.Error(codes.Unavailable, "error from xDS configuration for matched route configuration")
+		return status.Error(codes.Unavailable, fmt.Sprintf("error from xDS configuration for matched route configuration: %v", rc.Err))
 	}
 
 	mn, ok := grpc.Method(ctx)
@@ -273,7 +273,7 @@ func routeAndProcess(ctx context.Context) error {
 	authority := md.Get(":authority")
 	vh := xdsresource.FindBestMatchingVirtualHostServer(authority[0], rc.VHS)
 	if vh == nil {
-		return status.Error(codes.Unavailable, "the incoming RPC did not match a configured Virtual Host")
+		return rc.StatusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Virtual Host")
 	}
 
 	var rwi *xdsresource.RouteWithInterceptors
@@ -283,21 +283,22 @@ func routeAndProcess(ctx context.Context) error {
 	}
 	for _, r := range vh.Routes {
 		if r.M.Match(rpcInfo) {
-			// "NonForwardingAction is expected for all Routes used on server-side; a route with an inappropriate action causes
-			// RPCs matching that route to fail with UNAVAILABLE." - A36
+			// "NonForwardingAction is expected for all Routes used on
+			// server-side; a route with an inappropriate action causes RPCs
+			// matching that route to fail with UNAVAILABLE." - A36
 			if r.ActionType != xdsresource.RouteActionNonForwardingAction {
-				return status.Error(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding")
+				return rc.StatusErrWithNodeID(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding")
 			}
 			rwi = &r
 			break
 		}
 	}
 	if rwi == nil {
-		return status.Error(codes.Unavailable, "the incoming RPC did not match a configured Route")
+		return rc.StatusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Route")
 	}
 	for _, interceptor := range rwi.Interceptors {
 		if err := interceptor.AllowRPC(ctx); err != nil {
-			return status.Errorf(codes.PermissionDenied, "Incoming RPC is not allowed: %v", err)
+			return rc.StatusErrWithNodeID(codes.PermissionDenied, "Incoming RPC is not allowed: %v", err)
 		}
 	}
 	return nil



Test patch:
diff --git a/balancer/endpointsharding/endpointsharding_test.go b/balancer/endpointsharding/endpointsharding_test.go
index 514d4828..ebad977e 100644
--- a/balancer/endpointsharding/endpointsharding_test.go
+++ b/balancer/endpointsharding/endpointsharding_test.go
@@ -187,7 +187,7 @@ func (s) TestEndpointShardingBasic(t *testing.T) {
 
 	// When the resolver reports an error, the picker should get updated to
 	// return the resolver error.
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
 		_, err := client.EmptyCall(ctx, &testpb.Empty{})
diff --git a/balancer/grpclb/grpclb_test.go b/balancer/grpclb/grpclb_test.go
index e234ace6..b4789620 100644
--- a/balancer/grpclb/grpclb_test.go
+++ b/balancer/grpclb/grpclb_test.go
@@ -859,7 +859,7 @@ func (s) TestGRPCLB_Fallback(t *testing.T) {
 	// Push another update to the resolver, this time with a valid balancer
 	// address in the attributes field.
 	rs = resolver.State{
-		ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+		ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
 	}
 	rs = grpclbstate.Set(rs, &grpclbstate.State{BalancerAddresses: []resolver.Address{{Addr: tss.lbAddr, ServerName: lbServerName}}})
@@ -1023,7 +1023,7 @@ func (s) TestGRPCLB_FallBackWithNoServerAddress(t *testing.T) {
 		// fallback and use the fallback backend.
 		r.UpdateState(resolver.State{
 			Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
-			ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+			ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		})
 
 		sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
@@ -1051,7 +1051,7 @@ func (s) TestGRPCLB_FallBackWithNoServerAddress(t *testing.T) {
 		// be used.
 		rs := resolver.State{
 			Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
-			ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+			ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		}
 		rs = grpclbstate.Set(rs, &grpclbstate.State{BalancerAddresses: []resolver.Address{{Addr: tss.lbAddr, ServerName: lbServerName}}})
 		r.UpdateState(rs)
@@ -1112,7 +1112,7 @@ func (s) TestGRPCLB_PickFirst(t *testing.T) {
 
 	// Push a service config with grpclb as the load balancing policy and
 	// configure pick_first as its child policy.
-	rs := resolver.State{ServiceConfig: r.CC.ParseServiceConfig(`{"loadBalancingConfig":[{"grpclb":{"childPolicy":[{"pick_first":{}}]}}]}`)}
+	rs := resolver.State{ServiceConfig: r.CC().ParseServiceConfig(`{"loadBalancingConfig":[{"grpclb":{"childPolicy":[{"pick_first":{}}]}}]}`)}
 
 	// Push a resolver update with the remote balancer address specified via
 	// attributes.
@@ -1152,7 +1152,7 @@ func (s) TestGRPCLB_PickFirst(t *testing.T) {
 			},
 		},
 	}
-	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)}, s)
+	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)}, s)
 	r.UpdateState(rs)
 	testC := testgrpc.NewTestServiceClient(cc)
 	if err := roundrobin.CheckRoundRobinRPCs(ctx, testC, beServerAddrs[1:]); err != nil {
@@ -1261,7 +1261,7 @@ func testGRPCLBEmptyServerList(t *testing.T, svcfg string) {
 			},
 		},
 	}
-	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(svcfg)}, s)
+	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(svcfg)}, s)
 	r.UpdateState(rs)
 	t.Log("Perform an initial RPC and expect it to succeed...")
 	if _, err := testC.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
@@ -1329,7 +1329,7 @@ func (s) TestGRPCLBWithTargetNameFieldInConfig(t *testing.T) {
 	// Push a resolver update with grpclb configuration which does not contain the
 	// target_name field. Our fake remote balancer is configured to always
 	// expect `beServerName` as the server name in the initial request.
-	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)},
+	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)},
 		&grpclbstate.State{BalancerAddresses: []resolver.Address{{
 			Addr:       tss.lbAddr,
 			ServerName: lbServerName,
@@ -1366,7 +1366,7 @@ func (s) TestGRPCLBWithTargetNameFieldInConfig(t *testing.T) {
 			},
 		},
 	}
-	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(lbCfg)}, s)
+	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(lbCfg)}, s)
 	r.UpdateState(rs)
 	select {
 	case <-ctx.Done():
@@ -1432,7 +1432,7 @@ func runAndCheckStats(t *testing.T, drop bool, statsChan chan *lbpb.ClientStats,
 	cc.Connect()
 	defer cc.Close()
 
-	rstate := resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)}
+	rstate := resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)}
 	r.UpdateState(grpclbstate.Set(rstate, &grpclbstate.State{BalancerAddresses: []resolver.Address{{
 		Addr:       tss.lbAddr,
 		ServerName: lbServerName,
diff --git a/balancer/lazy/lazy_ext_test.go b/balancer/lazy/lazy_ext_test.go
index c2470ef7..c57ad28e 100644
--- a/balancer/lazy/lazy_ext_test.go
+++ b/balancer/lazy/lazy_ext_test.go
@@ -271,7 +271,7 @@ func (s) TestGoodUpdateThenResolverError(t *testing.T) {
 	defer cc.Close()
 	cc.Connect()
 
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	// The channel should remain in IDLE as the ExitIdle calls are not
 	// propagated to the lazy balancer from the stub balancer.
 	shortCtx, shortCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
@@ -367,7 +367,7 @@ func (s) TestResolverErrorThenGoodUpdate(t *testing.T) {
 	cc.Connect()
 
 	// Send an error followed by a good update.
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	mr.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{Addresses: []resolver.Address{{Addr: backend.Address}}},
diff --git a/balancer/pickfirst/pickfirst_ext_test.go b/balancer/pickfirst/pickfirst_ext_test.go
index 4354009f..81691fc8 100644
--- a/balancer/pickfirst/pickfirst_ext_test.go
+++ b/balancer/pickfirst/pickfirst_ext_test.go
@@ -73,7 +73,7 @@ func Test(t *testing.T) {
 func parseServiceConfig(t *testing.T, r *manual.Resolver, sc string) *serviceconfig.ParseResult {
 	t.Helper()
 
-	scpr := r.CC.ParseServiceConfig(sc)
+	scpr := r.CC().ParseServiceConfig(sc)
 	if scpr.Err != nil {
 		t.Fatalf("Failed to parse service config %q: %v", sc, scpr.Err)
 	}
@@ -756,7 +756,7 @@ func (s) TestPickFirst_ResolverError_NoPreviousUpdate(t *testing.T) {
 	cc, r, _ := setupPickFirst(t, 0)
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -789,7 +789,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_Ready(t *testing.T) {
 	}
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	// Ensure that RPCs continue to succeed for the next second.
 	client := testgrpc.NewTestServiceClient(cc)
@@ -848,7 +848,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_Connecting(t *testing.T)
 	testutils.AwaitState(ctx, t, cc, connectivity.Connecting)
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	// RPCs should fail with deadline exceed error as long as they are in
 	// CONNECTING and not the error returned by the name resolver.
@@ -909,7 +909,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_TransientFailure(t *test
 	// error instead of the old error that caused the channel to move to
 	// TRANSIENT_FAILURE in the first place.
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 	client := testgrpc.NewTestServiceClient(cc)
 	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
 		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); strings.Contains(err.Error(), nrErr.Error()) {
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
index 33970efb..a876fcb0 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
@@ -191,7 +191,7 @@ func (s) TestPickFirstLeaf_SimpleResolverUpdate_FirstServerReady(t *testing.T) {
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -233,7 +233,7 @@ func (s) TestPickFirstLeaf_SimpleResolverUpdate_FirstServerUnReady(t *testing.T)
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -278,7 +278,7 @@ func (s) TestPickFirstLeaf_SimpleResolverUpdate_DuplicateAddrs(t *testing.T) {
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -350,7 +350,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_DisjointLists(t *testing.T) {
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -412,7 +412,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_ActiveBackendInUpdatedList(t *testing
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -474,7 +474,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_InActiveBackendInUpdatedList(t *testi
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -535,7 +535,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_IdenticalLists(t *testing.T) {
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -608,7 +608,7 @@ func (s) TestPickFirstLeaf_StopConnectedServer_FirstServerRestart(t *testing.T)
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -677,7 +677,7 @@ func (s) TestPickFirstLeaf_StopConnectedServer_SecondServerRestart(t *testing.T)
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -746,7 +746,7 @@ func (s) TestPickFirstLeaf_StopConnectedServer_SecondServerToFirst(t *testing.T)
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -813,7 +813,7 @@ func (s) TestPickFirstLeaf_StopConnectedServer_FirstServerToSecond(t *testing.T)
 		connectivity.Connecting,
 		connectivity.Ready,
 	}
-	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantConnStateTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -866,7 +866,7 @@ func (s) TestPickFirstLeaf_EmptyAddressList(t *testing.T) {
 		connectivity.Ready,
 	}
 
-	if diff := cmp.Diff(wantTransitions, stateSubscriber.transitions); diff != "" {
+	if diff := cmp.Diff(wantTransitions, stateSubscriber.transitions()); diff != "" {
 		t.Errorf("ClientConn states mismatch (-want +got):\n%s", diff)
 	}
 }
@@ -1606,11 +1606,23 @@ func (b *backendManager) holds(dialer *testutils.BlockingDialer) []*testutils.Ho
 }
 
 type ccStateSubscriber struct {
-	transitions []connectivity.State
+	mu     sync.Mutex
+	states []connectivity.State
+}
+
+// transitions returns all the states that ccStateSubscriber recorded.
+// Without this a race condition occurs when the test compares the states
+// and the subscriber at the same time receives a connectivity.Shutdown.
+func (c *ccStateSubscriber) transitions() []connectivity.State {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.states
 }
 
 func (c *ccStateSubscriber) OnMessage(msg any) {
-	c.transitions = append(c.transitions, msg.(connectivity.State))
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.states = append(c.states, msg.(connectivity.State))
 }
 
 // mockTimer returns a fake timeAfterFunc that will not trigger automatically.
diff --git a/balancer/weightedroundrobin/balancer_test.go b/balancer/weightedroundrobin/balancer_test.go
index 5e369780..a5b74203 100644
--- a/balancer/weightedroundrobin/balancer_test.go
+++ b/balancer/weightedroundrobin/balancer_test.go
@@ -449,7 +449,7 @@ func (s) TestBalancer_TwoAddresses_OOBThenPerCall(t *testing.T) {
 
 	// Update to per-call weights.
 	c := svcConfig(t, perCallConfig)
-	parsedCfg := srv1.R.CC.ParseServiceConfig(c)
+	parsedCfg := srv1.R.CC().ParseServiceConfig(c)
 	if parsedCfg.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", c, parsedCfg.Err))
 	}
@@ -563,7 +563,7 @@ func (s) TestBalancer_TwoAddresses_ErrorPenalty(t *testing.T) {
 	newCfg := oobConfig
 	newCfg.ErrorUtilizationPenalty = float64p(0.9)
 	c := svcConfig(t, newCfg)
-	parsedCfg := srv1.R.CC.ParseServiceConfig(c)
+	parsedCfg := srv1.R.CC().ParseServiceConfig(c)
 	if parsedCfg.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", c, parsedCfg.Err))
 	}
diff --git a/clientconn_test.go b/clientconn_test.go
index 3b23c49f..691c007f 100644
--- a/clientconn_test.go
+++ b/clientconn_test.go
@@ -62,7 +62,7 @@ func init() {
 }
 
 func parseCfg(r *manual.Resolver, s string) *serviceconfig.ParseResult {
-	scpr := r.CC.ParseServiceConfig(s)
+	scpr := r.CC().ParseServiceConfig(s)
 	if scpr.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", s, scpr.Err))
 	}
@@ -666,7 +666,7 @@ func (s) TestResolverServiceConfigBeforeAddressNotPanic(t *testing.T) {
 	cc.Connect()
 	// SwitchBalancer before NewAddress. There was no balancer created, this
 	// makes sure we don't call close on nil balancerWrapper.
-	r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbServiceConfig)}) // This should not panic.
+	r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbServiceConfig)}) // This should not panic.
 
 	time.Sleep(time.Second) // Sleep to make sure the service config is handled by ClientConn.
 }
@@ -681,7 +681,7 @@ func (s) TestResolverServiceConfigWhileClosingNotPanic(t *testing.T) {
 		cc.Connect()
 		// Send a new service config while closing the ClientConn.
 		go cc.Close()
-		go r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(rrServiceConfig)}) // This should not panic.
+		go r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(rrServiceConfig)}) // This should not panic.
 	}
 }
 
@@ -775,7 +775,7 @@ func (s) TestDisableServiceConfigOption(t *testing.T) {
 	}
 	defer cc.Close()
 	cc.Connect()
-	r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(`{
+	r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(`{
     "methodConfig": [
         {
             "name": [
diff --git a/internal/testutils/xds/e2e/clientresources.go b/internal/testutils/xds/e2e/clientresources.go
index 0bfcad18..e94d226f 100644
--- a/internal/testutils/xds/e2e/clientresources.go
+++ b/internal/testutils/xds/e2e/clientresources.go
@@ -140,72 +140,6 @@ func marshalAny(m proto.Message) *anypb.Any {
 	return a
 }
 
-// filterChainWontMatch returns a filter chain that won't match if running the
-// test locally.
-func filterChainWontMatch(routeName string, addressPrefix string, srcPorts []uint32) *v3listenerpb.FilterChain {
-	hcm := &v3httppb.HttpConnectionManager{
-		RouteSpecifier: &v3httppb.HttpConnectionManager_Rds{
-			Rds: &v3httppb.Rds{
-				ConfigSource: &v3corepb.ConfigSource{
-					ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{Ads: &v3corepb.AggregatedConfigSource{}},
-				},
-				RouteConfigName: routeName,
-			},
-		},
-		HttpFilters: []*v3httppb.HttpFilter{RouterHTTPFilter},
-	}
-	return &v3listenerpb.FilterChain{
-		Name: routeName + "-wont-match",
-		FilterChainMatch: &v3listenerpb.FilterChainMatch{
-			PrefixRanges: []*v3corepb.CidrRange{
-				{
-					AddressPrefix: addressPrefix,
-					PrefixLen: &wrapperspb.UInt32Value{
-						Value: uint32(0),
-					},
-				},
-			},
-			SourceType:  v3listenerpb.FilterChainMatch_SAME_IP_OR_LOOPBACK,
-			SourcePorts: srcPorts,
-			SourcePrefixRanges: []*v3corepb.CidrRange{
-				{
-					AddressPrefix: addressPrefix,
-					PrefixLen: &wrapperspb.UInt32Value{
-						Value: uint32(0),
-					},
-				},
-			},
-		},
-		Filters: []*v3listenerpb.Filter{
-			{
-				Name:       "filter-1",
-				ConfigType: &v3listenerpb.Filter_TypedConfig{TypedConfig: marshalAny(hcm)},
-			},
-		},
-	}
-}
-
-// ListenerResourceThreeRouteResources returns a listener resource that points
-// to three route configurations. Only the filter chain that points to the first
-// route config can be matched to.
-func ListenerResourceThreeRouteResources(host string, port uint32, secLevel SecurityLevel, routeName string) *v3listenerpb.Listener {
-	lis := defaultServerListenerCommon(host, port, secLevel, routeName, false)
-	lis.FilterChains = append(lis.FilterChains, filterChainWontMatch("routeName2", "1.1.1.1", []uint32{1}))
-	lis.FilterChains = append(lis.FilterChains, filterChainWontMatch("routeName3", "2.2.2.2", []uint32{2}))
-	return lis
-}
-
-// ListenerResourceFallbackToDefault returns a listener resource that contains a
-// filter chain that will never get chosen to process traffic and a default
-// filter chain. The default filter chain points to routeName2.
-func ListenerResourceFallbackToDefault(host string, port uint32, secLevel SecurityLevel) *v3listenerpb.Listener {
-	lis := defaultServerListenerCommon(host, port, secLevel, "", false)
-	lis.FilterChains = nil
-	lis.FilterChains = append(lis.FilterChains, filterChainWontMatch("routeName", "1.1.1.1", []uint32{1}))
-	lis.DefaultFilterChain = filterChainWontMatch("routeName2", "2.2.2.2", []uint32{2})
-	return lis
-}
-
 // DefaultServerListener returns a basic xds Listener resource to be used on the
 // server side. The returned Listener resource contains an inline route
 // configuration with the name of routeName.
diff --git a/resolver/manual/manual_test.go b/resolver/manual/manual_test.go
index 9f1701b1..5a27dbac 100644
--- a/resolver/manual/manual_test.go
+++ b/resolver/manual/manual_test.go
@@ -36,26 +36,14 @@ func TestResolver(t *testing.T) {
 		},
 	})
 
-	t.Run("update_state_panics", func(t *testing.T) {
+	t.Run("cc_panics", func(t *testing.T) {
 		defer func() {
-			want := "cannot update state as channel has not exited IDLE state"
+			want := "Manual resolver instance has not yet been built."
 			if r := recover(); r != want {
 				t.Errorf("expected panic %q, got %q", want, r)
 			}
 		}()
-		r.UpdateState(resolver.State{Addresses: []resolver.Address{
-			{Addr: "address"},
-			{Addr: "anotheraddress"},
-		}})
-	})
-	t.Run("report_error_panics", func(t *testing.T) {
-		defer func() {
-			want := "cannot report error as channel has not exited IDLE state"
-			if r := recover(); r != want {
-				t.Errorf("expected panic %q, got %q", want, r)
-			}
-		}()
-		r.ReportError(errors.New("example"))
+		r.CC()
 	})
 
 	t.Run("happy_path", func(t *testing.T) {
@@ -70,6 +58,6 @@ func TestResolver(t *testing.T) {
 		r.UpdateState(resolver.State{Addresses: []resolver.Address{
 			{Addr: "ok"},
 		}})
-		r.ReportError(errors.New("example"))
+		r.CC().ReportError(errors.New("example"))
 	})
 }
diff --git a/resolver_test.go b/resolver_test.go
index ddd6a2ea..9262f6ce 100644
--- a/resolver_test.go
+++ b/resolver_test.go
@@ -144,6 +144,58 @@ func (s) TestResolverAddressesToEndpoints(t *testing.T) {
 	}
 }
 
+// Test ensures one Endpoint is created for each entry in
+// resolver.State.Addresses automatically. The test calls the deprecated
+// NewAddresses API to send a list of addresses to the channel.
+func (s) TestResolverAddressesToEndpointsUsingNewAddresses(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	const scheme = "testresolveraddressestoendpoints"
+	r := manual.NewBuilderWithScheme(scheme)
+
+	stateCh := make(chan balancer.ClientConnState, 1)
+	bf := stub.BalancerFuncs{
+		UpdateClientConnState: func(_ *stub.BalancerData, ccs balancer.ClientConnState) error {
+			stateCh <- ccs
+			return nil
+		},
+	}
+	balancerName := "stub-balancer-" + scheme
+	stub.Register(balancerName, bf)
+
+	a1 := attributes.New("x", "y")
+	a2 := attributes.New("a", "b")
+	addrs := []resolver.Address{
+		{Addr: "addr1", BalancerAttributes: a1},
+		{Addr: "addr2", BalancerAttributes: a2},
+	}
+
+	cc, err := NewClient(r.Scheme()+":///",
+		WithTransportCredentials(insecure.NewCredentials()),
+		WithResolvers(r),
+		WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%s":{}}]}`, balancerName)))
+	if err != nil {
+		t.Fatalf("grpc.NewClient() failed: %v", err)
+	}
+	cc.Connect()
+	defer cc.Close()
+	r.CC().NewAddress(addrs)
+
+	select {
+	case got := <-stateCh:
+		want := []resolver.Endpoint{
+			{Addresses: []resolver.Address{{Addr: "addr1"}}, Attributes: a1},
+			{Addresses: []resolver.Address{{Addr: "addr2"}}, Attributes: a2},
+		}
+		if diff := cmp.Diff(got.ResolverState.Endpoints, want); diff != "" {
+			t.Errorf("Did not receive expected endpoints.  Diff (-got +want):\n%v", diff)
+		}
+	case <-ctx.Done():
+		t.Fatalf("timed out waiting for endpoints")
+	}
+}
+
 // Test ensures that there is no panic if the attributes within
 // resolver.State.Addresses contains a typed-nil value.
 func (s) TestResolverAddressesWithTypedNilAttribute(t *testing.T) {
diff --git a/test/balancer_switching_test.go b/test/balancer_switching_test.go
index 9b9b890a..2c4672f3 100644
--- a/test/balancer_switching_test.go
+++ b/test/balancer_switching_test.go
@@ -510,7 +510,7 @@ func (s) TestBalancerSwitch_Graceful(t *testing.T) {
 	// report a ready picker until we ask it to do so.
 	r.UpdateState(resolver.State{
 		Addresses:     addrs[:1],
-		ServiceConfig: r.CC.ParseServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%v": {}}]}`, t.Name())),
+		ServiceConfig: r.CC().ParseServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%v": {}}]}`, t.Name())),
 	})
 	select {
 	case <-ctx.Done():
diff --git a/test/balancer_test.go b/test/balancer_test.go
index 9fe9a2c5..8c26ea1b 100644
--- a/test/balancer_test.go
+++ b/test/balancer_test.go
@@ -631,7 +631,7 @@ func (s) TestWaitForReady(t *testing.T) {
 	client := testgrpc.NewTestServiceClient(cc)
 
 	// Report an error so non-WFR RPCs will give up early.
-	r.CC.ReportError(errors.New("fake resolver error"))
+	r.CC().ReportError(errors.New("fake resolver error"))
 
 	// Ensure the client is not connected to anything and fails non-WFR RPCs.
 	if res, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Unavailable {
diff --git a/test/parse_config.go b/test/parse_config.go
index f375a3aa..b8355c79 100644
--- a/test/parse_config.go
+++ b/test/parse_config.go
@@ -30,7 +30,7 @@ import (
 func parseServiceConfig(t *testing.T, r *manual.Resolver, sc string) *serviceconfig.ParseResult {
 	t.Helper()
 
-	scpr := r.CC.ParseServiceConfig(sc)
+	scpr := r.CC().ParseServiceConfig(sc)
 	if scpr.Err != nil {
 		t.Fatalf("Failed to parse service config %q: %v", sc, scpr.Err)
 	}
diff --git a/test/resolver_update_test.go b/test/resolver_update_test.go
index b8a2d443..141cd553 100644
--- a/test/resolver_update_test.go
+++ b/test/resolver_update_test.go
@@ -119,7 +119,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAsFirstUpdate(t *testing.T) {
 	cc.Connect()
 	defer cc.Close()
 
-	scpr := r.CC.ParseServiceConfig("bad json service config")
+	scpr := r.CC().ParseServiceConfig("bad json service config")
 	r.UpdateState(resolver.State{ServiceConfig: scpr})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -205,7 +205,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAfterGoodUpdate(t *testing.T) {
 	// Push a resolver update and verify that our balancer receives the update.
 	addrs := []resolver.Address{{Addr: backend.Address}}
 	const lbCfg = "wrapping balancer LB policy config"
-	goodSC := r.CC.ParseServiceConfig(fmt.Sprintf(`
+	goodSC := r.CC().ParseServiceConfig(fmt.Sprintf(`
 {
   "loadBalancingConfig": [
     {
@@ -244,7 +244,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAfterGoodUpdate(t *testing.T) {
 	// Push a bad resolver update and ensure that the update is propagated to our
 	// stub balancer. But since the pushed update contains an invalid service
 	// config, our balancer should continue to see the old loadBalancingConfig.
-	badSC := r.CC.ParseServiceConfig("bad json service config")
+	badSC := r.CC().ParseServiceConfig("bad json service config")
 	wantCCS.ResolverState.ServiceConfig = badSC
 	r.UpdateState(resolver.State{Addresses: addrs, ServiceConfig: badSC})
 	ccs, err = ccUpdateCh.Receive(ctx)
diff --git a/test/xds/xds_client_ignore_resource_deletion_test.go b/test/xds/xds_client_ignore_resource_deletion_test.go
index 8210d472..d4bc7ee1 100644
--- a/test/xds/xds_client_ignore_resource_deletion_test.go
+++ b/test/xds/xds_client_ignore_resource_deletion_test.go
@@ -23,12 +23,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
@@ -37,6 +39,7 @@ import (
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/xds"
 
 	clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
@@ -201,7 +204,7 @@ func testResourceDeletionIgnored(t *testing.T, initialResource func(string) e2e.
 // deleted by the xDSClient when a resource is missing the xDS response and subsequent
 // RPCs fail.
 func testResourceDeletionNotIgnored(t *testing.T, initialResource func(string) e2e.UpdateOptions, updateResource func(r *e2e.UpdateOptions)) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout*1000)
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	t.Cleanup(cancel)
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 	nodeID := uuid.New().String()
@@ -230,14 +233,19 @@ func testResourceDeletionNotIgnored(t *testing.T, initialResource func(string) e
 		t.Fatal(err)
 	}
 
-	// Spin up go routines to verify RPCs fail after the update.
+	// Spin up go routines to verify RPCs fail after the update. The xDS node ID
+	// needs to be part of the error seen by the RPC caller.
 	client := testgrpc.NewTestServiceClient(cc)
 	wg := sync.WaitGroup{}
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
 		for ; ctx.Err() == nil; <-time.After(10 * time.Millisecond) {
-			if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+			_, err := client.EmptyCall(ctx, &testpb.Empty{})
+			if err == nil {
+				continue
+			}
+			if status.Code(err) == codes.Unavailable && strings.Contains(err.Error(), nodeID) {
 				return
 			}
 		}
@@ -245,7 +253,11 @@ func testResourceDeletionNotIgnored(t *testing.T, initialResource func(string) e
 	go func() {
 		defer wg.Done()
 		for ; ctx.Err() == nil; <-time.After(10 * time.Millisecond) {
-			if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
+			_, err := client.UnaryCall(ctx, &testpb.SimpleRequest{})
+			if err == nil {
+				continue
+			}
+			if status.Code(err) == codes.Unavailable && strings.Contains(err.Error(), nodeID) {
 				return
 			}
 		}
@@ -276,7 +288,7 @@ func generateBootstrapContents(t *testing.T, serverURI string, ignoreResourceDel
 	}
 	bootstrapContents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
 		Servers:                            serverCfgs,
-		Node:                               []byte(fmt.Sprintf(`{"id": "%s"}`, nodeID)),
+		Node:                               fmt.Appendf(nil, `{"id": "%s"}`, nodeID),
 		ServerListenerResourceNameTemplate: e2e.ServerListenerResourceNameTemplate,
 	})
 	if err != nil {
diff --git a/test/xds/xds_server_rbac_test.go b/test/xds/xds_server_rbac_test.go
index 5c618911..c0e51f5f 100644
--- a/test/xds/xds_server_rbac_test.go
+++ b/test/xds/xds_server_rbac_test.go
@@ -265,9 +265,13 @@ func (s) TestServerSideXDS_RouteConfiguration(t *testing.T) {
 	// This Unary Call should match to a route with an incorrect action. Thus,
 	// this RPC should not go through as per A36, and this call should receive
 	// an error with codes.Unavailable.
-	if _, err = client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Unavailable {
+	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
+	if status.Code(err) != codes.Unavailable {
 		t.Fatalf("client.UnaryCall() = _, %v, want _, error code %s", err, codes.Unavailable)
 	}
+	if !strings.Contains(err.Error(), nodeID) {
+		t.Fatalf("client.UnaryCall() = %v, want xDS node id %q", err, nodeID)
+	}
 
 	// This Streaming Call should match to a route with an incorrect action.
 	// Thus, this RPC should not go through as per A36, and this call should
@@ -276,8 +280,13 @@ func (s) TestServerSideXDS_RouteConfiguration(t *testing.T) {
 	if err != nil {
 		t.Fatalf("StreamingInputCall(_) = _, %v, want <nil>", err)
 	}
-	if _, err = stream.CloseAndRecv(); status.Code(err) != codes.Unavailable || !strings.Contains(err.Error(), "the incoming RPC matched to a route that was not of action type non forwarding") {
-		t.Fatalf("streaming RPC should have been denied")
+	_, err = stream.CloseAndRecv()
+	const wantStreamingErr = "the incoming RPC matched to a route that was not of action type non forwarding"
+	if status.Code(err) != codes.Unavailable || !strings.Contains(err.Error(), wantStreamingErr) {
+		t.Fatalf("client.StreamingInputCall() = %v, want error with code %s and message %q", err, codes.Unavailable, wantStreamingErr)
+	}
+	if !strings.Contains(err.Error(), nodeID) {
+		t.Fatalf("client.StreamingInputCall() = %v, want xDS node id %q", err, nodeID)
 	}
 
 	// This Full Duplex should not match to a route, and thus should return an
@@ -286,8 +295,13 @@ func (s) TestServerSideXDS_RouteConfiguration(t *testing.T) {
 	if err != nil {
 		t.Fatalf("FullDuplexCall(_) = _, %v, want <nil>", err)
 	}
-	if _, err = dStream.Recv(); status.Code(err) != codes.Unavailable || !strings.Contains(err.Error(), "the incoming RPC did not match a configured Route") {
-		t.Fatalf("streaming RPC should have been denied")
+	_, err = dStream.Recv()
+	const wantFullDuplexErr = "the incoming RPC did not match a configured Route"
+	if status.Code(err) != codes.Unavailable || !strings.Contains(err.Error(), wantFullDuplexErr) {
+		t.Fatalf("client.FullDuplexCall() = %v, want error with code %s and message %q", err, codes.Unavailable, wantFullDuplexErr)
+	}
+	if !strings.Contains(err.Error(), nodeID) {
+		t.Fatalf("client.FullDuplexCall() = %v, want xDS node id %q", err, nodeID)
 	}
 }
 
@@ -826,7 +840,7 @@ func serverListenerWithBadRouteConfiguration(t *testing.T, host string, port uin
 	}
 }
 
-func (s) TestRBACToggledOn_WithBadRouteConfiguration(t *testing.T) {
+func (s) TestRBAC_WithBadRouteConfiguration(t *testing.T) {
 	managementServer, nodeID, bootstrapContents, xdsResolver := setup.ManagementServerAndResolver(t)
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -867,11 +881,19 @@ func (s) TestRBACToggledOn_WithBadRouteConfiguration(t *testing.T) {
 	defer cc.Close()
 
 	client := testgrpc.NewTestServiceClient(cc)
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Unavailable {
-		t.Fatalf("EmptyCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
+	_, err = client.EmptyCall(ctx, &testpb.Empty{})
+	if status.Code(err) != codes.Unavailable {
+		t.Fatalf("EmptyCall() returned %v, want Unavailable", err)
+	}
+	if !strings.Contains(err.Error(), nodeID) {
+		t.Fatalf("EmptyCall() = %v, want xDS node id %q", err, nodeID)
+	}
+	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
+	if status.Code(err) != codes.Unavailable {
+		t.Fatalf("UnaryCall() returned %v, want Unavailable", err)
 	}
-	if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Unavailable {
-		t.Fatalf("UnaryCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
+	if !strings.Contains(err.Error(), nodeID) {
+		t.Fatalf("UnaryCall() = %v, want xDS node id %q", err, nodeID)
 	}
 }
 
diff --git a/test/xds/xds_server_serving_mode_test.go b/test/xds/xds_server_serving_mode_test.go
deleted file mode 100644
index a2e16489..00000000
--- a/test/xds/xds_server_serving_mode_test.go
+++ /dev/null
@@ -1,401 +0,0 @@
-/*
- *
- * Copyright 2021 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package xds_test
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"testing"
-	"time"
-
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/credentials/insecure"
-	xdscreds "google.golang.org/grpc/credentials/xds"
-	"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/testutils/xds/e2e/setup"
-	"google.golang.org/grpc/status"
-	"google.golang.org/grpc/xds"
-
-	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
-	testgrpc "google.golang.org/grpc/interop/grpc_testing"
-	testpb "google.golang.org/grpc/interop/grpc_testing"
-)
-
-// TestServerSideXDS_RedundantUpdateSuppression tests the scenario where the
-// control plane sends the same resource update. It verifies that the mode
-// change callback is not invoked and client connections to the server are not
-// recycled.
-func (s) TestServerSideXDS_RedundantUpdateSuppression(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-
-	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
-	if err != nil {
-		t.Fatal(err)
-	}
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	updateCh := make(chan connectivity.ServingMode, 1)
-
-	// Create a server option to get notified about serving mode changes.
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		updateCh <- args.Mode
-	})
-
-	// Initialize a test gRPC server, assign it to the stub server, and start
-	// the test service.
-	stub := createStubServer(t, lis, creds, modeChangeOpt, bootstrapContents)
-	defer stub.S.Stop()
-
-	// Setup the management server to respond with the listener resources.
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-	listener := e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, "routeName")
-	resources := e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener},
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for the listener to move to "serving" mode.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
-		}
-	}
-
-	// Create a ClientConn and make a successful RPCs.
-	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
-	}
-	defer cc.Close()
-	waitForSuccessfulRPC(ctx, t, cc)
-
-	// Start a goroutine to make sure that we do not see any connectivity state
-	// changes on the client connection. If redundant updates are not
-	// suppressed, server will recycle client connections.
-	errCh := make(chan error, 1)
-	go func() {
-		prev := connectivity.Ready // We know we are READY since we just did an RPC.
-		for {
-			curr := cc.GetState()
-			if !(curr == connectivity.Ready || curr == connectivity.Idle) {
-				errCh <- fmt.Errorf("unexpected connectivity state change {%s --> %s} on the client connection", prev, curr)
-				return
-			}
-			if !cc.WaitForStateChange(ctx, curr) {
-				// Break out of the for loop when the context has been cancelled.
-				break
-			}
-			prev = curr
-		}
-		errCh <- nil
-	}()
-
-	// Update the management server with the same listener resource. This will
-	// update the resource version though, and should result in a the management
-	// server sending the same resource to the xDS-enabled gRPC server.
-	if err := managementServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
-		t.Fatal(err)
-	}
-
-	// Since redundant resource updates are suppressed, we should not see the
-	// mode change callback being invoked.
-	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
-	defer sCancel()
-	select {
-	case <-sCtx.Done():
-	case mode := <-updateCh:
-		t.Fatalf("unexpected mode change callback with new mode %v", mode)
-	}
-
-	// Make sure RPCs continue to succeed.
-	waitForSuccessfulRPC(ctx, t, cc)
-
-	// Cancel the context to ensure that the WaitForStateChange call exits early
-	// and returns false.
-	cancel()
-	if err := <-errCh; err != nil {
-		t.Fatal(err)
-	}
-}
-
-// TestServerSideXDS_ServingModeChanges tests the serving mode functionality in
-// xDS enabled gRPC servers. It verifies that appropriate mode changes happen in
-// the server, and also verifies behavior of clientConns under these modes.
-func (s) TestServerSideXDS_ServingModeChanges(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-
-	// Configure xDS credentials to be used on the server-side.
-	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{
-		FallbackCreds: insecure.NewCredentials(),
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// Create two local listeners and pass it to Serve().
-	lis1, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	lis2, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-
-	// Create a couple of channels on which mode updates will be pushed.
-	updateCh1 := make(chan connectivity.ServingMode, 1)
-	updateCh2 := make(chan connectivity.ServingMode, 1)
-
-	// Create a server option to get notified about serving mode changes, and
-	// push the updated mode on the channels created above.
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		switch addr.String() {
-		case lis1.Addr().String():
-			updateCh1 <- args.Mode
-		case lis2.Addr().String():
-			updateCh2 <- args.Mode
-		default:
-			t.Errorf("serving mode callback invoked for unknown listener address: %q", addr.String())
-		}
-	})
-
-	// Initialize a test gRPC server, assign it to the stub server, and start
-	// the test service.
-	stub1 := createStubServer(t, lis1, creds, modeChangeOpt, bootstrapContents)
-	defer stub1.S.Stop()
-	stub2 := createStubServer(t, lis2, creds, modeChangeOpt, bootstrapContents)
-	defer stub2.S.Stop()
-
-	// Setup the management server to respond with server-side Listener
-	// resources for both listeners.
-	host1, port1, err := hostPortFromListener(lis1)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-	listener1 := e2e.DefaultServerListener(host1, port1, e2e.SecurityLevelNone, "routeName")
-	host2, port2, err := hostPortFromListener(lis2)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-	listener2 := e2e.DefaultServerListener(host2, port2, e2e.SecurityLevelNone, "routeName")
-	resources := e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener1, listener2},
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for both listeners to move to "serving" mode.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh1:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
-		}
-	}
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh2:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
-		}
-	}
-
-	// Create a ClientConn to the first listener and make a successful RPCs.
-	cc1, err := grpc.NewClient(lis1.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("grpc.NewClient() failed: %v", err)
-	}
-	defer cc1.Close()
-	waitForSuccessfulRPC(ctx, t, cc1)
-
-	// Create a ClientConn to the second listener and make a successful RPCs.
-	cc2, err := grpc.NewClient(lis2.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("grpc.NewClient() failed: %v", err)
-	}
-	defer cc2.Close()
-	waitForSuccessfulRPC(ctx, t, cc2)
-
-	// Update the management server to remove the second listener resource. This
-	// should push only the second listener into "not-serving" mode.
-	if err := managementServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener1},
-	}); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for lis2 to move to "not-serving" mode.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh2:
-		if mode != connectivity.ServingModeNotServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeNotServing)
-		}
-	}
-
-	// Make sure RPCs succeed on cc1 and fail on cc2.
-	waitForSuccessfulRPC(ctx, t, cc1)
-	waitForFailedRPC(ctx, t, cc2)
-
-	// Update the management server to remove the first listener resource as
-	// well. This should push the first listener into "not-serving" mode. Second
-	// listener is already in "not-serving" mode.
-	if err := managementServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{},
-	}); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for lis1 to move to "not-serving" mode. lis2 was already removed
-	// from the xdsclient's resource cache. So, lis2's callback will not be
-	// invoked this time around.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh1:
-		if mode != connectivity.ServingModeNotServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeNotServing)
-		}
-	}
-
-	// Make sure RPCs fail on both.
-	waitForFailedRPC(ctx, t, cc1)
-	waitForFailedRPC(ctx, t, cc2)
-
-	// Make sure new connection attempts to "not-serving" servers fail.
-	if cc1, err = grpc.NewClient(lis1.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
-		t.Fatal("Failed to create clientConn to a server in \"not-serving\" state")
-	}
-	defer cc1.Close()
-
-	if _, err := testgrpc.NewTestServiceClient(cc1).FullDuplexCall(ctx); status.Code(err) != codes.Unavailable {
-		t.Fatalf("FullDuplexCall failed with status code: %v, want: Unavailable", status.Code(err))
-	}
-
-	// Update the management server with both listener resources.
-	if err := managementServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener1, listener2},
-	}); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for both listeners to move to "serving" mode.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh1:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
-		}
-	}
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timed out waiting for a mode change update: %v", err)
-	case mode := <-updateCh2:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
-		}
-	}
-
-	// The clientConns created earlier should be able to make RPCs now.
-	waitForSuccessfulRPC(ctx, t, cc1)
-	waitForSuccessfulRPC(ctx, t, cc2)
-}
-
-func createStubServer(t *testing.T, lis net.Listener, creds credentials.TransportCredentials, modeChangeOpt grpc.ServerOption, bootstrapContents []byte) *stubserver.StubServer {
-	stub := &stubserver.StubServer{
-		Listener: lis,
-		EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-	}
-	server, err := xds.NewGRPCServer(grpc.Creds(creds), modeChangeOpt, xds.BootstrapContentsForTesting(bootstrapContents))
-	if err != nil {
-		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
-	}
-	stub.S = server
-	stubserver.StartTestService(t, stub)
-	return stub
-}
-
-func waitForSuccessfulRPC(ctx context.Context, t *testing.T, cc *grpc.ClientConn) {
-	t.Helper()
-
-	c := testgrpc.NewTestServiceClient(cc)
-	if _, err := c.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
-		t.Fatalf("rpc EmptyCall() failed: %v", err)
-	}
-}
-
-func waitForFailedRPC(ctx context.Context, t *testing.T, cc *grpc.ClientConn) {
-	t.Helper()
-
-	// Attempt one RPC before waiting for the ticker to expire.
-	c := testgrpc.NewTestServiceClient(cc)
-	if _, err := c.EmptyCall(ctx, &testpb.Empty{}); err != nil {
-		return
-	}
-
-	ticker := time.NewTicker(10 * time.Millisecond)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("failure when waiting for RPCs to fail: %v", ctx.Err())
-		case <-ticker.C:
-			if _, err := c.EmptyCall(ctx, &testpb.Empty{}); err != nil {
-				return
-			}
-		}
-	}
-}
diff --git a/test/xds/xds_server_test.go b/test/xds/xds_server_test.go
deleted file mode 100644
index bee9d401..00000000
--- a/test/xds/xds_server_test.go
+++ /dev/null
@@ -1,371 +0,0 @@
-/*
- *
- * Copyright 2023 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package xds_test
-
-import (
-	"context"
-	"io"
-	"net"
-	"strings"
-	"testing"
-	"time"
-
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/credentials/insecure"
-	xdscreds "google.golang.org/grpc/credentials/xds"
-	"google.golang.org/grpc/internal/grpcsync"
-	"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/testutils/xds/e2e/setup"
-	"google.golang.org/grpc/status"
-	"google.golang.org/grpc/xds"
-
-	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
-	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
-	testgrpc "google.golang.org/grpc/interop/grpc_testing"
-	testpb "google.golang.org/grpc/interop/grpc_testing"
-)
-
-var (
-	errAcceptAndClose = status.New(codes.Unavailable, "")
-)
-
-// TestServeLDSRDS tests the case where a server receives LDS resource which
-// specifies RDS. LDS and RDS resources are configured on the management server,
-// which the server should pick up. The server should successfully accept
-// connections and RPCs should work on these accepted connections. It then
-// switches the RDS resource to match incoming RPC's to a route type of type
-// that isn't non forwarding action. This should get picked up by the connection
-// dynamically, and subsequent RPC's on that connection should start failing
-// with status code UNAVAILABLE.
-func (s) TestServeLDSRDS(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	// Setup the management server to respond with a listener resource that
-	// specifies a route name to watch, and a RDS resource corresponding to this
-	// route name.
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-
-	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")
-	routeConfig := e2e.RouteConfigNonForwardingAction("routeName")
-
-	resources := e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener},
-		Routes:    []*v3routepb.RouteConfiguration{routeConfig},
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	serving := grpcsync.NewEvent()
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving.Fire()
-		}
-	})
-	// Configure xDS credentials with an insecure fallback to be used on the
-	// server-side.
-	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
-	if err != nil {
-		t.Fatalf("failed to create server credentials: %v", err)
-	}
-	stub := createStubServer(t, lis, creds, modeChangeOpt, bootstrapContents)
-	defer stub.S.Stop()
-
-	select {
-	case <-ctx.Done():
-		t.Fatal("timeout waiting for the xDS Server to go Serving")
-	case <-serving.Done():
-	}
-
-	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
-	}
-	defer cc.Close()
-
-	waitForSuccessfulRPC(ctx, t, cc) // Eventually, the LDS and dynamic RDS get processed, work, and RPC's should work as usual.
-
-	// Set the route config to be of type route action route, which the rpc will
-	// match to. This should eventually reflect in the Conn's routing
-	// configuration and fail the rpc with a status code UNAVAILABLE.
-	routeConfig = e2e.RouteConfigFilterAction("routeName")
-	resources = e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener}, // Same lis, so will get eaten by the xDS Client.
-		Routes:    []*v3routepb.RouteConfiguration{routeConfig},
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// "NonForwardingAction is expected for all Routes used on server-side; a
-	// route with an inappropriate action causes RPCs matching that route to
-	// fail with UNAVAILABLE." - A36
-	waitForFailedRPCWithStatus(ctx, t, cc, status.New(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding"))
-}
-
-// waitForFailedRPCWithStatus makes unary RPC's until it receives the expected
-// status in a polling manner. Fails if the RPC made does not return the
-// expected status before the context expires.
-func waitForFailedRPCWithStatus(ctx context.Context, t *testing.T, cc *grpc.ClientConn, st *status.Status) {
-	t.Helper()
-
-	c := testgrpc.NewTestServiceClient(cc)
-	ticker := time.NewTicker(10 * time.Millisecond)
-	defer ticker.Stop()
-	var err error
-	for {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("failure when waiting for RPCs to fail with certain status %v: %v. most recent error received from RPC: %v", st, ctx.Err(), err)
-		case <-ticker.C:
-			_, err = c.EmptyCall(ctx, &testpb.Empty{})
-			if status.Code(err) == st.Code() && strings.Contains(err.Error(), st.Message()) {
-				t.Logf("most recent error happy case: %v", err.Error())
-				return
-			}
-		}
-	}
-}
-
-// TestResourceNack tests the case where an LDS points to an RDS which returns
-// an RDS Resource which is NACKed. This should trigger server should move to
-// serving, successfully Accept Connections, and fail at the L7 level with a
-// certain error message.
-func (s) TestRDSNack(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	// Setup the management server to respond with a listener resource that
-	// specifies a route name to watch, and no RDS resource corresponding to
-	// this route name.
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-
-	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")
-	routeConfig := e2e.RouteConfigNoRouteMatch("routeName")
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{listener},
-		Routes:         []*v3routepb.RouteConfiguration{routeConfig},
-		SkipValidation: true,
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-	serving := grpcsync.NewEvent()
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving.Fire()
-		}
-	})
-
-	// Configure xDS credentials with an insecure fallback to be used on the
-	// server-side.
-	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
-	if err != nil {
-		t.Fatalf("failed to create server credentials: %v", err)
-	}
-
-	stub := createStubServer(t, lis, creds, modeChangeOpt, bootstrapContents)
-	defer stub.S.Stop()
-
-	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
-	}
-	defer cc.Close()
-
-	<-serving.Done()
-	waitForFailedRPCWithStatus(ctx, t, cc, status.New(codes.Unavailable, "error from xDS configuration for matched route configuration"))
-}
-
-// TestMultipleUpdatesImmediatelySwitch tests the case where you get an LDS
-// specifying RDS A, B, and C (with A being matched to). The Server should be in
-// not serving until it receives all 3 RDS Configurations, and then transition
-// into serving. RPCs will match to RDS A and work properly. Afterward, it
-// receives an LDS specifying RDS A, B. The Filter Chain pointing to RDS A
-// doesn't get matched, and the Default Filter Chain pointing to RDS B does get
-// matched. RDS B is of the wrong route type for server side, so RPC's are
-// expected to eventually fail with that information. However, any RPC's on the
-// old configuration should be allowed to complete due to the transition being
-// graceful stop.After, it receives an LDS specifying RDS A (which incoming
-// RPC's will match to). This configuration should eventually be represented in
-// the Server's state, and RPCs should proceed successfully.
-func (s) TestMultipleUpdatesImmediatelySwitch(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	// Setup the management server to respond with a listener resource that
-	// specifies three route names to watch.
-	ldsResource := e2e.ListenerResourceThreeRouteResources(host, port, e2e.SecurityLevelNone, "routeName")
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{ldsResource},
-		SkipValidation: true,
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	stub := &stubserver.StubServer{
-		Listener: lis,
-		EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
-			for {
-				_, err := stream.Recv() // hangs here forever if stream doesn't shut down...doesn't receive EOF without any errors
-				if err == io.EOF {
-					return nil
-				}
-			}
-		},
-	}
-
-	if stub.S, err = xds.NewGRPCServer(grpc.Creds(insecure.NewCredentials()), testModeChangeServerOption(t), xds.BootstrapContentsForTesting(bootstrapContents)); err != nil {
-		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
-	}
-	defer stub.S.Stop()
-	stubserver.StartTestService(t, stub)
-
-	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
-	}
-	defer cc.Close()
-
-	waitForFailedRPCWithStatus(ctx, t, cc, errAcceptAndClose)
-
-	routeConfig1 := e2e.RouteConfigNonForwardingAction("routeName")
-	routeConfig2 := e2e.RouteConfigFilterAction("routeName2")
-	routeConfig3 := e2e.RouteConfigFilterAction("routeName3")
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{ldsResource},
-		Routes:         []*v3routepb.RouteConfiguration{routeConfig1, routeConfig2, routeConfig3},
-		SkipValidation: true,
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-	pollForSuccessfulRPC(ctx, t, cc)
-
-	c := testgrpc.NewTestServiceClient(cc)
-	stream, err := c.FullDuplexCall(ctx)
-	if err != nil {
-		t.Fatalf("cc.FullDuplexCall failed: %f", err)
-	}
-	if err = stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
-		t.Fatalf("stream.Send() failed: %v, should continue to work due to graceful stop", err)
-	}
-
-	// Configure with LDS with a filter chain that doesn't get matched to and a
-	// default filter chain that matches to RDS A.
-	ldsResource = e2e.ListenerResourceFallbackToDefault(host, port, e2e.SecurityLevelNone)
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{ldsResource},
-		Routes:         []*v3routepb.RouteConfiguration{routeConfig1, routeConfig2, routeConfig3},
-		SkipValidation: true,
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatalf("error updating management server: %v", err)
-	}
-
-	// xDS is eventually consistent. So simply poll for the new change to be
-	// reflected.
-	// "NonForwardingAction is expected for all Routes used on server-side; a
-	// route with an inappropriate action causes RPCs matching that route to
-	// fail with UNAVAILABLE." - A36
-	waitForFailedRPCWithStatus(ctx, t, cc, status.New(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding"))
-
-	// Stream should be allowed to continue on the old working configuration -
-	// as it on a connection that is gracefully closed (old FCM/LDS
-	// Configuration which is allowed to continue).
-	if err = stream.CloseSend(); err != nil {
-		t.Fatalf("stream.CloseSend() failed: %v, should continue to work due to graceful stop", err)
-	}
-	if _, err = stream.Recv(); err != io.EOF {
-		t.Fatalf("unexpected error: %v, expected an EOF error", err)
-	}
-
-	ldsResource = e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, "routeName")
-	resources = e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{ldsResource},
-		Routes:         []*v3routepb.RouteConfiguration{routeConfig1, routeConfig2, routeConfig3},
-		SkipValidation: true,
-	}
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	pollForSuccessfulRPC(ctx, t, cc)
-}
-
-func pollForSuccessfulRPC(ctx context.Context, t *testing.T, cc *grpc.ClientConn) {
-	t.Helper()
-	c := testgrpc.NewTestServiceClient(cc)
-	ticker := time.NewTicker(10 * time.Millisecond)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("timeout waiting for RPCs to succeed")
-		case <-ticker.C:
-			if _, err := c.EmptyCall(ctx, &testpb.Empty{}); err == nil {
-				return
-			}
-		}
-	}
-}
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
index cb22582f..f76e9991 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -19,7 +19,6 @@ package cdsbalancer
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"net"
 	"strings"
@@ -302,6 +301,22 @@ func compareLoadBalancingConfig(ctx context.Context, lbCfgCh chan serviceconfig.
 	return nil
 }
 
+func verifyRPCError(gotErr error, wantCode codes.Code, wantErr, wantNodeID string) error {
+	if gotErr == nil {
+		return fmt.Errorf("RPC succeeded when expecting an error with code %v, message %q and nodeID %q", wantCode, wantErr, wantNodeID)
+	}
+	if gotCode := status.Code(gotErr); gotCode != wantCode {
+		return fmt.Errorf("RPC failed with code: %v, want code %v", gotCode, wantCode)
+	}
+	if !strings.Contains(gotErr.Error(), wantErr) {
+		return fmt.Errorf("RPC failed with error: %v, want %q", gotErr, wantErr)
+	}
+	if !strings.Contains(gotErr.Error(), wantNodeID) {
+		return fmt.Errorf("RPC failed with error: %v, want nodeID %q", gotErr, wantNodeID)
+	}
+	return nil
+}
+
 // Tests the functionality that handles LB policy configuration. Verifies that
 // the appropriate xDS resource is requested corresponding to the provided LB
 // policy configuration. Also verifies that when the LB policy receives the same
@@ -712,15 +727,13 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
 
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 
-	// Ensure that the NACK error is propagated to the RPC caller.
+	// Ensure that the NACK error and the xDS node ID are propagated to the RPC
+	// caller.
 	const wantClusterNACKErr = "unsupported config_source_specifier"
 	client := testgrpc.NewTestServiceClient(cc)
 	_, err := client.EmptyCall(ctx, &testpb.Empty{})
-	if code := status.Code(err); code != codes.Unavailable {
-		t.Fatalf("EmptyCall() failed with code: %v, want %v", code, codes.Unavailable)
-	}
-	if err != nil && !strings.Contains(err.Error(), wantClusterNACKErr) {
-		t.Fatalf("EmptyCall() failed with err: %v, want err containing: %v", err, wantClusterNACKErr)
+	if err := verifyRPCError(err, codes.Unavailable, wantClusterNACKErr, nodeID); err != nil {
+		t.Fatal(err)
 	}
 
 	// Start a test service backend.
@@ -814,9 +827,11 @@ func (s) TestResolverError(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Push a resolver error that is not a resource-not-found error.
-	resolverErr := errors.New("resolver-error-not-a-resource-not-found-error")
-	r.ReportError(resolverErr)
+	// Push a resolver error that is not a resource-not-found error. Here, we
+	// assume that errors from the xDS client or from the xDS resolver contain
+	// the xDS node ID.
+	resolverErr := fmt.Errorf("[xds node id: %s]: resolver-error-not-a-resource-not-found-error", nodeID)
+	r.CC().ReportError(resolverErr)
 
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 
@@ -829,11 +844,8 @@ func (s) TestResolverError(t *testing.T) {
 	// Ensure that the resolver error is propagated to the RPC caller.
 	client := testgrpc.NewTestServiceClient(cc)
 	_, err = client.EmptyCall(ctx, &testpb.Empty{})
-	if code := status.Code(err); code != codes.Unavailable {
-		t.Fatalf("EmptyCall() failed with code: %v, want %v", code, codes.Unavailable)
-	}
-	if err != nil && !strings.Contains(err.Error(), resolverErr.Error()) {
-		t.Fatalf("EmptyCall() failed with err: %v, want %v", err, resolverErr)
+	if err := verifyRPCError(err, codes.Unavailable, resolverErr.Error(), nodeID); err != nil {
+		t.Fatal(err)
 	}
 
 	// Also verify that the watch for the cluster resource is not cancelled.
@@ -866,7 +878,7 @@ func (s) TestResolverError(t *testing.T) {
 	}
 
 	// Again push a resolver error that is not a resource-not-found error.
-	r.ReportError(resolverErr)
+	r.CC().ReportError(resolverErr)
 
 	// And again verify that the watch for the cluster resource is not
 	// cancelled.
@@ -894,9 +906,16 @@ func (s) TestResolverError(t *testing.T) {
 		t.Fatal("Timeout when waiting for resolver error to be pushed to the child policy")
 	}
 
-	// Push a resource-not-found-error this time around.
-	resolverErr = xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")
-	r.ReportError(resolverErr)
+	// Push a resource-not-found-error this time around. Our xDS resolver does
+	// not send this error though. When an LDS or RDS resource is missing, the
+	// xDS resolver instead sends an erroring config selector which returns an
+	// error at RPC time with the xDS node ID, for new RPCs. Once ongoing RPCs
+	// complete, the xDS resolver will send an empty service config with no
+	// addresses, which will result in pick_first being configured on the
+	// channel. And pick_first will put the channel in TRANSIENT_FAILURE since
+	// it would have received an update with no addresses.
+	resolverErr = fmt.Errorf("[xds node id: %s]: %w", nodeID, xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error"))
+	r.CC().ReportError(resolverErr)
 
 	// Wait for the CDS resource to be not requested anymore, or the connection
 	// to the management server to be closed (which happens as part of the last
@@ -917,11 +936,10 @@ func (s) TestResolverError(t *testing.T) {
 
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 
-	// Ensure RPC fails with Unavailable. The actual error message depends on
-	// the picker returned from the priority LB policy, and therefore not
-	// checking for it here.
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Unavailable {
-		t.Fatalf("EmptyCall() failed with code: %v, want %v", status.Code(err), codes.Unavailable)
+	// Ensure that the resolver error is propagated to the RPC caller.
+	_, err = client.EmptyCall(ctx, &testpb.Empty{})
+	if err := verifyRPCError(err, codes.Unavailable, resolverErr.Error(), nodeID); err != nil {
+		t.Fatal(err)
 	}
 }
 
@@ -983,14 +1001,12 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 
-	// Ensure RPC fails with Unavailable.
+	// Ensure RPC fails with Unavailable status code and the error message is
+	// meaningful and contains the xDS node ID.
 	wantErr := fmt.Sprintf("cluster %q not found", clusterName)
 	_, err := client.EmptyCall(ctx, &testpb.Empty{})
-	if status.Code(err) != codes.Unavailable {
-		t.Fatalf("EmptyCall() failed with code: %v, want %v", status.Code(err), codes.Unavailable)
-	}
-	if !strings.Contains(err.Error(), wantErr) {
-		t.Fatalf("EmptyCall() failed with error: %v, want %v", err, wantErr)
+	if err := verifyRPCError(err, codes.Unavailable, wantErr, nodeID); err != nil {
+		t.Fatal(err)
 	}
 
 	// Re-add the cluster resource to the management server.
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go b/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
index 96812c68..36d22abf 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
@@ -768,7 +768,7 @@ func (s) TestAggregateCluster_BadEDS_GoodToBadDNS(t *testing.T) {
 
 	// Push an error from the DNS resolver as well.
 	dnsErr := fmt.Errorf("DNS error")
-	dnsR.ReportError(dnsErr)
+	dnsR.CC().ReportError(dnsErr)
 
 	// Ensure that RPCs continue to succeed for the next second.
 	for end := time.Now().Add(time.Second); time.Now().Before(end); <-time.After(defaultTestShortTimeout) {
@@ -902,7 +902,7 @@ func (s) TestAggregateCluster_BadDNS_GoodEDS(t *testing.T) {
 
 	// Produce a bad resolver update from the DNS resolver.
 	dnsErr := fmt.Errorf("DNS error")
-	dnsR.ReportError(dnsErr)
+	dnsR.CC().ReportError(dnsErr)
 
 	// RPCs should work, higher level DNS cluster errors so should fallback to
 	// EDS cluster.
@@ -974,7 +974,7 @@ func (s) TestAggregateCluster_BadEDS_BadDNS(t *testing.T) {
 	}
 
 	// Produce a bad resolver update from the DNS resolver.
-	dnsR.ReportError(fmt.Errorf("DNS error"))
+	dnsR.CC().ReportError(fmt.Errorf("DNS error"))
 
 	// Ensure that the error from the DNS Resolver leads to an empty address
 	// update for both priorities.
diff --git a/xds/internal/balancer/priority/balancer_test.go b/xds/internal/balancer/priority/balancer_test.go
index 7183cefc..dee06006 100644
--- a/xds/internal/balancer/priority/balancer_test.go
+++ b/xds/internal/balancer/priority/balancer_test.go
@@ -493,7 +493,11 @@ func (s) TestPriority_HigherDownWhileAddingLower(t *testing.T) {
 	sc0 := <-cc.NewSubConnCh
 
 	t.Log("Turn down 0, 1 is used.")
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
+	testErr := errors.New("test error")
+	sc0.UpdateState(balancer.SubConnState{
+		ConnectivityState: connectivity.TransientFailure,
+		ConnectionError:   testErr,
+	})
 
 	// Before 1 gets READY, picker should return NoSubConnAvailable, so RPCs
 	// will retry.
@@ -508,10 +512,13 @@ func (s) TestPriority_HigherDownWhileAddingLower(t *testing.T) {
 	sc1 := <-cc.NewSubConnCh
 
 	t.Log("Turn down 1, pick should error.")
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
+	sc1.UpdateState(balancer.SubConnState{
+		ConnectivityState: connectivity.TransientFailure,
+		ConnectionError:   testErr,
+	})
 
 	// Test pick failure.
-	if err := cc.WaitForErrPicker(ctx); err != nil {
+	if err := cc.WaitForPickerWithErr(ctx, testErr); err != nil {
 		t.Fatal(err.Error())
 	}
 	<-cc.NewStateCh // Drain to match picker
diff --git a/xds/internal/clients/config_test.go b/xds/internal/clients/config_test.go
deleted file mode 100644
index 48417db0..00000000
--- a/xds/internal/clients/config_test.go
+++ /dev/null
@@ -1,353 +0,0 @@
-/*
- *
- * Copyright 2024 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package clients
-
-import (
-	"testing"
-
-	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	"github.com/google/go-cmp/cmp"
-	"google.golang.org/grpc/internal/grpctest"
-	"google.golang.org/protobuf/testing/protocmp"
-	"google.golang.org/protobuf/types/known/structpb"
-)
-
-type s struct {
-	grpctest.Tester
-}
-
-func Test(t *testing.T) {
-	grpctest.RunSubTests(t, s{})
-}
-
-type testServerConfigExtension struct{ x int }
-
-func (ts testServerConfigExtension) Equal(other any) bool {
-	ots, ok := other.(testServerConfigExtension)
-	if !ok {
-		return false
-	}
-	return ts.x == ots.x
-}
-
-func newStructProtoFromMap(t *testing.T, input map[string]any) *structpb.Struct {
-	t.Helper()
-
-	ret, err := structpb.NewStruct(input)
-	if err != nil {
-		t.Fatalf("Failed to create new struct proto from map %v: %v", input, err)
-	}
-	return ret
-}
-
-func (s) TestServerConfig_Equal(t *testing.T) {
-	tests := []struct {
-		name   string
-		s1     *ServerConfig
-		s2     *ServerConfig
-		wantEq bool
-	}{
-		{
-			name:   "both_nil",
-			s1:     nil,
-			s2:     nil,
-			wantEq: true,
-		},
-		{
-			name:   "one_nil",
-			s1:     nil,
-			s2:     &ServerConfig{},
-			wantEq: false,
-		},
-		{
-			name:   "other_nil",
-			s1:     &ServerConfig{},
-			s2:     nil,
-			wantEq: false,
-		},
-		{
-			name:   "both_empty_and_equal",
-			s1:     &ServerConfig{},
-			s2:     &ServerConfig{},
-			wantEq: true,
-		},
-		{
-			name:   "different_ServerURI",
-			s1:     &ServerConfig{ServerURI: "foo"},
-			s2:     &ServerConfig{ServerURI: "bar"},
-			wantEq: false,
-		},
-		{
-			name:   "different_IgnoreResourceDeletion",
-			s1:     &ServerConfig{IgnoreResourceDeletion: true},
-			s2:     &ServerConfig{},
-			wantEq: false,
-		},
-		{
-			name: "different_Extensions_with_no_Equal_method",
-			s1: &ServerConfig{
-				Extensions: 1,
-			},
-			s2: &ServerConfig{
-				Extensions: 2,
-			},
-			wantEq: false, // By default, if there's no Equal method, they are unequal
-		},
-		{
-			name: "same_Extensions_with_no_Equal_method",
-			s1: &ServerConfig{
-				Extensions: 1,
-			},
-			s2: &ServerConfig{
-				Extensions: 1,
-			},
-			wantEq: false, // By default, if there's no Equal method, they are unequal
-		},
-		{
-			name: "different_Extensions_with_Equal_method",
-			s1: &ServerConfig{
-				Extensions: testServerConfigExtension{1},
-			},
-			s2: &ServerConfig{
-				Extensions: testServerConfigExtension{2},
-			},
-			wantEq: false,
-		},
-		{
-			name: "same_Extensions_same_with_Equal_method",
-			s1: &ServerConfig{
-				Extensions: testServerConfigExtension{1},
-			},
-			s2: &ServerConfig{
-				Extensions: testServerConfigExtension{1},
-			},
-			wantEq: true,
-		},
-		{
-			name: "first_config_Extensions_is_nil",
-			s1: &ServerConfig{
-				Extensions: testServerConfigExtension{1},
-			},
-			s2: &ServerConfig{
-				Extensions: nil,
-			},
-			wantEq: false,
-		},
-		{
-			name: "other_config_Extensions_is_nil",
-			s1: &ServerConfig{
-				Extensions: nil,
-			},
-			s2: &ServerConfig{
-				Extensions: testServerConfigExtension{2},
-			},
-			wantEq: false,
-		},
-		{
-			name: "all_fields_same",
-			s1: &ServerConfig{
-				ServerURI:              "foo",
-				IgnoreResourceDeletion: true,
-				Extensions:             testServerConfigExtension{1},
-			},
-			s2: &ServerConfig{
-				ServerURI:              "foo",
-				IgnoreResourceDeletion: true,
-				Extensions:             testServerConfigExtension{1},
-			},
-			wantEq: true,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			if gotEq := test.s1.equal(test.s2); gotEq != test.wantEq {
-				t.Errorf("Equal() = %v, want %v", gotEq, test.wantEq)
-			}
-		})
-	}
-}
-
-func (s) TestLocality_IsEmpty(t *testing.T) {
-	tests := []struct {
-		name     string
-		locality Locality
-		want     bool
-	}{
-		{
-			name:     "empty_locality",
-			locality: Locality{},
-			want:     true,
-		},
-		{
-			name:     "non_empty_region",
-			locality: Locality{Region: "region"},
-			want:     false,
-		},
-		{
-			name:     "non_empty_zone",
-			locality: Locality{Zone: "zone"},
-			want:     false,
-		},
-		{
-			name:     "non_empty_subzone",
-			locality: Locality{SubZone: "subzone"},
-			want:     false,
-		},
-		{
-			name:     "non_empty_all_fields",
-			locality: Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
-			want:     false,
-		},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			if got := test.locality.isEmpty(); got != test.want {
-				t.Errorf("IsEmpty() = %v, want %v", got, test.want)
-			}
-		})
-	}
-}
-
-func (s) TestLocality_Equal(t *testing.T) {
-	tests := []struct {
-		name   string
-		l1     Locality
-		l2     Locality
-		wantEq bool
-	}{
-		{
-			name:   "both_equal",
-			l1:     Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
-			l2:     Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
-			wantEq: true,
-		},
-		{
-			name:   "different_regions",
-			l1:     Locality{Region: "region1", Zone: "zone", SubZone: "subzone"},
-			l2:     Locality{Region: "region2", Zone: "zone", SubZone: "subzone"},
-			wantEq: false,
-		},
-
-		{
-			name:   "different_zones",
-			l1:     Locality{Region: "region", Zone: "zone1", SubZone: "subzone"},
-			l2:     Locality{Region: "region", Zone: "zone2", SubZone: "subzone"},
-			wantEq: false,
-		},
-		{
-			name:   "different_subzones",
-			l1:     Locality{Region: "region", Zone: "zone", SubZone: "subzone1"},
-			l2:     Locality{Region: "region", Zone: "zone", SubZone: "subzone2"},
-			wantEq: false,
-		},
-		{
-			name:   "one_empty",
-			l1:     Locality{},
-			l2:     Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
-			wantEq: false,
-		},
-		{
-			name:   "both_empty",
-			l1:     Locality{},
-			l2:     Locality{},
-			wantEq: true,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			if gotEq := test.l1.equal(test.l2); gotEq != test.wantEq {
-				t.Errorf("Equal() = %v, want %v", gotEq, test.wantEq)
-			}
-		})
-	}
-}
-
-func (s) TestNode_ToProto(t *testing.T) {
-	tests := []struct {
-		desc      string
-		inputNode Node
-		wantProto *v3corepb.Node
-	}{
-		{
-			desc: "all_fields_set",
-			inputNode: Node{
-				ID:      "id",
-				Cluster: "cluster",
-				Locality: Locality{
-					Region:  "region",
-					Zone:    "zone",
-					SubZone: "sub_zone",
-				},
-				Metadata:         newStructProtoFromMap(t, map[string]any{"k1": "v1", "k2": 101, "k3": 280.0}),
-				UserAgentName:    "user agent",
-				UserAgentVersion: "version",
-				clientFeatures:   []string{"feature1", "feature2"},
-			},
-			wantProto: &v3corepb.Node{
-				Id:      "id",
-				Cluster: "cluster",
-				Locality: &v3corepb.Locality{
-					Region:  "region",
-					Zone:    "zone",
-					SubZone: "sub_zone",
-				},
-				Metadata:             newStructProtoFromMap(t, map[string]any{"k1": "v1", "k2": 101, "k3": 280.0}),
-				UserAgentName:        "user agent",
-				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: "version"},
-				ClientFeatures:       []string{"feature1", "feature2"},
-			},
-		},
-		{
-			desc: "some_fields_unset",
-			inputNode: Node{
-				ID: "id",
-			},
-			wantProto: &v3corepb.Node{
-				Id:                   "id",
-				UserAgentName:        "",
-				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: ""},
-				ClientFeatures:       nil,
-			},
-		},
-		{
-			desc: "empty_locality",
-			inputNode: Node{
-				ID:       "id",
-				Locality: Locality{},
-			},
-			wantProto: &v3corepb.Node{
-				Id:                   "id",
-				UserAgentName:        "",
-				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: ""},
-				ClientFeatures:       nil,
-			},
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.desc, func(t *testing.T) {
-			gotProto := test.inputNode.toProto()
-			if diff := cmp.Diff(test.wantProto, gotProto, protocmp.Transform()); diff != "" {
-				t.Fatalf("Unexpected diff in node proto: (-want, +got):\n%s", diff)
-			}
-		})
-	}
-}
diff --git a/xds/internal/clients/grpctransport/grpc_transport_test.go b/xds/internal/clients/grpctransport/grpc_transport_test.go
index 0ab48707..cb308d8e 100644
--- a/xds/internal/clients/grpctransport/grpc_transport_test.go
+++ b/xds/internal/clients/grpctransport/grpc_transport_test.go
@@ -105,6 +105,8 @@ func (s *testServer) StreamAggregatedResources(stream v3discoverygrpc.Aggregated
 			return err // Handle other errors
 		}
 
+		// Push received request for client to verify the correct request was
+		// received.
 		select {
 		case s.requestChan <- req:
 		case <-ctx.Done():
@@ -130,9 +132,9 @@ func (tc *testCredentials) TransportCredentials() credentials.TransportCredentia
 // TestBuild_Success verifies that the Builder successfully creates a new
 // Transport with a non-nil grpc.ClientConn.
 func (s) TestBuild_Success(t *testing.T) {
-	serverCfg := clients.ServerConfig{
+	serverCfg := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerConfigExtension{Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+		Extensions: ServerIdentifierExtension{Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
 
 	b := &Builder{}
@@ -151,41 +153,41 @@ func (s) TestBuild_Success(t *testing.T) {
 }
 
 // TestBuild_Failure verifies that the Builder returns error when incorrect
-// ServerConfig is provided.
+// ServerIdentifier is provided.
 //
 // It covers the following scenarios:
 // - ServerURI is empty.
 // - Extensions is nil.
-// - Extensions is not ServerConfigExtension.
+// - Extensions is not ServerIdentifierExtension.
 // - Credentials are nil.
 func (s) TestBuild_Failure(t *testing.T) {
 	tests := []struct {
 		name      string
-		serverCfg clients.ServerConfig
+		serverCfg clients.ServerIdentifier
 	}{
 		{
 			name: "ServerURI is empty",
-			serverCfg: clients.ServerConfig{
+			serverCfg: clients.ServerIdentifier{
 				ServerURI:  "",
-				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+				Extensions: ServerIdentifierExtension{Credentials: insecure.NewBundle()},
 			},
 		},
 		{
 			name:      "Extensions is nil",
-			serverCfg: clients.ServerConfig{ServerURI: "server-address"},
+			serverCfg: clients.ServerIdentifier{ServerURI: "server-address"},
 		},
 		{
-			name: "Extensions is not a ServerConfigExtension",
-			serverCfg: clients.ServerConfig{
+			name: "Extensions is not a ServerIdentifierExtension",
+			serverCfg: clients.ServerIdentifier{
 				ServerURI:  "server-address",
 				Extensions: 1,
 			},
 		},
 		{
-			name: "ServerConfigExtension Credentials is nil",
-			serverCfg: clients.ServerConfig{
+			name: "ServerIdentifierExtension Credentials is nil",
+			serverCfg: clients.ServerIdentifier{
 				ServerURI:  "server-address",
-				Extensions: ServerConfigExtension{},
+				Extensions: ServerIdentifierExtension{},
 			},
 		},
 	}
@@ -208,9 +210,9 @@ func (s) TestBuild_Failure(t *testing.T) {
 func (s) TestNewStream_Success(t *testing.T) {
 	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
 
-	serverCfg := clients.ServerConfig{
+	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+		Extensions: ServerIdentifierExtension{Credentials: insecure.NewBundle()},
 	}
 	builder := Builder{}
 	transport, err := builder.Build(serverCfg)
@@ -229,9 +231,9 @@ func (s) TestNewStream_Success(t *testing.T) {
 // TestNewStream_Error verifies that NewStream() returns an error
 // when attempting to create a stream with an invalid server URI.
 func (s) TestNewStream_Error(t *testing.T) {
-	serverCfg := clients.ServerConfig{
+	serverCfg := clients.ServerIdentifier{
 		ServerURI:  "invalid-server-uri",
-		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+		Extensions: ServerIdentifierExtension{Credentials: insecure.NewBundle()},
 	}
 	builder := Builder{}
 	transport, err := builder.Build(serverCfg)
@@ -262,9 +264,9 @@ func (s) TestStream_SendAndRecv(t *testing.T) {
 	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
 
 	// Build a grpc-based transport to the above server.
-	serverCfg := clients.ServerConfig{
+	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
+		Extensions: ServerIdentifierExtension{Credentials: insecure.NewBundle()},
 	}
 	builder := Builder{}
 	transport, err := builder.Build(serverCfg)
diff --git a/xds/internal/clients/internal/internal_test.go b/xds/internal/clients/internal/internal_test.go
new file mode 100644
index 00000000..8815b925
--- /dev/null
+++ b/xds/internal/clients/internal/internal_test.go
@@ -0,0 +1,215 @@
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package internal
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+
+	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/xds/internal/clients"
+
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+)
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+func newStructProtoFromMap(t *testing.T, input map[string]any) *structpb.Struct {
+	t.Helper()
+
+	ret, err := structpb.NewStruct(input)
+	if err != nil {
+		t.Fatalf("Failed to create new struct proto from map %v: %v", input, err)
+	}
+	return ret
+}
+
+func (s) TestIsLocalityEmpty(t *testing.T) {
+	tests := []struct {
+		name     string
+		locality clients.Locality
+		want     bool
+	}{
+		{
+			name:     "empty_locality",
+			locality: clients.Locality{},
+			want:     true,
+		},
+		{
+			name:     "non_empty_region",
+			locality: clients.Locality{Region: "region"},
+			want:     false,
+		},
+		{
+			name:     "non_empty_zone",
+			locality: clients.Locality{Zone: "zone"},
+			want:     false,
+		},
+		{
+			name:     "non_empty_subzone",
+			locality: clients.Locality{SubZone: "subzone"},
+			want:     false,
+		},
+		{
+			name:     "non_empty_all_fields",
+			locality: clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
+			want:     false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := isLocalityEmpty(test.locality); got != test.want {
+				t.Errorf("IsEmpty() = %v, want %v", got, test.want)
+			}
+		})
+	}
+}
+
+func (s) TestIsLocalityEqual(t *testing.T) {
+	tests := []struct {
+		name   string
+		l1     clients.Locality
+		l2     clients.Locality
+		wantEq bool
+	}{
+		{
+			name:   "both_equal",
+			l1:     clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
+			l2:     clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
+			wantEq: true,
+		},
+		{
+			name:   "different_regions",
+			l1:     clients.Locality{Region: "region1", Zone: "zone", SubZone: "subzone"},
+			l2:     clients.Locality{Region: "region2", Zone: "zone", SubZone: "subzone"},
+			wantEq: false,
+		},
+
+		{
+			name:   "different_zones",
+			l1:     clients.Locality{Region: "region", Zone: "zone1", SubZone: "subzone"},
+			l2:     clients.Locality{Region: "region", Zone: "zone2", SubZone: "subzone"},
+			wantEq: false,
+		},
+		{
+			name:   "different_subzones",
+			l1:     clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone1"},
+			l2:     clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone2"},
+			wantEq: false,
+		},
+		{
+			name:   "one_empty",
+			l1:     clients.Locality{},
+			l2:     clients.Locality{Region: "region", Zone: "zone", SubZone: "subzone"},
+			wantEq: false,
+		},
+		{
+			name:   "both_empty",
+			l1:     clients.Locality{},
+			l2:     clients.Locality{},
+			wantEq: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if gotEq := isLocalityEqual(test.l1, test.l2); gotEq != test.wantEq {
+				t.Errorf("Equal() = %v, want %v", gotEq, test.wantEq)
+			}
+		})
+	}
+}
+
+func (s) TestNodeProto(t *testing.T) {
+	tests := []struct {
+		desc      string
+		inputNode clients.Node
+		wantProto *v3corepb.Node
+	}{
+		{
+			desc: "all_fields_set",
+			inputNode: clients.Node{
+				ID:      "id",
+				Cluster: "cluster",
+				Locality: clients.Locality{
+					Region:  "region",
+					Zone:    "zone",
+					SubZone: "sub_zone",
+				},
+				Metadata:         newStructProtoFromMap(t, map[string]any{"k1": "v1", "k2": 101, "k3": 280.0}),
+				UserAgentName:    "user agent",
+				UserAgentVersion: "version",
+			},
+			wantProto: &v3corepb.Node{
+				Id:      "id",
+				Cluster: "cluster",
+				Locality: &v3corepb.Locality{
+					Region:  "region",
+					Zone:    "zone",
+					SubZone: "sub_zone",
+				},
+				Metadata:             newStructProtoFromMap(t, map[string]any{"k1": "v1", "k2": 101, "k3": 280.0}),
+				UserAgentName:        "user agent",
+				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: "version"},
+			},
+		},
+		{
+			desc: "some_fields_unset",
+			inputNode: clients.Node{
+				ID: "id",
+			},
+			wantProto: &v3corepb.Node{
+				Id:                   "id",
+				UserAgentName:        "",
+				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: ""},
+			},
+		},
+		{
+			desc: "empty_locality",
+			inputNode: clients.Node{
+				ID:       "id",
+				Locality: clients.Locality{},
+			},
+			wantProto: &v3corepb.Node{
+				Id:                   "id",
+				UserAgentName:        "",
+				UserAgentVersionType: &v3corepb.Node_UserAgentVersion{UserAgentVersion: ""},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			gotProto := NodeProto(test.inputNode)
+			if diff := cmp.Diff(test.wantProto, gotProto, protocmp.Transform()); diff != "" {
+				t.Fatalf("Unexpected diff in node proto: (-want, +got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/xds/internal/resolver/cluster_specifier_plugin_test.go b/xds/internal/resolver/cluster_specifier_plugin_test.go
index 4f807e39..7f1144b8 100644
--- a/xds/internal/resolver/cluster_specifier_plugin_test.go
+++ b/xds/internal/resolver/cluster_specifier_plugin_test.go
@@ -115,7 +115,7 @@ func (s) TestResolverClusterSpecifierPlugin(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
@@ -205,7 +205,7 @@ func (s) TestXDSResolverDelayedOnCommittedCSP(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
diff --git a/xds/internal/resolver/helpers_test.go b/xds/internal/resolver/helpers_test.go
index 708b51bb..cdb9511b 100644
--- a/xds/internal/resolver/helpers_test.go
+++ b/xds/internal/resolver/helpers_test.go
@@ -28,6 +28,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpctest"
 	iresolver "google.golang.org/grpc/internal/resolver"
@@ -35,6 +36,7 @@ import (
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
+	"google.golang.org/grpc/status"
 	xdsresolver "google.golang.org/grpc/xds/internal/resolver"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 
@@ -180,19 +182,40 @@ func verifyNoUpdateFromResolver(ctx context.Context, t *testing.T, stateCh chan
 	}
 }
 
-// verifyErrorFromResolver waits for the resolver to push an error and verifies
-// that it matches the expected error.
-func verifyErrorFromResolver(ctx context.Context, t *testing.T, errCh chan error, wantErr string) {
-	t.Helper()
-
+// waitForErrorFromResolver waits for the resolver to push an error and verifies
+// that it matches the expected error and contains the expected node ID.
+func waitForErrorFromResolver(ctx context.Context, errCh chan error, wantErr, wantNodeID string) error {
 	select {
 	case <-ctx.Done():
-		t.Fatal("Timeout when waiting for error to be propagated to the ClientConn")
+		return fmt.Errorf("timeout when waiting for error to be propagated to the ClientConn")
 	case gotErr := <-errCh:
-		if gotErr == nil || !strings.Contains(gotErr.Error(), wantErr) {
-			t.Fatalf("Received error from resolver %q, want %q", gotErr, wantErr)
+		if gotErr == nil {
+			return fmt.Errorf("got nil error from resolver, want %q", wantErr)
+		}
+		if !strings.Contains(gotErr.Error(), wantErr) {
+			return fmt.Errorf("got error from resolver %q, want %q", gotErr, wantErr)
 		}
+		if !strings.Contains(gotErr.Error(), wantNodeID) {
+			return fmt.Errorf("got error from resolver %q, want nodeID %q", gotErr, wantNodeID)
+		}
+	}
+	return nil
+}
+
+func verifyResolverError(gotErr error, wantCode codes.Code, wantErr, wantNodeID string) error {
+	if gotErr == nil {
+		return fmt.Errorf("got nil error from resolver, want error with code %v", wantCode)
+	}
+	if !strings.Contains(gotErr.Error(), wantErr) {
+		return fmt.Errorf("got error from resolver %q, want %q", gotErr, wantErr)
+	}
+	if gotCode := status.Code(gotErr); gotCode != wantCode {
+		return fmt.Errorf("got error from resolver with code %v, want %v", gotCode, wantCode)
+	}
+	if !strings.Contains(gotErr.Error(), wantNodeID) {
+		return fmt.Errorf("got error from resolver %q, want nodeID %q", gotErr, wantNodeID)
 	}
+	return nil
 }
 
 // Spins up an xDS management server and sets up an xDS bootstrap configuration
@@ -204,7 +227,7 @@ func verifyErrorFromResolver(ctx context.Context, t *testing.T, errCh chan error
 //   - A channel to read requested RouteConfiguration resource names
 //   - Contents of the bootstrap configuration pointing to xDS management
 //     server
-func setupManagementServerForTest(ctx context.Context, t *testing.T, nodeID string) (*e2e.ManagementServer, chan []string, chan []string, []byte) {
+func setupManagementServerForTest(t *testing.T, nodeID string) (*e2e.ManagementServer, chan []string, chan []string, []byte) {
 	t.Helper()
 
 	listenerResourceNamesCh := make(chan []string, 1)
diff --git a/xds/internal/resolver/watch_service_test.go b/xds/internal/resolver/watch_service_test.go
index 4f2a21e5..7df159c6 100644
--- a/xds/internal/resolver/watch_service_test.go
+++ b/xds/internal/resolver/watch_service_test.go
@@ -46,7 +46,7 @@ func (s) TestServiceWatch_ListenerPointsToNewRouteConfiguration(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
@@ -103,7 +103,7 @@ func (s) TestServiceWatch_ListenerPointsToInlineRouteConfiguration(t *testing.T)
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
diff --git a/xds/internal/resolver/xds_resolver_test.go b/xds/internal/resolver/xds_resolver_test.go
index 8916c4e8..49ddf814 100644
--- a/xds/internal/resolver/xds_resolver_test.go
+++ b/xds/internal/resolver/xds_resolver_test.go
@@ -41,7 +41,6 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/xds/internal/balancer/clustermanager"
 	"google.golang.org/grpc/xds/internal/balancer/ringhash"
 	"google.golang.org/grpc/xds/internal/httpfilter"
@@ -165,7 +164,7 @@ func (s) TestResolverResourceName(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 			defer cancel()
 			nodeID := uuid.New().String()
-			mgmtServer, lisCh, _, _ := setupManagementServerForTest(ctx, t, nodeID)
+			mgmtServer, lisCh, _, _ := setupManagementServerForTest(t, nodeID)
 
 			// Create a bootstrap configuration with test options.
 			opts := bootstrap.ConfigOptionsForTesting{
@@ -300,7 +299,7 @@ func (s) TestResolverBadServiceUpdate(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure a listener resource that is expected to be NACKed because it
 	// does not contain the `RouteSpecifier` field in the HTTPConnectionManager.
@@ -323,7 +322,9 @@ func (s) TestResolverBadServiceUpdate(t *testing.T) {
 	// Build the resolver and expect an error update from it.
 	stateCh, errCh, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 	wantErr := "no RouteSpecifier"
-	verifyErrorFromResolver(ctx, t, errCh, wantErr)
+	if err := waitForErrorFromResolver(ctx, errCh, wantErr, nodeID); err != nil {
+		t.Fatal(err)
+	}
 
 	// Configure good listener and route configuration resources on the
 	// management server.
@@ -337,7 +338,9 @@ func (s) TestResolverBadServiceUpdate(t *testing.T) {
 	// Configure another bad resource on the management server and expect an
 	// error update from the resolver.
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, []*v3listenerpb.Listener{lis}, nil)
-	verifyErrorFromResolver(ctx, t, errCh, wantErr)
+	if err := waitForErrorFromResolver(ctx, errCh, wantErr, nodeID); err != nil {
+		t.Fatal(err)
+	}
 }
 
 // TestResolverGoodServiceUpdate tests the case where the resource returned by
@@ -404,7 +407,7 @@ func (s) TestResolverGoodServiceUpdate(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 			defer cancel()
 			nodeID := uuid.New().String()
-			mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+			mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 			// Configure the management server with a good listener resource and a
 			// route configuration resource, as specified by the test case.
@@ -445,7 +448,7 @@ func (s) TestResolverRequestHash(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure the management server with a good listener resource and a
 	// route configuration resource that specifies a hash policy.
@@ -508,7 +511,7 @@ func (s) TestResolverRemovedWithRPCs(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
@@ -537,8 +540,8 @@ func (s) TestResolverRemovedWithRPCs(t *testing.T) {
 	// return an erroring config selector which will fail new RPCs.
 	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
 	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
-	if err == nil || status.Code(err) != codes.Unavailable {
-		t.Fatalf("cs.SelectConfig() returned: %v, want: %v", err, codes.Unavailable)
+	if err := verifyResolverError(err, codes.Unavailable, "no valid clusters", nodeID); err != nil {
+		t.Fatal(err)
 	}
 
 	// "Finish the RPC"; this could cause a panic if the resolver doesn't
@@ -610,14 +613,14 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
 	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
 
-	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
+	stateCh, errCh, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Read the update pushed by the resolver to the ClientConn.
 	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
@@ -643,9 +646,9 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
 
 	// "Make another RPC" by invoking the config selector.
-	res, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
-	if err == nil || status.Code(err) != codes.Unavailable {
-		t.Fatalf("cs.SelectConfig() got %v, %v, expected UNAVAILABLE error", res, err)
+	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
+	if err := verifyResolverError(err, codes.Unavailable, "no valid clusters", nodeID); err != nil {
+		t.Fatal(err)
 	}
 
 	// In the meantime, an empty ServiceConfig update should have been sent.
@@ -662,6 +665,16 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 			t.Fatalf("Got service config:\n%s \nWant service config:\n%s", cmp.Diff(nil, state.ServiceConfig.Config), cmp.Diff(nil, wantSCParsed.Config))
 		}
 	}
+
+	// The xDS resolver is expected to report an error to the channel.
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for an error from the resolver: %v", ctx.Err())
+	case err := <-errCh:
+		if err := verifyResolverError(err, codes.Unavailable, "no valid clusters", nodeID); err != nil {
+			t.Fatal(err)
+		}
+	}
 }
 
 // Tests the case where the resolver receives max stream duration as part of the
@@ -674,7 +687,7 @@ func (s) TestResolverMaxStreamDuration(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
@@ -807,7 +820,7 @@ func (s) TestResolverDelayedOnCommitted(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
 	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
@@ -917,7 +930,7 @@ func (s) TestResolverMultipleLDSUpdates(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure the management server with a listener resource, but no route
 	// configuration resource.
@@ -975,7 +988,7 @@ func (s) TestResolverWRR(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	nodeID := uuid.New().String()
-	mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
@@ -1174,7 +1187,7 @@ func (s) TestConfigSelector_FailureCases(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 			defer cancel()
 			nodeID := uuid.New().String()
-			mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+			mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 			// Build an xDS resolver.
 			stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
@@ -1188,8 +1201,8 @@ func (s) TestConfigSelector_FailureCases(t *testing.T) {
 
 			// Ensure that it returns the expected error.
 			_, err := cs.SelectConfig(iresolver.RPCInfo{Method: methodName, Context: ctx})
-			if err == nil || !strings.Contains(err.Error(), test.wantErr) {
-				t.Errorf("SelectConfig(_) = _, %v; want _, Contains(%v)", err, test.wantErr)
+			if err := verifyResolverError(err, codes.Unavailable, test.wantErr, nodeID); err != nil {
+				t.Fatal(err)
 			}
 		})
 	}
@@ -1413,7 +1426,7 @@ func (s) TestXDSResolverHTTPFilters(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 			defer cancel()
 			nodeID := uuid.New().String()
-			mgmtServer, _, _, bc := setupManagementServerForTest(ctx, t, nodeID)
+			mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 			// Build an xDS resolver.
 			stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
diff --git a/xds/internal/server/listener_wrapper_test.go b/xds/internal/server/listener_wrapper_test.go
deleted file mode 100644
index e181a7b8..00000000
--- a/xds/internal/server/listener_wrapper_test.go
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
- *
- * Copyright 2021 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package server
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"strconv"
-	"testing"
-	"time"
-
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/internal"
-	"google.golang.org/grpc/internal/stubserver"
-	"google.golang.org/grpc/internal/testutils"
-	"google.golang.org/grpc/internal/testutils/xds/e2e"
-	testgrpc "google.golang.org/grpc/interop/grpc_testing"
-	testpb "google.golang.org/grpc/interop/grpc_testing"
-	xdsinternal "google.golang.org/grpc/xds/internal"
-	"google.golang.org/grpc/xds/internal/xdsclient"
-	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
-	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
-
-	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
-	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
-
-	_ "google.golang.org/grpc/xds/internal/httpfilter/router" // Register the router filter
-)
-
-type verifyMode struct {
-	modeCh chan connectivity.ServingMode
-}
-
-func (vm *verifyMode) verifyModeCallback(_ net.Addr, mode connectivity.ServingMode, _ error) {
-	vm.modeCh <- mode
-}
-
-func hostPortFromListener(t *testing.T, lis net.Listener) (string, uint32) {
-	t.Helper()
-
-	host, p, err := net.SplitHostPort(lis.Addr().String())
-	if err != nil {
-		t.Fatalf("net.SplitHostPort(%s) failed: %v", lis.Addr().String(), err)
-	}
-	port, err := strconv.ParseInt(p, 10, 32)
-	if err != nil {
-		t.Fatalf("strconv.ParseInt(%s, 10, 32) failed: %v", p, err)
-	}
-	return host, uint32(port)
-}
-
-// TestListenerWrapper tests the listener wrapper. It configures the listener
-// wrapper with a certain LDS, and makes sure that it requests the LDS name. It
-// then receives an LDS resource that points to an RDS for Route Configuration.
-// The listener wrapper should then start a watch for the RDS name. This should
-// not trigger a mode change (the mode starts out non serving). Then a RDS
-// resource is configured to return for the RDS name. This should transition the
-// Listener Wrapper to READY.
-func (s) TestListenerWrapper(t *testing.T) {
-	mgmtServer, nodeID, ldsResourceNamesCh, rdsResourceNamesCh, xdsC := xdsSetupForTests(t)
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("Failed to create a local TCP listener: %v", err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	modeCh := make(chan connectivity.ServingMode, 1)
-	vm := verifyMode{
-		modeCh: modeCh,
-	}
-	host, port := hostPortFromListener(t, lis)
-	lisResourceName := fmt.Sprintf(e2e.ServerListenerResourceNameTemplate, net.JoinHostPort(host, strconv.Itoa(int(port))))
-	params := ListenerWrapperParams{
-		Listener:             lis,
-		ListenerResourceName: lisResourceName,
-		XDSClient:            xdsC,
-		ModeCallback:         vm.verifyModeCallback,
-	}
-	l := NewListenerWrapper(params)
-	if l == nil {
-		t.Fatalf("NewListenerWrapper(%+v) returned nil", params)
-	}
-	defer l.Close()
-	waitForResourceNames(ctx, t, ldsResourceNamesCh, []string{lisResourceName})
-	// Configure the management server with a listener resource that specifies
-	// the name of RDS resources that need to be resolved.
-	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, route1)
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, route1)},
-		SkipValidation: true,
-	}
-	if err := mgmtServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	waitForResourceNames(ctx, t, rdsResourceNamesCh, []string{route1})
-
-	// Verify that there is no mode change.
-	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
-	defer sCancel()
-	select {
-	case mode := <-modeCh:
-		t.Fatalf("received mode change to %v when no mode expected", mode)
-	case <-sCtx.Done():
-	}
-
-	// Configure the management server with the route configuration resource
-	// specified by the listener resource.
-	resources.Routes = []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(route1, lisResourceName, clusterName)}
-	if err := mgmtServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// Mode should go serving.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timeout waiting for mode change")
-	case mode := <-modeCh:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("mode change received: %v, want: %v", mode, connectivity.ServingModeServing)
-		}
-	}
-
-	// Invoke lds resource not found - should go back to non serving.
-	triggerResourceNotFound := internal.TriggerXDSResourceNotFoundForTesting.(func(xdsclient.XDSClient, xdsresource.Type, string) error)
-	listenerResourceType := xdsinternal.ResourceTypeMapForTesting[version.V3ListenerURL].(xdsresource.Type)
-	if err := triggerResourceNotFound(xdsC, listenerResourceType, listener.GetName()); err != nil {
-		t.Fatalf("Failed to trigger resource name not found for testing: %v", err)
-	}
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timeout waiting for mode change")
-	case mode := <-modeCh:
-		if mode != connectivity.ServingModeNotServing {
-			t.Fatalf("mode change received: %v, want: %v", mode, connectivity.ServingModeNotServing)
-		}
-	}
-}
-
-// TestConnsCleanup tests that the listener wrapper clears it's connection
-// references when connections close. It sets up a listener wrapper and gRPC
-// Server, and connects to the server 100 times and makes an RPC each time, and
-// then closes the connection. After these 100 connections Close, the listener
-// wrapper should have no more references to any connections.
-func (s) TestConnsCleanup(t *testing.T) {
-	mgmtServer, nodeID, _, _, xdsC := xdsSetupForTests(t)
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("Failed to create a local TCP listener: %v", err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	modeCh := make(chan connectivity.ServingMode, 1)
-	vm := verifyMode{
-		modeCh: modeCh,
-	}
-
-	host, port := hostPortFromListener(t, lis)
-	lisResourceName := fmt.Sprintf(e2e.ServerListenerResourceNameTemplate, net.JoinHostPort(host, strconv.Itoa(int(port))))
-	params := ListenerWrapperParams{
-		Listener:             lis,
-		ListenerResourceName: lisResourceName,
-		XDSClient:            xdsC,
-		ModeCallback:         vm.verifyModeCallback,
-	}
-	lw := NewListenerWrapper(params)
-	if lw == nil {
-		t.Fatalf("NewListenerWrapper(%+v) returned nil", params)
-	}
-	defer lw.Close()
-
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, route1)},
-		SkipValidation: true,
-	}
-	if err := mgmtServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-
-	// Wait for Listener Mode to go serving.
-	select {
-	case <-ctx.Done():
-		t.Fatalf("timeout waiting for mode change")
-	case mode := <-modeCh:
-		if mode != connectivity.ServingModeServing {
-			t.Fatalf("mode change received: %v, want: %v", mode, connectivity.ServingModeServing)
-		}
-	}
-
-	ss := &stubserver.StubServer{
-		Listener: lis,
-		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-		S: grpc.NewServer(grpc.Creds(insecure.NewCredentials())),
-	}
-	stubserver.StartTestService(t, ss)
-	defer ss.S.Stop()
-
-	// Make 100 connections to the server, and make an RPC on each one.
-	for i := 0; i < 100; i++ {
-		cc, err := grpc.NewClient(lw.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-		if err != nil {
-			t.Fatalf("grpc.NewClient failed with err: %v", err)
-		}
-		client := testgrpc.NewTestServiceClient(cc)
-		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
-			t.Fatalf("client.EmptyCall() failed: %v", err)
-		}
-		cc.Close()
-	}
-
-	lisWrapper := lw.(*listenerWrapper)
-	// Eventually when the server processes the connection shutdowns, the
-	// listener wrapper should clear its references to the wrapped connections.
-	lenConns := 1
-	for ; ctx.Err() == nil && lenConns > 0; <-time.After(time.Millisecond) {
-		lisWrapper.mu.Lock()
-		lenConns = len(lisWrapper.conns)
-		lisWrapper.mu.Unlock()
-	}
-	if lenConns > 0 {
-		t.Fatalf("timeout waiting for lis wrapper conns to clear, size: %v", lenConns)
-	}
-
-}
diff --git a/xds/server_ext_test.go b/xds/server_ext_test.go
index 1de20d1c..80f33a0b 100644
--- a/xds/server_ext_test.go
+++ b/xds/server_ext_test.go
@@ -25,15 +25,16 @@ import (
 	"net"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"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/grpcsync"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
@@ -42,20 +43,15 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/xds"
-	xdsinternal "google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/xdsclient"
-	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
-	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
-	"github.com/google/uuid"
+	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"
 )
 
-var errAcceptAndClose = status.New(codes.Unavailable, "")
-
 type s struct {
 	grpctest.Tester
 }
@@ -81,159 +77,333 @@ func hostPortFromListener(lis net.Listener) (string, uint32, error) {
 	return host, uint32(port), nil
 }
 
-// TestServingModeChanges tests the Server's logic as it transitions from Not
-// Ready to Ready, then to Not Ready. Before it goes Ready, connections should
-// be accepted and closed. After it goes ready, RPC's should proceed as normal
-// according to matched route configuration. After it transitions back into not
-// ready (through an explicit LDS Resource Not Found), previously running RPC's
-// should be gracefully closed and still work, and new RPC's should fail.
-func (s) TestServingModeChanges(t *testing.T) {
-	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+// servingModeChangeHandler handles changes to the serving mode of an
+// xDS-enabled gRPC server. It logs the changes and sends the new mode and any
+// errors on appropriate channels for the test to consume.
+type servingModeChangeHandler struct {
+	logger interface {
+		Logf(format string, args ...any)
+	}
+	modeCh chan connectivity.ServingMode
+	errCh  chan error
+}
+
+func newServingModeChangeHandler(t *testing.T) *servingModeChangeHandler {
+	return &servingModeChangeHandler{
+		logger: t,
+		modeCh: make(chan connectivity.ServingMode, 1),
+		errCh:  make(chan error, 1),
+	}
+}
+
+func (m *servingModeChangeHandler) modeChangeCallback(addr net.Addr, args xds.ServingModeChangeArgs) {
+	m.logger.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
+	m.modeCh <- args.Mode
+	if args.Err != nil {
+		m.errCh <- args.Err
+	}
+}
+
+// createStubServer creates a new xDS-enabled gRPC server and returns a
+// stubserver.StubServer that can be used for testing. The server is configured
+// with the provided modeChangeOpt and xdsclient.Pool.
+func createStubServer(t *testing.T, lis net.Listener, opts ...grpc.ServerOption) *stubserver.StubServer {
+	stub := &stubserver.StubServer{
+		Listener: lis,
+		EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) {
+			return &testpb.Empty{}, nil
+		},
+		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+			for {
+				if _, err := stream.Recv(); err == io.EOF {
+					return nil
+				} else if err != nil {
+					return err
+				}
+			}
+		},
+	}
+	server, err := xds.NewGRPCServer(opts...)
+	if err != nil {
+		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
+	}
+	stub.S = server
+	stubserver.StartTestService(t, stub)
+	t.Cleanup(stub.Stop)
+	return stub
+}
+
+// waitForSuccessfulRPC waits for an RPC to succeed, repeatedly calling the
+// EmptyCall RPC on the provided client connection until the call succeeds.
+// If the context is canceled or the expected error is not before the context
+// timeout expires, the test will fail.
+func waitForSuccessfulRPC(ctx context.Context, t *testing.T, cc *grpc.ClientConn, opts ...grpc.CallOption) {
+	t.Helper()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatalf("Timeout waiting for RPCs to succeed")
+		case <-time.After(defaultTestShortTimeout):
+			if _, err := client.EmptyCall(ctx, &testpb.Empty{}, opts...); err == nil {
+				return
+			}
+		}
+	}
+}
+
+// waitForFailedRPCWithStatus waits for an RPC to fail with the expected status
+// code, error message, and node ID.  It repeatedly calls the EmptyCall RPC on
+// the provided client connection until the error matches the expected values.
+// If the context is canceled or the expected error is not before the context
+// timeout expires, the test will fail.
+func waitForFailedRPCWithStatus(ctx context.Context, t *testing.T, cc *grpc.ClientConn, wantCode codes.Code, wantErr, wantNodeID string) {
+	t.Helper()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	var err error
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatalf("RPCs failed with most recent error: %v. Want status code %v, error: %s, node id: %s", err, wantCode, wantErr, wantNodeID)
+		case <-time.After(defaultTestShortTimeout):
+			_, err = client.EmptyCall(ctx, &testpb.Empty{})
+			if gotCode := status.Code(err); gotCode != wantCode {
+				continue
+			}
+			if gotErr := err.Error(); !strings.Contains(gotErr, wantErr) {
+				continue
+			}
+			if !strings.Contains(err.Error(), wantNodeID) {
+				continue
+			}
+			t.Logf("Most recent error happy case: %v", err.Error())
+			return
+		}
+	}
+}
+
+// Tests the basic scenario for an xDS enabled gRPC server.
+//
+//   - Verifies that the xDS enabled gRPC server requests for the expected
+//     listener resource.
+//   - Once the listener resource is received from the management server, it
+//     verifies that the xDS enabled gRPC server requests for the appropriate
+//     route configuration name. Also verifies that at this point, the server has
+//     not yet started serving RPCs
+//   - Once the route configuration is received from the management server, it
+//     verifies that the server can serve RPCs successfully.
+func (s) TestServer_Basic(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	listenerNamesCh := make(chan []string, 1)
+	routeNamesCh := make(chan []string, 1)
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
+			switch req.GetTypeUrl() {
+			case "type.googleapis.com/envoy.config.listener.v3.Listener":
+				select {
+				case listenerNamesCh <- req.GetResourceNames():
+				case <-ctx.Done():
+				}
+			case "type.googleapis.com/envoy.config.route.v3.RouteConfiguration":
+				select {
+				case routeNamesCh <- req.GetResourceNames():
+				case <-ctx.Done():
+				}
+			}
+			return nil
+		},
+		AllowResourceSubset: true,
+	})
 
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
 
+	// Create a listener on a local port to act as the xDS enabled gRPC server.
 	lis, err := testutils.LocalTCPListener()
 	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+		t.Fatalf("Failed to listen to local port: %v", err)
 	}
-	// Setup the management server to respond with a listener resource that
-	// specifies a route name to watch. Due to not having received the full
-	// configuration, this should cause the server to be in mode Serving.
 	host, port, err := hostPortFromListener(lis)
 	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
 	}
 
-	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")
+	// Configure the managegement server with a listener resource for the above
+	// xDS enabled gRPC server.
+	const routeConfigName = "routeName"
 	resources := e2e.UpdateOptions{
 		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{listener},
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
 		SkipValidation: true,
 	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
 
-	serving := grpcsync.NewEvent()
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving.Fire()
-		}
-	})
-
-	stub := &stubserver.StubServer{
-		Listener: lis,
-		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-		UnaryCallF: func(context.Context, *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
-			return &testpb.SimpleResponse{}, nil
-		},
-		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
-			for {
-				_, err := stream.Recv() // hangs here forever if stream doesn't shut down...doesn't receive EOF without any errors
-				if err == io.EOF {
-					return nil
-				}
-			}
-		},
-	}
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
 	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
 	if err != nil {
 		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
 	}
 	pool := xdsclient.NewPool(config)
-	sopts := []grpc.ServerOption{grpc.Creds(insecure.NewCredentials()), modeChangeOpt, xds.ClientPoolForTesting(pool)}
-	if stub.S, err = xds.NewGRPCServer(sopts...); err != nil {
-		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Wait for the expected listener resource to be requested.
+	wantLisResourceNames := []string{fmt.Sprintf(e2e.ServerListenerResourceNameTemplate, net.JoinHostPort(host, strconv.Itoa(int(port))))}
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the expected listener resource to be requested")
+	case gotLisResourceName := <-listenerNamesCh:
+		if !cmp.Equal(gotLisResourceName, wantLisResourceNames) {
+			t.Fatalf("Got unexpected listener resource names: %v, want %v", gotLisResourceName, wantLisResourceNames)
+		}
 	}
-	stubserver.StartTestService(t, stub)
-	defer stub.S.Stop()
+
+	// Wait for the expected route config resource to be requested.
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the expected route config resource to be requested")
+	case gotRouteNames := <-routeNamesCh:
+		if !cmp.Equal(gotRouteNames, []string{routeConfigName}) {
+			t.Fatalf("Got unexpected route config resource names: %v, want %v", gotRouteNames, []string{routeConfigName})
+		}
+	}
+
+	// Ensure that the server is not serving RPCs yet.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	select {
+	case <-sCtx.Done():
+	case <-modeChangeHandler.modeCh:
+		t.Fatal("Server started serving RPCs before the route config was received")
+	}
+
+	// Create a gRPC channel to the xDS enabled server.
 	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
 	}
 	defer cc.Close()
 
-	waitForFailedRPCWithStatus(ctx, t, cc, errAcceptAndClose)
-	routeConfig := e2e.RouteConfigNonForwardingAction("routeName")
-	resources = e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Listeners: []*v3listenerpb.Listener{listener},
-		Routes:    []*v3routepb.RouteConfiguration{routeConfig},
+	// Ensure that the server isnt't serving RPCs successfully.
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err == nil || status.Code(err) != codes.Unavailable {
+		t.Fatalf("EmptyCall() returned %v, want %v", err, codes.Unavailable)
 	}
-	defer cancel()
+
+	// Configure the management server with the expected route config resource,
+	// and expext RPCs to succeed.
+	resources.Routes = []*v3routepb.RouteConfiguration{e2e.RouteConfigNonForwardingAction(routeConfigName)}
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
-
 	select {
 	case <-ctx.Done():
-		t.Fatal("timeout waiting for the xDS Server to go Serving")
-	case <-serving.Done():
+		t.Fatal("Timeout waiting for the server to start serving RPCs")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
 	}
+	waitForSuccessfulRPC(ctx, t, cc)
+}
+
+// Tests that the xDS-enabled gRPC server cleans up all its resources when all
+// connections to it are closed.
+func (s) TestServer_ConnectionCleanup(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
-	// A unary RPC should work once it transitions into serving. (need this same
-	// assertion from LDS resource not found triggering it).
-	waitForSuccessfulRPC(ctx, t, cc, grpc.WaitForReady(true))
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
 
-	// Start a stream before switching the server to not serving. Due to the
-	// stream being created before the graceful stop of the underlying
-	// connection, it should be able to continue even after the server switches
-	// to not serving.
-	c := testgrpc.NewTestServiceClient(cc)
-	stream, err := c.FullDuplexCall(ctx)
+	// Create a listener on a local port to act as the xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
 	if err != nil {
-		t.Fatalf("cc.FullDuplexCall failed: %f", err)
+		t.Fatalf("Failed to listen to local port: %v", err)
 	}
-
-	// Lookup the xDS client in use based on the dedicated well-known key, as
-	// defined in A71, used by the xDS enabled gRPC server.
-	xdsC, close, err := pool.GetClientForTesting(xdsclient.NameForServer)
+	host, port, err := hostPortFromListener(lis)
 	if err != nil {
-		t.Fatalf("Failed to find xDS client for configuration: %v", string(bootstrapContents))
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
 	}
-	defer close()
 
-	// Invoke LDS Resource not found here (tests graceful close).
-	triggerResourceNotFound := internal.TriggerXDSResourceNotFoundForTesting.(func(xdsclient.XDSClient, xdsresource.Type, string) error)
-	listenerResourceType := xdsinternal.ResourceTypeMapForTesting[version.V3ListenerURL].(xdsresource.Type)
-	if err := triggerResourceNotFound(xdsC, listenerResourceType, listener.GetName()); err != nil {
-		t.Fatalf("Failed to trigger resource name not found for testing: %v", err)
+	// Configure the managegement server with a listener and route configuration
+	// resource for the above xDS enabled gRPC server.
+	const routeConfigName = "routeName"
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.RouteConfigNonForwardingAction(routeConfigName)},
+		SkipValidation: true,
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
 	}
 
-	// New RPCs on that connection should eventually start failing. Due to
-	// Graceful Stop any started streams continue to work.
-	if err = stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
-		t.Fatalf("stream.Send() failed: %v, should continue to work due to graceful stop", err)
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
 	}
-	if err = stream.CloseSend(); err != nil {
-		t.Fatalf("stream.CloseSend() failed: %v, should continue to work due to graceful stop", err)
+	pool := xdsclient.NewPool(config)
+	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
+		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
+	})
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Create a gRPC channel and verify that RPCs succeed.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
 	}
-	if _, err = stream.Recv(); err != io.EOF {
-		t.Fatalf("unexpected error: %v, expected an EOF error", err)
+	defer cc.Close()
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Create multiple channels to the server, and make an RPC on each one. When
+	// everything is closed, the server should have cleaned up all its resources
+	// as well (and this will be verified by the leakchecker).
+	const numConns = 100
+	var wg sync.WaitGroup
+	wg.Add(numConns)
+	for range numConns {
+		go func() {
+			defer wg.Done()
+			cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+			if err != nil {
+				t.Errorf("grpc.NewClient failed with err: %v", err)
+			}
+			client := testgrpc.NewTestServiceClient(cc)
+			if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+				t.Errorf("EmptyCall() failed: %v", err)
+			}
+			cc.Close()
+		}()
 	}
-
-	// New RPCs on that connection should eventually start failing.
-	waitForFailedRPCWithStatus(ctx, t, cc, errAcceptAndClose)
+	wg.Wait()
 }
 
-// TestMultipleServers_DifferentBootstrapConfigurations verifies that multiple
-// xDS-enabled gRPC servers can be created with different bootstrap
-// configurations, and that they correctly request different LDS resources from
-// the management server based on their respective listening ports.  It also
-// ensures that gRPC clients can connect to the intended server and that RPCs
-// function correctly. The test uses the grpc.Peer() call option to validate
-// that the client is connected to the correct server.
-func (s) TestMultipleServers_DifferentBootstrapConfigurations(t *testing.T) {
-	// Setup an xDS management server.
-	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+// Tests that multiple xDS-enabled gRPC servers can be created with different
+// bootstrap configurations, and that they correctly request different LDS
+// resources from the management server based on their respective listening
+// ports.  It also ensures that gRPC clients can connect to the intended server
+// and that RPCs function correctly. The test uses the grpc.Peer() call option
+// to validate that the client is connected to the correct server.
+func (s) TestServer_MultipleServers_DifferentBootstrapConfigurations(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
 
 	// Create two bootstrap configurations pointing to the above management server.
 	nodeID1 := uuid.New().String()
@@ -251,50 +421,25 @@ func (s) TestMultipleServers_DifferentBootstrapConfigurations(t *testing.T) {
 		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
 	}
 
-	serving1 := grpcsync.NewEvent()
-	modeChangeOpt1 := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving1.Fire()
-		}
-	})
-	serving2 := grpcsync.NewEvent()
-	modeChangeOpt2 := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving2.Fire()
-		}
-	})
-
-	stub1 := &stubserver.StubServer{
-		Listener: lis1,
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-	}
-	if stub1.S, err = xds.NewGRPCServer(xds.BootstrapContentsForTesting(bootstrapContents1), modeChangeOpt1); err != nil {
-		t.Fatalf("Failed to create first xDS enabled gRPC server: %v", err)
-	}
-	stubserver.StartTestService(t, stub1)
-	defer stub1.S.Stop()
-
-	stub2 := &stubserver.StubServer{
-		Listener: lis2,
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
+	modeChangeHandler1 := newServingModeChangeHandler(t)
+	modeChangeOpt1 := xds.ServingModeCallback(modeChangeHandler1.modeChangeCallback)
+	modeChangeHandler2 := newServingModeChangeHandler(t)
+	modeChangeOpt2 := xds.ServingModeCallback(modeChangeHandler2.modeChangeCallback)
+	config1, err := bootstrap.NewConfigFromContents(bootstrapContents1)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents1), err)
 	}
-	if stub2.S, err = xds.NewGRPCServer(xds.BootstrapContentsForTesting(bootstrapContents2), modeChangeOpt2); err != nil {
-		t.Fatalf("Failed to create second xDS enabled gRPC server: %v", err)
+	pool1 := xdsclient.NewPool(config1)
+	config2, err := bootstrap.NewConfigFromContents(bootstrapContents2)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents2), err)
 	}
-	stubserver.StartTestService(t, stub2)
-	defer stub2.S.Stop()
+	pool2 := xdsclient.NewPool(config2)
+	createStubServer(t, lis1, modeChangeOpt1, xds.ClientPoolForTesting(pool1))
+	createStubServer(t, lis2, modeChangeOpt2, xds.ClientPoolForTesting(pool2))
 
 	// Update the management server with the listener resources pointing to the
 	// corresponding gRPC servers.
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
 	host1, port1, err := hostPortFromListener(lis1)
 	if err != nil {
 		t.Fatalf("Failed to retrieve host and port of server: %v", err)
@@ -322,13 +467,19 @@ func (s) TestMultipleServers_DifferentBootstrapConfigurations(t *testing.T) {
 
 	select {
 	case <-ctx.Done():
-		t.Fatal("Timeout waiting for the server 1 to go Serving")
-	case <-serving1.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go SERVING")
+	case gotMode := <-modeChangeHandler1.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
 	}
 	select {
 	case <-ctx.Done():
-		t.Fatal("Timeout waiting for the server 2 to go Serving")
-	case <-serving2.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go SERVING")
+	case gotMode := <-modeChangeHandler2.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
 	}
 
 	// Create two gRPC clients, one for each server.
@@ -357,149 +508,3 @@ func (s) TestMultipleServers_DifferentBootstrapConfigurations(t *testing.T) {
 		t.Errorf("Connected to wrong peer: %s, want %s", peer2.Addr, lis2.Addr())
 	}
 }
-
-// TestResourceNotFoundRDS tests the case where an LDS points to an RDS which
-// returns resource not found. Before getting the resource not found, the xDS
-// Server has not received all configuration needed, so it should Accept and
-// Close any new connections. After it has received the resource not found
-// error, the server should move to serving, successfully Accept Connections,
-// and fail at the L7 level with resource not found specified.
-func (s) TestResourceNotFoundRDS(t *testing.T) {
-	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
-
-	// Create bootstrap configuration pointing to the above management server.
-	nodeID := uuid.New().String()
-	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
-
-	lis, err := testutils.LocalTCPListener()
-	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
-	}
-	// Setup the management server to respond with a listener resource that
-	// specifies a route name to watch, and no RDS resource corresponding to
-	// this route name.
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("failed to retrieve host and port of server: %v", err)
-	}
-
-	const routeConfigResourceName = "routeName"
-	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, routeConfigResourceName)
-	resources := e2e.UpdateOptions{
-		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{listener},
-		SkipValidation: true,
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if err := managementServer.Update(ctx, resources); err != nil {
-		t.Fatal(err)
-	}
-	serving := grpcsync.NewEvent()
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		t.Logf("serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
-		if args.Mode == connectivity.ServingModeServing {
-			serving.Fire()
-		}
-	})
-
-	stub := &stubserver.StubServer{
-		Listener: lis,
-		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
-			return &testpb.Empty{}, nil
-		},
-		UnaryCallF: func(context.Context, *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
-			return &testpb.SimpleResponse{}, nil
-		},
-		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
-			for {
-				_, err := stream.Recv() // hangs here forever if stream doesn't shut down...doesn't receive EOF without any errors
-				if err == io.EOF {
-					return nil
-				}
-			}
-		},
-	}
-	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
-	if err != nil {
-		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
-	}
-	pool := xdsclient.NewPool(config)
-	sopts := []grpc.ServerOption{grpc.Creds(insecure.NewCredentials()), modeChangeOpt, xds.ClientPoolForTesting(pool)}
-	if stub.S, err = xds.NewGRPCServer(sopts...); err != nil {
-		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
-	}
-	stubserver.StartTestService(t, stub)
-	defer stub.S.Stop()
-
-	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
-	}
-	defer cc.Close()
-
-	waitForFailedRPCWithStatus(ctx, t, cc, errAcceptAndClose)
-
-	// Lookup the xDS client in use based on the dedicated well-known key, as
-	// defined in A71, used by the xDS enabled gRPC server.
-	xdsC, close, err := pool.GetClientForTesting(xdsclient.NameForServer)
-	if err != nil {
-		t.Fatalf("Failed to find xDS client for configuration: %v", string(bootstrapContents))
-	}
-	defer close()
-
-	// Invoke resource not found - this should result in L7 RPC error with
-	// unavailable receive on serving as a result, should trigger it to go
-	// serving. Poll as watch might not be started yet to trigger resource not
-	// found.
-	triggerResourceNotFound := internal.TriggerXDSResourceNotFoundForTesting.(func(xdsclient.XDSClient, xdsresource.Type, string) error)
-	routeConfigResourceType := xdsinternal.ResourceTypeMapForTesting[version.V3RouteConfigURL].(xdsresource.Type)
-loop:
-	for {
-		if err := triggerResourceNotFound(xdsC, routeConfigResourceType, routeConfigResourceName); err != nil {
-			t.Fatalf("Failed to trigger resource name not found for testing: %v", err)
-		}
-		select {
-		case <-serving.Done():
-			break loop
-		case <-ctx.Done():
-			t.Fatalf("timed out waiting for serving mode to go serving")
-		case <-time.After(time.Millisecond):
-		}
-	}
-	waitForFailedRPCWithStatus(ctx, t, cc, status.New(codes.Unavailable, "error from xDS configuration for matched route configuration"))
-}
-
-func waitForSuccessfulRPC(ctx context.Context, t *testing.T, cc *grpc.ClientConn, opts ...grpc.CallOption) {
-	t.Helper()
-
-	c := testgrpc.NewTestServiceClient(cc)
-	if _, err := c.EmptyCall(ctx, &testpb.Empty{}, opts...); err != nil {
-		t.Fatalf("rpc EmptyCall() failed: %v", err)
-	}
-}
-
-// waitForFailedRPCWithStatus makes unary RPC's until it receives the expected
-// status in a polling manner. Fails if the RPC made does not return the
-// expected status before the context expires.
-func waitForFailedRPCWithStatus(ctx context.Context, t *testing.T, cc *grpc.ClientConn, st *status.Status) {
-	t.Helper()
-
-	c := testgrpc.NewTestServiceClient(cc)
-	ticker := time.NewTicker(10 * time.Millisecond)
-	defer ticker.Stop()
-	var err error
-	for {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("failure when waiting for RPCs to fail with certain status %v: %v. most recent error received from RPC: %v", st, ctx.Err(), err)
-		case <-ticker.C:
-			_, err = c.EmptyCall(ctx, &testpb.Empty{})
-			if status.Code(err) == st.Code() && strings.Contains(err.Error(), st.Message()) {
-				t.Logf("most recent error happy case: %v", err.Error())
-				return
-			}
-		}
-	}
-}
diff --git a/xds/server_resource_ext_test.go b/xds/server_resource_ext_test.go
new file mode 100644
index 00000000..364d27a0
--- /dev/null
+++ b/xds/server_resource_ext_test.go
@@ -0,0 +1,650 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xds_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
+	"google.golang.org/grpc"
+	"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/testutils"
+	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/xds/bootstrap"
+	"google.golang.org/grpc/xds"
+	xdsinternal "google.golang.org/grpc/xds/internal"
+	"google.golang.org/grpc/xds/internal/xdsclient"
+	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
+	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	v3routerpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3"
+	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/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"
+)
+
+// Tests the case where an LDS points to an RDS which returns resource not
+// found. Before getting the resource not found, the xDS Server has not received
+// all configuration needed, so it should Accept and Close any new connections.
+// After it has received the resource not found error, the server should move to
+// serving, successfully Accept Connections, and fail at the L7 level with
+// resource not found specified.
+func (s) TestServer_RouteConfiguration_ResourceNotFound(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	routeConfigNamesCh := make(chan []string, 1)
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
+			if req.TypeUrl == version.V3RouteConfigURL {
+				select {
+				case routeConfigNamesCh <- req.GetResourceNames():
+				case <-ctx.Done():
+				}
+			}
+			return nil
+		},
+		AllowResourceSubset: true,
+	})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Setup the management server to respond with a listener resource that
+	// specifies a route name to watch, and no RDS resource corresponding to
+	// this route name.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+
+	const routeConfigResourceName = "routeName"
+	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, routeConfigResourceName)
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{listener},
+		SkipValidation: true,
+	}
+
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Wait for the route configuration resource to be requested from the
+	// management server.
+	select {
+	case gotNames := <-routeConfigNamesCh:
+		if !cmp.Equal(gotNames, []string{routeConfigResourceName}) {
+			t.Fatalf("Requested route config resource names: %v, want %v", gotNames, []string{routeConfigResourceName})
+		}
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for route config resource to be requested")
+	}
+
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("failed to dial local test server: %v", err)
+	}
+	defer cc.Close()
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "", "")
+
+	// Lookup the xDS client in use based on the dedicated well-known key, as
+	// defined in A71, used by the xDS enabled gRPC server.
+	xdsC, close, err := pool.GetClientForTesting(xdsclient.NameForServer)
+	if err != nil {
+		t.Fatalf("Failed to find xDS client for configuration: %v", string(bootstrapContents))
+	}
+	defer close()
+
+	// Invoke resource not found error for the route configuration resource.
+	// This should cause the server to go SERVING, but fail RPCs with the
+	// appropriate error code.
+	triggerResourceNotFound := internal.TriggerXDSResourceNotFoundForTesting.(func(xdsclient.XDSClient, xdsresource.Type, string) error)
+	routeConfigResourceType := xdsinternal.ResourceTypeMapForTesting[version.V3RouteConfigURL].(xdsresource.Type)
+	if err := triggerResourceNotFound(xdsC, routeConfigResourceType, routeConfigResourceName); err != nil {
+		t.Fatalf("Failed to trigger resource name not found for testing: %v", err)
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go SERVING")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
+	}
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "error from xDS configuration for matched route configuration", nodeID)
+}
+
+// Tests the scenario where the control plane sends the same resource update. It
+// verifies that the mode change callback is not invoked and client connections
+// to the server are not recycled.
+func (s) TestServer_RedundantUpdateSuppression(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Setup the management server to respond with the listener resources.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	listener := e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, "routeName")
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{listener},
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler.modeCh:
+		if mode != connectivity.ServingModeServing {
+			t.Fatalf("Listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
+		}
+	}
+
+	// Create a ClientConn and make a successful RPCs.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Start a goroutine to make sure that we do not see any connectivity state
+	// changes on the client connection. If redundant updates are not
+	// suppressed, server will recycle client connections.
+	errCh := make(chan error, 1)
+	go func() {
+		prev := connectivity.Ready // We know we are READY since we just did an RPC.
+		for {
+			curr := cc.GetState()
+			if !(curr == connectivity.Ready || curr == connectivity.Idle) {
+				errCh <- fmt.Errorf("unexpected connectivity state change {%s --> %s} on the client connection", prev, curr)
+				return
+			}
+			if !cc.WaitForStateChange(ctx, curr) {
+				// Break out of the for loop when the context has been cancelled.
+				break
+			}
+			prev = curr
+		}
+		errCh <- nil
+	}()
+
+	// Update the management server with the same listener resource. This will
+	// update the resource version though, and should result in a the management
+	// server sending the same resource to the xDS-enabled gRPC server.
+	if err := managementServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{listener},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Since redundant resource updates are suppressed, we should not see the
+	// mode change callback being invoked.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	select {
+	case <-sCtx.Done():
+	case mode := <-modeChangeHandler.modeCh:
+		t.Fatalf("Unexpected mode change callback with new mode %v", mode)
+	}
+
+	// Make sure RPCs continue to succeed.
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Cancel the context to ensure that the WaitForStateChange call exits early
+	// and returns false.
+	cancel()
+	if err := <-errCh; err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Tests the case where the route configuration contains an unsupported route
+// action.  Verifies that RPCs fail with UNAVAILABLE.
+func (s) TestServer_FailWithRouteActionRoute(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Configure the managegement server with a listener and route configuration
+	// resource for the above xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen to local port: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	const routeConfigName = "routeName"
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.RouteConfigNonForwardingAction(routeConfigName)},
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
+		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
+	})
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Create a gRPC channel and verify that RPCs succeed.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Update the route config resource to contain an unsupported action.
+	//
+	// "NonForwardingAction is expected for all Routes used on server-side; a
+	// route with an inappropriate action causes RPCs matching that route to
+	// fail with UNAVAILABLE." - A36
+	resources.Routes = []*v3routepb.RouteConfiguration{e2e.RouteConfigFilterAction("routeName")}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding", nodeID)
+}
+
+// Tests the case where the listener resource is removed from the management
+// server. This should cause the xDS server to transition to NOT_SERVING mode,
+// and the error message should contain the xDS node ID.
+func (s) TestServer_ListenerResourceRemoved(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Configure the managegement server with a listener and route configuration
+	// resource for the above xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen to local port: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	const routeConfigName = "routeName"
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.RouteConfigNonForwardingAction(routeConfigName)},
+		SkipValidation: true,
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go SERVING")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
+	}
+
+	// Create a gRPC channel and verify that RPCs succeed.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Remove the listener resource from the management server. This should
+	// cause the server to go NOT_SERVING, and the error message should contain
+	// the xDS node ID.
+	resources.Listeners = nil
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for server to go NOT_SERVING")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeNotServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeNotServing)
+		}
+		gotErr := <-modeChangeHandler.errCh
+		if gotErr == nil || !strings.Contains(gotErr.Error(), nodeID) {
+			t.Fatalf("Unexpected error: %v, want xDS Node id: %s", gotErr, nodeID)
+		}
+	}
+}
+
+// Tests the case where the listener resource points to a route configuration
+// name that is NACKed. This should trigger the server to move to SERVING,
+// successfully accept connections, and fail at RPCs with an expected error
+// message.
+func (s) TestServer_RouteConfiguration_ResourceNACK(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Configure the managegement server with a listener and route configuration
+	// resource (that will be NACKed) for the above xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen to local port: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	const routeConfigName = "routeName"
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.RouteConfigNoRouteMatch(routeConfigName)},
+		SkipValidation: true,
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Create a gRPC channel and verify that RPCs succeed.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the server to start serving RPCs")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
+	}
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "error from xDS configuration for matched route configuration", nodeID)
+}
+
+// Tests the case where the listener resource points to multiple route
+// configuration resources.
+//
+//   - Initially the listener resource points to three route configuration
+//     resources (A, B and C). The filter chain in the listener matches incoming
+//     connections to route A, and RPCs are expected to succeed.
+//   - A streaming RPC is also kept open at this point.
+//   - The listener resource is then updated to point to two route configuration
+//     resources (A and B). The filter chain in the listener resource does not
+//     match to any of the configured routes. The default filter chain though
+//     matches to route B, which contains a route action of type "Route", and this
+//     is not supported on the server side. New RPCs are expected to fail, while
+//     any ongoing RPCs should be allowed to complete.
+//   - The listener resource is then updated to point to a single route
+//     configuration (A), and the filter chain in the listener matches to route A.
+//     New RPCs are expected to succeed at this point.
+func (s) TestServer_MultipleRouteConfigurations(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Create a listener on a local port to act as the xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen to local port: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+
+	// Setup the management server to respond with a listener resource that
+	// specifies three route names to watch, and the corresponding route
+	// configuration resources.
+	const routeConfigNameA = "routeName-A"
+	const routeConfigNameB = "routeName-B"
+	const routeConfigNameC = "routeName-C"
+	ldsResource := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, routeConfigNameA)
+	ldsResource.FilterChains = append(ldsResource.FilterChains,
+		filterChainWontMatch(t, routeConfigNameB, "1.1.1.1", []uint32{1}),
+		filterChainWontMatch(t, routeConfigNameC, "2.2.2.2", []uint32{2}),
+	)
+	routeConfigA := e2e.RouteConfigNonForwardingAction(routeConfigNameA)
+	routeConfigB := e2e.RouteConfigFilterAction(routeConfigNameB) // Unsupported route action on server.
+	routeConfigC := e2e.RouteConfigFilterAction(routeConfigNameC) // Unsupported route action on server.
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{ldsResource},
+		Routes:         []*v3routepb.RouteConfiguration{routeConfigA, routeConfigB, routeConfigC},
+		SkipValidation: true,
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
+		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
+	})
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Create a gRPC channel and verify that RPCs succeed.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Start a streaming RPC and keep the stream open.
+	client := testgrpc.NewTestServiceClient(cc)
+	stream, err := client.FullDuplexCall(ctx)
+	if err != nil {
+		t.Fatalf("FullDuplexCall failed: %v", err)
+	}
+	if err = stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
+		t.Fatalf("stream.Send() failed: %v, should continue to work due to graceful stop", err)
+	}
+
+	// Update the listener resource such that the filter chain does not match
+	// incoming connections to route A. Instead a default filter chain matches
+	// to route B, which contains an unsupported route action.
+	ldsResource = e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, routeConfigNameA)
+	ldsResource.FilterChains = []*v3listenerpb.FilterChain{filterChainWontMatch(t, routeConfigNameA, "1.1.1.1", []uint32{1})}
+	ldsResource.DefaultFilterChain = filterChainWontMatch(t, routeConfigNameB, "2.2.2.2", []uint32{2})
+	resources.Listeners = []*v3listenerpb.Listener{ldsResource}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// xDS is eventually consistent. So simply poll for the new change to be
+	// reflected.
+	// "NonForwardingAction is expected for all Routes used on server-side; a
+	// route with an inappropriate action causes RPCs matching that route to
+	// fail with UNAVAILABLE." - A36
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding", nodeID)
+
+	// Stream should be allowed to continue on the old working configuration -
+	// as it on a connection that is gracefully closed (old FCM/LDS
+	// Configuration which is allowed to continue).
+	if err = stream.CloseSend(); err != nil {
+		t.Fatalf("stream.CloseSend() failed: %v, should continue to work due to graceful stop", err)
+	}
+	if _, err = stream.Recv(); err != io.EOF {
+		t.Fatalf("unexpected error: %v, expected an EOF error", err)
+	}
+
+	// Update the listener resource to point to a single route configuration
+	// that is expected to match and verify that RPCs succeed.
+	resources.Listeners = []*v3listenerpb.Listener{e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, routeConfigNameA)}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+	waitForSuccessfulRPC(ctx, t, cc)
+}
+
+// filterChainWontMatch returns a filter chain that won't match if running the
+// test locally.
+func filterChainWontMatch(t *testing.T, routeName string, addressPrefix string, srcPorts []uint32) *v3listenerpb.FilterChain {
+	hcm := &v3httppb.HttpConnectionManager{
+		RouteSpecifier: &v3httppb.HttpConnectionManager_Rds{
+			Rds: &v3httppb.Rds{
+				ConfigSource: &v3corepb.ConfigSource{
+					ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{Ads: &v3corepb.AggregatedConfigSource{}},
+				},
+				RouteConfigName: routeName,
+			},
+		},
+		HttpFilters: []*v3httppb.HttpFilter{e2e.HTTPFilter("router", &v3routerpb.Router{})},
+	}
+	return &v3listenerpb.FilterChain{
+		Name: routeName + "-wont-match",
+		FilterChainMatch: &v3listenerpb.FilterChainMatch{
+			PrefixRanges: []*v3corepb.CidrRange{
+				{
+					AddressPrefix: addressPrefix,
+					PrefixLen: &wrapperspb.UInt32Value{
+						Value: uint32(0),
+					},
+				},
+			},
+			SourceType:  v3listenerpb.FilterChainMatch_SAME_IP_OR_LOOPBACK,
+			SourcePorts: srcPorts,
+			SourcePrefixRanges: []*v3corepb.CidrRange{
+				{
+					AddressPrefix: addressPrefix,
+					PrefixLen: &wrapperspb.UInt32Value{
+						Value: uint32(0),
+					},
+				},
+			},
+		},
+		Filters: []*v3listenerpb.Filter{
+			{
+				Name:       "filter-1",
+				ConfigType: &v3listenerpb.Filter_TypedConfig{TypedConfig: testutils.MarshalAny(t, hcm)},
+			},
+		},
+	}
+}
diff --git a/test/xds/xds_server_certificate_providers_test.go b/xds/server_security_ext_test.go
similarity index 77%
rename from test/xds/xds_server_certificate_providers_test.go
rename to xds/server_security_ext_test.go
index 9fcf6f49..28685b02 100644
--- a/test/xds/xds_server_certificate_providers_test.go
+++ b/xds/server_security_ext_test.go
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2023 gRPC authors.
+ * Copyright 2025 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,15 +28,14 @@ import (
 
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	xdscreds "google.golang.org/grpc/credentials/xds"
-	"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/testutils/xds/e2e/setup"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/xds"
+	"google.golang.org/grpc/xds/internal/xdsclient"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -56,33 +55,59 @@ import (
 // verifies that a gRPC client configured with insecure credentials is able to
 // make RPCs to the backend. This ensures that the insecure fallback
 // credentials are getting used on the server.
-func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Success(t *testing.T) {
-	// Spin up an xDS management server.
-	mgmtServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
-
-	// Spin up an xDS-enabled gRPC server that uses xDS credentials with
-	// insecure fallback, and the above bootstrap configuration.
-	lis, cleanup := setupGRPCServer(t, bootstrapContents)
-	defer cleanup()
-
-	// Create an inbound xDS listener resource for the server side that does not
-	// contain any security configuration. This should force the server-side
-	// xdsCredentials to use fallback.
+func (s) TestServer_Security_NoCertProvidersInBootstrap_Success(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server.
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Create a listener on a local port to act as the xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen to local port: %v", err)
+	}
 	host, port, err := hostPortFromListener(lis)
 	if err != nil {
 		t.Fatalf("Failed to retrieve host and port of server: %v", err)
 	}
+
+	// Configure the managegement server with a listener and route configuration
+	// resource for the above xDS enabled gRPC server.
+	const routeConfigName = "routeName"
 	resources := e2e.UpdateOptions{
 		NodeID:         nodeID,
-		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListener(host, port, e2e.SecurityLevelNone, "routeName")},
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")},
+		Routes:         []*v3routepb.RouteConfiguration{e2e.RouteConfigNonForwardingAction(routeConfigName)},
 		SkipValidation: true,
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if err := mgmtServer.Update(ctx, resources); err != nil {
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration
+	// and configure xDS credentials to be used on the server-side.
+	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{
+		FallbackCreds: insecure.NewCredentials(),
+	})
+	if err != nil {
 		t.Fatal(err)
 	}
 
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
+		t.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
+	})
+	createStubServer(t, lis, grpc.Creds(creds), modeChangeOpt, xds.ClientPoolForTesting(pool))
+
 	// Create a client that uses insecure creds and verify RPCs.
 	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
@@ -104,7 +129,7 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Success(t *test
 // instance name specified in the Listener resource will not be present in the
 // bootstrap file. The test verifies that server creation does not fail and that
 // the xDS-enabled gRPC server does not enter "serving" mode.
-func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *testing.T) {
+func (s) TestServer_Security_NoCertificateProvidersInBootstrap_Failure(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -130,7 +155,7 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 
 	// Create bootstrap configuration with no certificate providers.
 	nodeID := uuid.New().String()
-	bs, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
+	bootstrapContents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
 		Servers: []byte(fmt.Sprintf(`[{
 			"server_uri": %q,
 			"channel_creds": [{"type": "insecure"}]
@@ -142,46 +167,37 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
 
-	// Configure xDS credentials with an insecure fallback to be used on the
-	// server-side.
-	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
+	// Create a listener on a local port to act as the xDS enabled gRPC server.
+	lis, err := testutils.LocalTCPListener()
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("Failed to listen to local port: %v", err)
 	}
-
-	// Initialize an xDS-enabled gRPC server and register the stubServer on it.
-	// Pass it a mode change server option that pushes on a channel the mode
-	// changes to "not serving".
-	servingModeCh := make(chan struct{})
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		if args.Mode == connectivity.ServingModeServing {
-			close(servingModeCh)
-		}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration
+	// and configure xDS credentials to be used on the server-side.
+	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{
+		FallbackCreds: insecure.NewCredentials(),
 	})
-
-	// Create a local listener and assign it to the stub server.
-	lis, err := testutils.LocalTCPListener()
 	if err != nil {
-		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+		t.Fatal(err)
 	}
 
-	stub := &stubserver.StubServer{
-		Listener: lis,
-	}
-	if stub.S, err = xds.NewGRPCServer(grpc.Creds(creds), modeChangeOpt, xds.BootstrapContentsForTesting(bs)); err != nil {
-		t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
 	}
-	defer stub.S.Stop()
-	stubserver.StartTestService(t, stub)
+	pool := xdsclient.NewPool(config)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, grpc.Creds(creds), modeChangeOpt, xds.ClientPoolForTesting(pool))
 
 	// Create an inbound xDS listener resource for the server side that contains
 	// mTLS security configuration. Since the received certificate provider
 	// instance name would be missing in the bootstrap configuration, this
 	// resource is expected to NACKed by the xDS client.
-	host, port, err := hostPortFromListener(lis)
-	if err != nil {
-		t.Fatalf("Failed to retrieve host and port of server: %v", err)
-	}
 	resources := e2e.UpdateOptions{
 		NodeID:         nodeID,
 		Listeners:      []*v3listenerpb.Listener{e2e.DefaultServerListener(host, port, e2e.SecurityLevelMTLS, "routeName")},
@@ -200,10 +216,12 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 
 	// Wait a short duration and ensure that the server does not enter "serving"
 	// mode.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
 	select {
-	case <-time.After(2 * defaultTestShortTimeout):
-	case <-servingModeCh:
-		t.Fatal("Server changed to serving mode when not expected to")
+	case <-sCtx.Done():
+	case <-modeChangeHandler.modeCh:
+		t.Fatal("Server started serving RPCs before the route config was received")
 	}
 
 	// Create a client that uses insecure creds and verify that RPCs don't
@@ -214,7 +232,7 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 	}
 	defer cc.Close()
 
-	waitForFailedRPCWithStatus(ctx, t, cc, errAcceptAndClose)
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "", "")
 }
 
 // Tests the case where the bootstrap configuration contains one certificate
@@ -228,7 +246,7 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 //
 // The test verifies that an RPC to the first listener succeeds, while the
 // second listener never moves to "serving" mode and RPCs to it fail.
-func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T) {
+func (s) TestServer_Security_WithValidAndInvalidSecurityConfiguration(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -256,7 +274,7 @@ func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 	nodeID := uuid.New().String()
 	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
 
-	// Create two local listeners.
+	// Create two xDS-enabled gRPC servers using the above bootstrap configs.
 	lis1, err := testutils.LocalTCPListener()
 	if err != nil {
 		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
@@ -266,26 +284,21 @@ func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
 	}
 
-	// Create an xDS-enabled gRPC server that is configured to use xDS
-	// credentials and assigned to a stub server, configuring a mode change
-	// option that closes a channel when listener2 enters serving mode.
+	modeChangeHandler1 := newServingModeChangeHandler(t)
+	modeChangeOpt1 := xds.ServingModeCallback(modeChangeHandler1.modeChangeCallback)
+	modeChangeHandler2 := newServingModeChangeHandler(t)
+	modeChangeOpt2 := xds.ServingModeCallback(modeChangeHandler2.modeChangeCallback)
 	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
 	if err != nil {
 		t.Fatal(err)
 	}
-	servingModeCh := make(chan struct{})
-	modeChangeOpt := xds.ServingModeCallback(func(addr net.Addr, args xds.ServingModeChangeArgs) {
-		if addr.String() == lis2.Addr().String() {
-			if args.Mode == connectivity.ServingModeServing {
-				close(servingModeCh)
-			}
-		}
-	})
-
-	stub1 := createStubServer(t, lis1, creds, modeChangeOpt, bootstrapContents)
-	defer stub1.S.Stop()
-	stub2 := createStubServer(t, lis2, creds, modeChangeOpt, bootstrapContents)
-	defer stub2.S.Stop()
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	createStubServer(t, lis1, grpc.Creds(creds), modeChangeOpt1, xds.ClientPoolForTesting(pool))
+	createStubServer(t, lis2, grpc.Creds(creds), modeChangeOpt2, xds.ClientPoolForTesting(pool))
 
 	// Create inbound xDS listener resources for the server side that contains
 	// mTLS security configuration.
@@ -439,7 +452,7 @@ func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 	// mode.
 	select {
 	case <-time.After(2 * defaultTestShortTimeout):
-	case <-servingModeCh:
+	case <-modeChangeHandler2.modeCh:
 		t.Fatal("Server changed to serving mode when not expected to")
 	}
 
@@ -451,5 +464,5 @@ func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 	}
 	defer cc2.Close()
 
-	waitForFailedRPCWithStatus(ctx, t, cc2, errAcceptAndClose)
+	waitForFailedRPCWithStatus(ctx, t, cc2, codes.Unavailable, "", "")
 }
diff --git a/xds/server_serving_mode_ext_test.go b/xds/server_serving_mode_ext_test.go
new file mode 100644
index 00000000..ce4f65f1
--- /dev/null
+++ b/xds/server_serving_mode_ext_test.go
@@ -0,0 +1,349 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xds_test
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/testutils/xds/e2e/setup"
+	"google.golang.org/grpc/internal/xds/bootstrap"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/xds"
+	"google.golang.org/grpc/xds/internal/xdsclient"
+
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
+)
+
+// Tests the Server's logic as it transitions from NOT_SERVING to SERVING, then
+// to NOT_SERVING again. Before it goes to SERVING, connections should be
+// accepted and closed. After it goes SERVING, RPC's should proceed as normal
+// according to matched route configuration. After it transitions back into
+// NOT_SERVING, (through an explicit LDS Resource Not Found), previously running
+// RPC's should be gracefully closed and still work, and new RPC's should fail.
+func (s) TestServer_ServingModeChanges_SingleServer(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Setup the management server to respond with a listener resource that
+	// specifies a route name to watch. Due to not having received the full
+	// configuration, this should cause the server to be in mode NOT_SERVING.
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+	}
+	host, port, err := hostPortFromListener(lis)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	listener := e2e.DefaultServerListenerWithRouteConfigName(host, port, e2e.SecurityLevelNone, "routeName")
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{listener},
+		SkipValidation: true,
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start an xDS-enabled gRPC server with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	modeChangeHandler := newServingModeChangeHandler(t)
+	modeChangeOpt := xds.ServingModeCallback(modeChangeHandler.modeChangeCallback)
+	createStubServer(t, lis, modeChangeOpt, xds.ClientPoolForTesting(pool))
+
+	// Start a gRPC channel to the above server. The server is yet to receive
+	// route configuration, and therefore RPCs must fail at this time.
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to dial local test server: %v", err)
+	}
+	defer cc.Close()
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "", "")
+
+	// Setup the route configuration resource on the management server. This
+	// should cause the xDS-enabled gRPC server to move to SERVING mode.
+	routeConfig := e2e.RouteConfigNonForwardingAction("routeName")
+	resources = e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{listener},
+		Routes:         []*v3routepb.RouteConfiguration{routeConfig},
+		SkipValidation: true,
+	}
+	defer cancel()
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go SERVING")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeServing)
+		}
+	}
+	waitForSuccessfulRPC(ctx, t, cc)
+
+	// Start a stream before switching the server to not serving. Due to the
+	// stream being created before the graceful stop of the underlying
+	// connection, it should be able to continue even after the server switches
+	// to not serving.
+	c := testgrpc.NewTestServiceClient(cc)
+	stream, err := c.FullDuplexCall(ctx)
+	if err != nil {
+		t.Fatalf("cc.FullDuplexCall failed: %f", err)
+	}
+
+	// Remove the listener resource from the management server.
+	resources.Listeners = nil
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Ensure the server is in NOT_SERVING mode.
+	select {
+	case <-ctx.Done():
+		t.Fatal("Timeout waiting for the xDS-enabled gRPC server to go NOT_SERVING")
+	case gotMode := <-modeChangeHandler.modeCh:
+		if gotMode != connectivity.ServingModeNotServing {
+			t.Fatalf("Mode changed to %v, want %v", gotMode, connectivity.ServingModeNotServing)
+		}
+		gotErr := <-modeChangeHandler.errCh
+		if gotErr == nil || !strings.Contains(gotErr.Error(), nodeID) {
+			t.Fatalf("Unexpected error: %v, want xDS Node id: %s", gotErr, nodeID)
+		}
+	}
+
+	// Due to graceful stop, any started streams continue to work.
+	if err = stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
+		t.Fatalf("stream.Send() failed: %v, should continue to work due to graceful stop", err)
+	}
+	if err = stream.CloseSend(); err != nil {
+		t.Fatalf("stream.CloseSend() failed: %v, should continue to work due to graceful stop", err)
+	}
+	if _, err = stream.Recv(); err != io.EOF {
+		t.Fatalf("stream.Recv() failed with %v, want io.EOF", err)
+	}
+
+	// New RPCs on that connection should eventually start failing.
+	waitForFailedRPCWithStatus(ctx, t, cc, codes.Unavailable, "", "")
+}
+
+// Tests the serving mode functionality with multiple xDS enabled gRPC servers.
+func (s) TestServer_ServingModeChanges_MultipleServers(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	managementServer, nodeID, bootstrapContents, _ := setup.ManagementServerAndResolver(t)
+
+	// Create two local listeners and pass it to Serve().
+	lis1, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+	}
+	lis2, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+	}
+
+	// Create a server option to get notified about serving mode changes.
+	modeChangeHandler1 := newServingModeChangeHandler(t)
+	modeChangeOpt1 := xds.ServingModeCallback(modeChangeHandler1.modeChangeCallback)
+	modeChangeHandler2 := newServingModeChangeHandler(t)
+	modeChangeOpt2 := xds.ServingModeCallback(modeChangeHandler2.modeChangeCallback)
+
+	// Start two xDS-enabled gRPC servers with the above bootstrap configuration.
+	config, err := bootstrap.NewConfigFromContents(bootstrapContents)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bootstrapContents), err)
+	}
+	pool := xdsclient.NewPool(config)
+	createStubServer(t, lis1, modeChangeOpt1, xds.ClientPoolForTesting(pool))
+	createStubServer(t, lis2, modeChangeOpt2, xds.ClientPoolForTesting(pool))
+
+	// Setup the management server to respond with server-side Listener
+	// resources for both listeners.
+	host1, port1, err := hostPortFromListener(lis1)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	listener1 := e2e.DefaultServerListener(host1, port1, e2e.SecurityLevelNone, "routeName")
+	host2, port2, err := hostPortFromListener(lis2)
+	if err != nil {
+		t.Fatalf("Failed to retrieve host and port of server: %v", err)
+	}
+	listener2 := e2e.DefaultServerListener(host2, port2, e2e.SecurityLevelNone, "routeName")
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{listener1, listener2},
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for both listeners to move to "serving" mode.
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler1.modeCh:
+		if mode != connectivity.ServingModeServing {
+			t.Fatalf("Listener 1 received new mode %v, want %v", mode, connectivity.ServingModeServing)
+		}
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler2.modeCh:
+		if mode != connectivity.ServingModeServing {
+			t.Fatalf("Listener 2 received new mode %v, want %v", mode, connectivity.ServingModeServing)
+		}
+	}
+
+	// Create a ClientConn to the first listener and make a successful RPCs.
+	cc1, err := grpc.NewClient(lis1.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient() failed: %v", err)
+	}
+	defer cc1.Close()
+	waitForSuccessfulRPC(ctx, t, cc1)
+
+	// Create a ClientConn to the second listener and make a successful RPCs.
+	cc2, err := grpc.NewClient(lis2.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient() failed: %v", err)
+	}
+	defer cc2.Close()
+	waitForSuccessfulRPC(ctx, t, cc2)
+
+	// Update the management server to remove the second listener resource. This
+	// should push only the second listener into "not-serving" mode.
+	if err := managementServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{listener1},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for lis2 to move to "not-serving" mode.
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler2.modeCh:
+		if mode != connectivity.ServingModeNotServing {
+			t.Fatalf("Listener received new mode %v, want %v", mode, connectivity.ServingModeNotServing)
+		}
+		gotErr := <-modeChangeHandler2.errCh
+		if gotErr == nil || !strings.Contains(gotErr.Error(), nodeID) {
+			t.Fatalf("Unexpected error: %v, want xDS Node id: %s", gotErr, nodeID)
+		}
+	}
+
+	// Make sure RPCs succeed on cc1 and fail on cc2.
+	waitForSuccessfulRPC(ctx, t, cc1)
+	waitForFailedRPCWithStatus(ctx, t, cc2, codes.Unavailable, "", "")
+
+	// Update the management server to remove the first listener resource as
+	// well. This should push the first listener into "not-serving" mode. Second
+	// listener is already in "not-serving" mode.
+	if err := managementServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for lis1 to move to "not-serving" mode. lis2 was already removed
+	// from the xdsclient's resource cache. So, lis2's callback will not be
+	// invoked this time around.
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler1.modeCh:
+		if mode != connectivity.ServingModeNotServing {
+			t.Fatalf("Listener received new mode %v, want %v", mode, connectivity.ServingModeNotServing)
+		}
+		gotErr := <-modeChangeHandler1.errCh
+		if gotErr == nil || !strings.Contains(gotErr.Error(), nodeID) {
+			t.Fatalf("Unexpected error: %v, want xDS Node id: %s", gotErr, nodeID)
+		}
+	}
+
+	// Make sure RPCs fail on both.
+	waitForFailedRPCWithStatus(ctx, t, cc1, codes.Unavailable, "", "")
+	waitForFailedRPCWithStatus(ctx, t, cc2, codes.Unavailable, "", "")
+
+	// Make sure new connection attempts to "not-serving" servers fail.
+	if cc1, err = grpc.NewClient(lis1.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())); err != nil {
+		t.Fatal("Failed to create clientConn to a server in \"not-serving\" state")
+	}
+	defer cc1.Close()
+	if _, err := testgrpc.NewTestServiceClient(cc1).FullDuplexCall(ctx); status.Code(err) != codes.Unavailable {
+		t.Fatalf("FullDuplexCall failed with status code: %v, want: Unavailable", status.Code(err))
+	}
+
+	// Update the management server with both listener resources.
+	if err := managementServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{listener1, listener2},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for both listeners to move to "serving" mode.
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler1.modeCh:
+		if mode != connectivity.ServingModeServing {
+			t.Fatalf("Listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
+		}
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for a mode change update: %v", err)
+	case mode := <-modeChangeHandler2.modeCh:
+		if mode != connectivity.ServingModeServing {
+			t.Fatalf("Listener received new mode %v, want %v", mode, connectivity.ServingModeServing)
+		}
+	}
+
+	// The clientConns created earlier should be able to make RPCs now.
+	waitForSuccessfulRPC(ctx, t, cc1)
+	waitForSuccessfulRPC(ctx, t, cc2)
+}