Repository: grpc/grpc-go

Issue:
stats/opentelemetry: Write e2e tests for tracing more than single message sent and received

In all of the [e2e tests](https://github.com/grpc/grpc-go/blob/732f3f32f5365f4f5b89a3f1be19534cd28db794/stats/opentelemetry/e2e_test.go), we only have single message being sent/received, which means that the logic on the sequence numbers incremented is untested. Write a separate test for streaming RPC where multiple messages are being sent and received and validate the traces accordingly

Discussion:
@vinothkumarr227: @purnesh42H Please assign the issue to me.

Solution patch:
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
index d5ad4a5c..494314f2 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
@@ -923,6 +923,5 @@ func (al *addressList) hasNext() bool {
 // fields that are meaningful to the SubConn.
 func equalAddressIgnoringBalAttributes(a, b *resolver.Address) bool {
 	return a.Addr == b.Addr && a.ServerName == b.ServerName &&
-		a.Attributes.Equal(b.Attributes) &&
-		a.Metadata == b.Metadata
+		a.Attributes.Equal(b.Attributes)
 }
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index cc5713fd..f122a041 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -36,7 +36,7 @@ var (
 	// LeastRequestLB is set if we should support the least_request_experimental
 	// LB policy, which can be enabled by setting the environment variable
 	// "GRPC_EXPERIMENTAL_ENABLE_LEAST_REQUEST" to "true".
-	LeastRequestLB = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_LEAST_REQUEST", false)
+	LeastRequestLB = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_LEAST_REQUEST", true)
 	// ALTSMaxConcurrentHandshakes is the maximum number of concurrent ALTS
 	// handshakes that can be performed.
 	ALTSMaxConcurrentHandshakes = uint64FromEnv("GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES", 100, 1, 100)
diff --git a/internal/xds/bootstrap/bootstrap.go b/internal/xds/bootstrap/bootstrap.go
index 69b7ee80..142e8039 100644
--- a/internal/xds/bootstrap/bootstrap.go
+++ b/internal/xds/bootstrap/bootstrap.go
@@ -206,7 +206,7 @@ func (sc *ServerConfig) ServerFeatures() []string {
 //
 // This feature controls the behavior of the xDS client when the server deletes
 // a previously sent Listener or Cluster resource. If set, the xDS client will
-// not invoke the watchers' OnResourceDoesNotExist() method when a resource is
+// not invoke the watchers' ResourceError() method when a resource is
 // deleted, nor will it remove the existing resource value from its cache.
 func (sc *ServerConfig) ServerFeaturesIgnoreResourceDeletion() bool {
 	for _, sf := range sc.serverFeatures {
diff --git a/stats/opentelemetry/client_tracing.go b/stats/opentelemetry/client_tracing.go
index 8aea7c9d..4ddaf095 100644
--- a/stats/opentelemetry/client_tracing.go
+++ b/stats/opentelemetry/client_tracing.go
@@ -60,7 +60,7 @@ func (h *clientStatsHandler) createCallTraceSpan(ctx context.Context, method str
 		logger.Error("TraceProvider is not provided in trace options")
 		return ctx, nil
 	}
-	mn := strings.Replace(removeLeadingSlash(method), "/", ".", -1)
+	mn := "Sent." + strings.Replace(removeLeadingSlash(method), "/", ".", -1)
 	tracer := h.options.TraceOptions.TracerProvider.Tracer(tracerName, trace.WithInstrumentationVersion(grpc.Version))
 	ctx, span := tracer.Start(ctx, mn, trace.WithSpanKind(trace.SpanKindClient))
 	return ctx, span
diff --git a/stats/opentelemetry/server_tracing.go b/stats/opentelemetry/server_tracing.go
index f04d28c5..26185519 100644
--- a/stats/opentelemetry/server_tracing.go
+++ b/stats/opentelemetry/server_tracing.go
@@ -33,7 +33,7 @@ import (
 // If TextMapPropagator is not provided in the trace options, it returns context
 // as is.
 func (h *serverStatsHandler) traceTagRPC(ctx context.Context, ai *attemptInfo) (context.Context, *attemptInfo) {
-	mn := strings.Replace(ai.method, "/", ".", -1)
+	mn := "Recv." + strings.Replace(ai.method, "/", ".", -1)
 	var span trace.Span
 	tracer := h.options.TraceOptions.TracerProvider.Tracer(tracerName, trace.WithInstrumentationVersion(grpc.Version))
 	ctx = h.options.TraceOptions.TextMapPropagator.Extract(ctx, otelinternaltracing.NewIncomingCarrier(ctx))
diff --git a/stats/opentelemetry/trace.go b/stats/opentelemetry/trace.go
index 4a49d8b0..efafdd07 100644
--- a/stats/opentelemetry/trace.go
+++ b/stats/opentelemetry/trace.go
@@ -57,18 +57,24 @@ func populateSpan(rs stats.RPCStats, ai *attemptInfo) {
 	case *stats.InPayload:
 		// message id - "must be calculated as two different counters starting
 		// from one for sent messages and one for received messages."
-		span.AddEvent("Inbound compressed message", trace.WithAttributes(
+		attrs := []attribute.KeyValue{
 			attribute.Int64("sequence-number", int64(ai.countRecvMsg)),
 			attribute.Int64("message-size", int64(rs.Length)),
-			attribute.Int64("message-size-compressed", int64(rs.CompressedLength)),
-		))
+		}
+		if rs.CompressedLength != rs.Length {
+			attrs = append(attrs, attribute.Int64("message-size-compressed", int64(rs.CompressedLength)))
+		}
+		span.AddEvent("Inbound message", trace.WithAttributes(attrs...))
 		ai.countRecvMsg++
 	case *stats.OutPayload:
-		span.AddEvent("Outbound compressed message", trace.WithAttributes(
+		attrs := []attribute.KeyValue{
 			attribute.Int64("sequence-number", int64(ai.countSentMsg)),
 			attribute.Int64("message-size", int64(rs.Length)),
-			attribute.Int64("message-size-compressed", int64(rs.CompressedLength)),
-		))
+		}
+		if rs.CompressedLength != rs.Length {
+			attrs = append(attrs, attribute.Int64("message-size-compressed", int64(rs.CompressedLength)))
+		}
+		span.AddEvent("Outbound message", trace.WithAttributes(attrs...))
 		ai.countSentMsg++
 	case *stats.End:
 		if rs.Error != nil {
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer.go b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
index 3572ff34..fa34748c 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer.go
@@ -542,38 +542,27 @@ func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpd
 	}
 }
 
-// Handles an error Cluster update from the xDS client. Propagates the error
-// down to the child policy if one exists, or puts the channel in
-// TRANSIENT_FAILURE.
+// Handles an ambient error Cluster update from the xDS client to not stop
+// using the previously seen resource.
 //
 // Only executed in the context of a serializer callback.
-func (b *cdsBalancer) onClusterError(name string, err error) {
-	b.logger.Warningf("Cluster resource %q received error update: %v", name, err)
+func (b *cdsBalancer) onClusterAmbientError(name string, err error) {
+	b.logger.Warningf("Cluster resource %q received ambient error update: %v", name, err)
 
-	if b.childLB != nil {
-		if xdsresource.ErrType(err) != xdsresource.ErrorTypeConnection {
-			// Connection errors will be sent to the child balancers directly.
-			// There's no need to forward them.
-			b.childLB.ResolverError(err)
-		}
-	} else {
-		// If child balancer was never created, fail the RPCs with
-		// errors.
-		b.ccw.UpdateState(balancer.State{
-			ConnectivityState: connectivity.TransientFailure,
-			Picker:            base.NewErrPicker(fmt.Errorf("%q: %v", name, err)),
-		})
+	if xdsresource.ErrType(err) != xdsresource.ErrorTypeConnection && b.childLB != nil {
+		// Connection errors will be sent to the child balancers directly.
+		// There's no need to forward them.
+		b.childLB.ResolverError(err)
 	}
 }
 
-// Handles a resource-not-found error from the xDS client. Propagates the error
-// down to the child policy if one exists, or puts the channel in
-// TRANSIENT_FAILURE.
+// Handles an error Cluster update from the xDS client to stop using the
+// previously seen resource. Propagates the error down to the child policy
+// if one exists, and puts the channel in TRANSIENT_FAILURE.
 //
 // Only executed in the context of a serializer callback.
-func (b *cdsBalancer) onClusterResourceNotFound(name string) {
-	b.logger.Warningf("CDS watch for resource %q reported resource-does-not-exist error", name)
-	err := b.annotateErrorWithNodeID(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "cluster %q not found", name))
+func (b *cdsBalancer) onClusterResourceError(name string, err error) {
+	b.logger.Warningf("CDS watch for resource %q reported resource error", name)
 	b.closeChildPolicyAndReportTF(err)
 }
 
@@ -677,6 +666,14 @@ func (b *cdsBalancer) generateDMsForCluster(name string, depth int, dms []cluste
 	return append(dms, dm), true, nil
 }
 
+func (b *cdsBalancer) onClusterError(name string, err error) {
+	if b.childLB != nil {
+		b.onClusterAmbientError(name, err)
+	} else {
+		b.onClusterResourceError(name, err)
+	}
+}
+
 // ccWrapper wraps the balancer.ClientConn passed to the CDS balancer at
 // creation and intercepts the NewSubConn() and UpdateAddresses() call from the
 // child policy to add security configuration required by xDS credentials.
diff --git a/xds/internal/balancer/cdsbalancer/cluster_watcher.go b/xds/internal/balancer/cdsbalancer/cluster_watcher.go
index 835461d0..a9adea0c 100644
--- a/xds/internal/balancer/cdsbalancer/cluster_watcher.go
+++ b/xds/internal/balancer/cdsbalancer/cluster_watcher.go
@@ -32,19 +32,19 @@ type clusterWatcher struct {
 	parent *cdsBalancer
 }
 
-func (cw *clusterWatcher) OnUpdate(u *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcher) ResourceChanged(u *xdsresource.ClusterResourceData, onDone func()) {
 	handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u.Resource); onDone() }
 	cw.parent.serializer.ScheduleOr(handleUpdate, onDone)
 }
 
-func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
-	handleError := func(context.Context) { cw.parent.onClusterError(cw.name, err); onDone() }
-	cw.parent.serializer.ScheduleOr(handleError, onDone)
+func (cw *clusterWatcher) ResourceError(err error, onDone func()) {
+	handleResourceError := func(context.Context) { cw.parent.onClusterResourceError(cw.name, err); onDone() }
+	cw.parent.serializer.ScheduleOr(handleResourceError, onDone)
 }
 
-func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	handleNotFound := func(context.Context) { cw.parent.onClusterResourceNotFound(cw.name); onDone() }
-	cw.parent.serializer.ScheduleOr(handleNotFound, onDone)
+func (cw *clusterWatcher) AmbientError(err error, onDone func()) {
+	handleError := func(context.Context) { cw.parent.onClusterAmbientError(cw.name, err); onDone() }
+	cw.parent.serializer.ScheduleOr(handleError, onDone)
 }
 
 // watcherState groups the state associated with a clusterWatcher.
diff --git a/xds/internal/balancer/clusterresolver/resource_resolver.go b/xds/internal/balancer/clusterresolver/resource_resolver.go
index d9315c3a..c1a656c5 100644
--- a/xds/internal/balancer/clusterresolver/resource_resolver.go
+++ b/xds/internal/balancer/clusterresolver/resource_resolver.go
@@ -38,7 +38,7 @@ type resourceUpdate struct {
 	priorities []priorityConfig
 	// To be invoked once the update is completely processed, or is dropped in
 	// favor of a newer update.
-	onDone xdsresource.OnDoneFunc
+	onDone func()
 }
 
 // topLevelResolver is used by concrete endpointsResolver implementations for
@@ -50,7 +50,7 @@ type topLevelResolver interface {
 	// endpointsResolver implementation. The onDone callback is to be invoked
 	// once the update is completely processed, or is dropped in favor of a
 	// newer update.
-	onUpdate(onDone xdsresource.OnDoneFunc)
+	onUpdate(onDone func())
 }
 
 // endpointsResolver wraps the functionality to resolve a given resource name to
@@ -282,7 +282,7 @@ func (rr *resourceResolver) stop(closing bool) {
 // clusterresolver LB policy.
 //
 // Caller must hold rr.mu.
-func (rr *resourceResolver) generateLocked(onDone xdsresource.OnDoneFunc) {
+func (rr *resourceResolver) generateLocked(onDone func()) {
 	var ret []priorityConfig
 	for _, rDM := range rr.children {
 		u, ok := rDM.r.lastUpdate()
@@ -312,7 +312,7 @@ func (rr *resourceResolver) generateLocked(onDone xdsresource.OnDoneFunc) {
 	rr.updateChannel <- &resourceUpdate{priorities: ret, onDone: onDone}
 }
 
-func (rr *resourceResolver) onUpdate(onDone xdsresource.OnDoneFunc) {
+func (rr *resourceResolver) onUpdate(onDone func()) {
 	handleUpdate := func(context.Context) {
 		rr.mu.Lock()
 		rr.generateLocked(onDone)
diff --git a/xds/internal/balancer/clusterresolver/resource_resolver_eds.go b/xds/internal/balancer/clusterresolver/resource_resolver_eds.go
index 31a06c4c..043def95 100644
--- a/xds/internal/balancer/clusterresolver/resource_resolver_eds.go
+++ b/xds/internal/balancer/clusterresolver/resource_resolver_eds.go
@@ -75,8 +75,8 @@ func newEDSResolver(nameToWatch string, producer xdsresource.Producer, topLevelR
 	return ret
 }
 
-// OnUpdate is invoked to report an update for the resource being watched.
-func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
+// ResourceChanged is invoked to report an update for the resource being watched.
+func (er *edsDiscoveryMechanism) ResourceChanged(update *xdsresource.EndpointsResourceData, onDone func()) {
 	if er.stopped.HasFired() {
 		onDone()
 		return
@@ -89,54 +89,36 @@ func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceD
 	er.topLevelResolver.onUpdate(onDone)
 }
 
-func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (er *edsDiscoveryMechanism) ResourceError(err error, onDone func()) {
 	if er.stopped.HasFired() {
 		onDone()
 		return
 	}
 
 	if er.logger.V(2) {
-		er.logger.Infof("EDS discovery mechanism for resource %q reported error: %v", er.nameToWatch, err)
+		er.logger.Infof("EDS discovery mechanism for resource %q reported resource error: %v", er.nameToWatch, err)
 	}
 
-	er.mu.Lock()
-	if er.update != nil {
-		// Continue using a previously received good configuration if one
-		// exists.
-		er.mu.Unlock()
-		onDone()
-		return
-	}
-
-	// Else report an empty update that would result in no priority child being
+	// Report an empty update that would result in no priority child being
 	// created for this discovery mechanism. This would result in the priority
 	// LB policy reporting TRANSIENT_FAILURE (as there would be no priorities or
 	// localities) if this was the only discovery mechanism, or would result in
 	// the priority LB policy using a lower priority discovery mechanism when
 	// that becomes available.
+	er.mu.Lock()
 	er.update = &xdsresource.EndpointsUpdate{}
 	er.mu.Unlock()
 
 	er.topLevelResolver.onUpdate(onDone)
 }
 
-func (er *edsDiscoveryMechanism) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (er *edsDiscoveryMechanism) AmbientError(err error, onDone func()) {
 	if er.stopped.HasFired() {
 		onDone()
 		return
 	}
 
-	er.logger.Warningf("EDS discovery mechanism for resource %q reported resource-does-not-exist error", er.nameToWatch)
-
-	// Report an empty update that would result in no priority child being
-	// created for this discovery mechanism. This would result in the priority
-	// LB policy reporting TRANSIENT_FAILURE (as there would be no priorities or
-	// localities) if this was the only discovery mechanism, or would result in
-	// the priority LB policy using a lower priority discovery mechanism when
-	// that becomes available.
-	er.mu.Lock()
-	er.update = &xdsresource.EndpointsUpdate{}
-	er.mu.Unlock()
-
-	er.topLevelResolver.onUpdate(onDone)
+	if er.logger.V(2) {
+		er.logger.Infof("EDS discovery mechanism for resource %q reported ambient error: %v", er.nameToWatch, err)
+	}
 }
diff --git a/xds/internal/balancer/ringhash/config.go b/xds/internal/balancer/ringhash/config.go
index eaa6ca23..044d698a 100644
--- a/xds/internal/balancer/ringhash/config.go
+++ b/xds/internal/balancer/ringhash/config.go
@@ -73,6 +73,7 @@ func parseConfig(c json.RawMessage) (*LBConfig, error) {
 		cfg.RequestHashHeader = ""
 	}
 	if cfg.RequestHashHeader != "" {
+		cfg.RequestHashHeader = strings.ToLower(cfg.RequestHashHeader)
 		// See rules in https://github.com/grpc/proposal/blob/master/A76-ring-hash-improvements.md#explicitly-setting-the-request-hash-key
 		if err := metadata.ValidateKey(cfg.RequestHashHeader); err != nil {
 			return nil, fmt.Errorf("invalid requestHashHeader %q: %v", cfg.RequestHashHeader, err)
diff --git a/xds/internal/resolver/serviceconfig.go b/xds/internal/resolver/serviceconfig.go
index dbefe980..f8c8a965 100644
--- a/xds/internal/resolver/serviceconfig.go
+++ b/xds/internal/resolver/serviceconfig.go
@@ -141,8 +141,8 @@ type erroringConfigSelector struct {
 	err error
 }
 
-func newErroringConfigSelector(xdsNodeID string) *erroringConfigSelector {
-	return &erroringConfigSelector{err: annotateErrorWithNodeID(status.Errorf(codes.Unavailable, "no valid clusters"), xdsNodeID)}
+func newErroringConfigSelector(err error, xdsNodeID string) *erroringConfigSelector {
+	return &erroringConfigSelector{err: annotateErrorWithNodeID(status.Errorf(codes.Unavailable, err.Error()), xdsNodeID)}
 }
 
 func (cs *erroringConfigSelector) SelectConfig(iresolver.RPCInfo) (*iresolver.RPCConfig, error) {
diff --git a/xds/internal/resolver/watch_service.go b/xds/internal/resolver/watch_service.go
index 0de66044..e8d52d0e 100644
--- a/xds/internal/resolver/watch_service.go
+++ b/xds/internal/resolver/watch_service.go
@@ -36,19 +36,19 @@ func newListenerWatcher(resourceName string, parent *xdsResolver) *listenerWatch
 	return lw
 }
 
-func (l *listenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (l *listenerWatcher) ResourceChanged(update *xdsresource.ListenerResourceData, onDone func()) {
 	handleUpdate := func(context.Context) { l.parent.onListenerResourceUpdate(update.Resource); onDone() }
 	l.parent.serializer.ScheduleOr(handleUpdate, onDone)
 }
 
-func (l *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (l *listenerWatcher) ResourceError(err error, onDone func()) {
 	handleError := func(context.Context) { l.parent.onListenerResourceError(err); onDone() }
 	l.parent.serializer.ScheduleOr(handleError, onDone)
 }
 
-func (l *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	handleNotFound := func(context.Context) { l.parent.onListenerResourceNotFound(); onDone() }
-	l.parent.serializer.ScheduleOr(handleNotFound, onDone)
+func (l *listenerWatcher) AmbientError(err error, onDone func()) {
+	handleError := func(context.Context) { l.parent.onListenerResourceAmbientError(err); onDone() }
+	l.parent.serializer.ScheduleOr(handleError, onDone)
 }
 
 func (l *listenerWatcher) stop() {
@@ -68,7 +68,7 @@ func newRouteConfigWatcher(resourceName string, parent *xdsResolver) *routeConfi
 	return rw
 }
 
-func (r *routeConfigWatcher) OnUpdate(u *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (r *routeConfigWatcher) ResourceChanged(u *xdsresource.RouteConfigResourceData, onDone func()) {
 	handleUpdate := func(context.Context) {
 		r.parent.onRouteConfigResourceUpdate(r.resourceName, u.Resource)
 		onDone()
@@ -76,14 +76,14 @@ func (r *routeConfigWatcher) OnUpdate(u *xdsresource.RouteConfigResourceData, on
 	r.parent.serializer.ScheduleOr(handleUpdate, onDone)
 }
 
-func (r *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (r *routeConfigWatcher) ResourceError(err error, onDone func()) {
 	handleError := func(context.Context) { r.parent.onRouteConfigResourceError(r.resourceName, err); onDone() }
 	r.parent.serializer.ScheduleOr(handleError, onDone)
 }
 
-func (r *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	handleNotFound := func(context.Context) { r.parent.onRouteConfigResourceNotFound(r.resourceName); onDone() }
-	r.parent.serializer.ScheduleOr(handleNotFound, onDone)
+func (r *routeConfigWatcher) AmbientError(err error, onDone func()) {
+	handleError := func(context.Context) { r.parent.onRouteConfigResourceAmbientError(r.resourceName, err); onDone() }
+	r.parent.serializer.ScheduleOr(handleError, onDone)
 }
 
 func (r *routeConfigWatcher) stop() {
diff --git a/xds/internal/resolver/xds_resolver.go b/xds/internal/resolver/xds_resolver.go
index 3f9e187c..a66719d0 100644
--- a/xds/internal/resolver/xds_resolver.go
+++ b/xds/internal/resolver/xds_resolver.go
@@ -459,7 +459,10 @@ func (r *xdsResolver) onResolutionComplete() {
 func (r *xdsResolver) applyRouteConfigUpdate(update xdsresource.RouteConfigUpdate) {
 	matchVh := xdsresource.FindBestMatchingVirtualHost(r.dataplaneAuthority, update.VirtualHosts)
 	if matchVh == nil {
-		r.onError(fmt.Errorf("no matching virtual host found for %q", r.dataplaneAuthority))
+		// TODO(purnesh42h): Should this be a resource or ambient error? Note
+		// that its being called only from resource update methods when we have
+		// finished removing the previous update.
+		r.onAmbientError(fmt.Errorf("no matching virtual host found for %q", r.dataplaneAuthority))
 		return
 	}
 	r.currentRouteConfig = update
@@ -469,12 +472,12 @@ func (r *xdsResolver) applyRouteConfigUpdate(update xdsresource.RouteConfigUpdat
 	r.onResolutionComplete()
 }
 
-// onError propagates the error up to the channel. And since this is invoked
-// only for non resource-not-found errors, we don't have to update resolver
+// onAmbientError propagates the error up to the channel. And since this is
+// invoked only for non resource errors, we don't have to update resolver
 // state and we can keep using the old config.
 //
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) onError(err error) {
+func (r *xdsResolver) onAmbientError(err error) {
 	r.cc.ReportError(err)
 }
 
@@ -482,7 +485,7 @@ func (r *xdsResolver) onError(err error) {
 // are removed.
 //
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) onResourceNotFound() {
+func (r *xdsResolver) onResourceError(err error) {
 	// We cannot remove clusters from the service config that have ongoing RPCs.
 	// Instead, what we can do is to send an erroring config selector
 	// along with normal service config. This will ensure that new RPCs will
@@ -491,7 +494,7 @@ func (r *xdsResolver) onResourceNotFound() {
 	// 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.
-	cs := newErroringConfigSelector(r.xdsClient.BootstrapConfig().Node().GetId())
+	cs := newErroringConfigSelector(err, r.xdsClient.BootstrapConfig().Node().GetId())
 	r.sendNewServiceConfig(cs)
 
 	// Stop and dereference the active config selector, if one exists.
@@ -546,16 +549,18 @@ func (r *xdsResolver) onListenerResourceUpdate(update xdsresource.ListenerUpdate
 	r.routeConfigWatcher = newRouteConfigWatcher(r.rdsResourceName, r)
 }
 
-func (r *xdsResolver) onListenerResourceError(err error) {
+func (r *xdsResolver) onListenerResourceAmbientError(err error) {
 	if r.logger.V(2) {
-		r.logger.Infof("Received error for Listener resource %q: %v", r.ldsResourceName, err)
+		r.logger.Infof("Received ambient error for Listener resource %q: %v", r.ldsResourceName, err)
 	}
-	r.onError(err)
+	r.onAmbientError(err)
 }
 
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) onListenerResourceNotFound() {
-	r.logger.Warningf("Received resource-not-found-error for Listener resource %q", r.ldsResourceName)
+func (r *xdsResolver) onListenerResourceError(err error) {
+	if r.logger.V(2) {
+		r.logger.Infof("Received resource error for Listener resource %q: %v", r.ldsResourceName, err)
+	}
 
 	r.listenerUpdateRecvd = false
 	if r.routeConfigWatcher != nil {
@@ -566,7 +571,7 @@ func (r *xdsResolver) onListenerResourceNotFound() {
 	r.routeConfigUpdateRecvd = false
 	r.routeConfigWatcher = nil
 
-	r.onResourceNotFound()
+	r.onResourceError(err)
 }
 
 // Only executed in the context of a serializer callback.
@@ -584,21 +589,23 @@ func (r *xdsResolver) onRouteConfigResourceUpdate(name string, update xdsresourc
 }
 
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) onRouteConfigResourceError(name string, err error) {
+func (r *xdsResolver) onRouteConfigResourceAmbientError(name string, err error) {
 	if r.logger.V(2) {
-		r.logger.Infof("Received error for RouteConfiguration resource %q: %v", name, err)
+		r.logger.Infof("Received ambient error for RouteConfiguration resource %q: %v", name, err)
 	}
-	r.onError(err)
+	r.onAmbientError(err)
 }
 
 // Only executed in the context of a serializer callback.
-func (r *xdsResolver) onRouteConfigResourceNotFound(name string) {
-	r.logger.Warningf("Received resource-not-found-error for RouteConfiguration resource %q", name)
+func (r *xdsResolver) onRouteConfigResourceError(name string, err error) {
+	if r.logger.V(2) {
+		r.logger.Infof("Received resource error for RouteConfiguration resource %q: %v", name, err)
+	}
 
 	if r.rdsResourceName != name {
 		return
 	}
-	r.onResourceNotFound()
+	r.onResourceError(err)
 }
 
 // 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 62d59c48..2c32ace8 100644
--- a/xds/internal/server/listener_wrapper.go
+++ b/xds/internal/server/listener_wrapper.go
@@ -159,36 +159,6 @@ type listenerWrapper struct {
 	rdsHandler *rdsHandler
 }
 
-func (l *listenerWrapper) handleLDSUpdate(update xdsresource.ListenerUpdate) {
-	ilc := update.InboundListenerCfg
-	// Make sure that the socket address on the received Listener resource
-	// matches the address of the net.Listener passed to us by the user. This
-	// check is done here instead of at the XDSClient layer because of the
-	// following couple of reasons:
-	// - XDSClient cannot know the listening address of every listener in the
-	//   system, and hence cannot perform this check.
-	// - this is a very context-dependent check and only the server has the
-	//   appropriate context to perform this check.
-	//
-	// What this means is that the XDSClient has ACKed a resource which can push
-	// the server into a "not serving" mode. This is not ideal, but this is
-	// what we have decided to do.
-	if ilc.Address != l.addr || ilc.Port != l.port {
-		l.mu.Lock()
-		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
-	}
-
-	l.pendingFilterChainManager = ilc.FilterChains
-	l.rdsHandler.updateRouteNamesToWatch(ilc.FilterChains.RouteConfigNames)
-
-	if l.rdsHandler.determineRouteConfigurationReady() {
-		l.maybeUpdateFilterChains()
-	}
-}
-
 // maybeUpdateFilterChains swaps in the pending filter chain manager to the
 // active one if the pending filter chain manager is present. If a swap occurs,
 // it also drains (gracefully stops) any connections that were accepted on the
@@ -408,19 +378,6 @@ 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, l.annotateErrorWithNodeID(err))
-	l.activeFilterChainManager = nil
-	l.pendingFilterChainManager = nil
-	l.rdsHandler.updateRouteNamesToWatch(make(map[string]bool))
-}
-
 // ldsWatcher implements the xdsresource.ListenerWatcher interface and is
 // passed to the WatchListener API.
 type ldsWatcher struct {
@@ -429,7 +386,7 @@ type ldsWatcher struct {
 	name   string
 }
 
-func (lw *ldsWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (lw *ldsWatcher) ResourceChanged(update *xdsresource.ListenerResourceData, onDone func()) {
 	defer onDone()
 	if lw.parent.closed.HasFired() {
 		lw.logger.Warningf("Resource %q received update: %#v after listener was closed", lw.name, update)
@@ -438,30 +395,64 @@ func (lw *ldsWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone
 	if lw.logger.V(2) {
 		lw.logger.Infof("LDS watch for resource %q received update: %#v", lw.name, update.Resource)
 	}
-	lw.parent.handleLDSUpdate(update.Resource)
+	l := lw.parent
+	ilc := update.Resource.InboundListenerCfg
+	// Make sure that the socket address on the received Listener resource
+	// matches the address of the net.Listener passed to us by the user. This
+	// check is done here instead of at the XDSClient layer because of the
+	// following couple of reasons:
+	// - XDSClient cannot know the listening address of every listener in the
+	//   system, and hence cannot perform this check.
+	// - this is a very context-dependent check and only the server has the
+	//   appropriate context to perform this check.
+	//
+	// What this means is that the XDSClient has ACKed a resource which can push
+	// the server into a "not serving" mode. This is not ideal, but this is
+	// what we have decided to do.
+	if ilc.Address != l.addr || ilc.Port != l.port {
+		// TODO(purnesh42h): Are there any other cases where this can be
+		// treated as an ambient error?
+		l.mu.Lock()
+		err := fmt.Errorf("[xDS node id: %v]: %w", l.xdsNodeID, 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
+	}
+
+	l.pendingFilterChainManager = ilc.FilterChains
+	l.rdsHandler.updateRouteNamesToWatch(ilc.FilterChains.RouteConfigNames)
+
+	if l.rdsHandler.determineRouteConfigurationReady() {
+		l.maybeUpdateFilterChains()
+	}
 }
 
-func (lw *ldsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (lw *ldsWatcher) ResourceError(err error, onDone func()) {
 	defer onDone()
 	if lw.parent.closed.HasFired() {
-		lw.logger.Warningf("Resource %q received error: %v after listener was closed", lw.name, err)
+		lw.logger.Warningf("Resource %q received resource error: %v after listener was closed", lw.name, err)
 		return
 	}
 	if lw.logger.V(2) {
-		lw.logger.Infof("LDS watch for resource %q reported error: %v", lw.name, err)
+		lw.logger.Infof("LDS watch for resource %q reported resource error: %v", lw.name, err)
 	}
-	// For errors which are anything other than "resource-not-found", we
-	// continue to use the old configuration.
+
+	l := lw.parent
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.switchModeLocked(connectivity.ServingModeNotServing, err)
+	l.activeFilterChainManager = nil
+	l.pendingFilterChainManager = nil
+	l.rdsHandler.updateRouteNamesToWatch(make(map[string]bool))
 }
 
-func (lw *ldsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (lw *ldsWatcher) AmbientError(err error, onDone func()) {
 	defer onDone()
 	if lw.parent.closed.HasFired() {
-		lw.logger.Warningf("Resource %q received resource-does-not-exist error after listener was closed", lw.name)
+		lw.logger.Warningf("Resource %q received ambient error: %v after listener was closed", lw.name, err)
 		return
 	}
-	lw.logger.Warningf("LDS watch for resource %q reported resource-does-not-exist error", lw.name)
-
-	err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "resource name %q of type Listener not found in received response", lw.name)
-	lw.parent.onLDSResourceDoesNotExist(err)
+	if lw.logger.V(2) {
+		lw.logger.Infof("LDS watch for resource %q reported ambient error: %v", lw.name, err)
+	}
 }
diff --git a/xds/internal/server/rds_handler.go b/xds/internal/server/rds_handler.go
index 29bbf180..4b8eb22d 100644
--- a/xds/internal/server/rds_handler.go
+++ b/xds/internal/server/rds_handler.go
@@ -19,7 +19,6 @@
 package server
 
 import (
-	"fmt"
 	"sync"
 
 	igrpclog "google.golang.org/grpc/internal/grpclog"
@@ -109,21 +108,6 @@ func (rh *rdsHandler) determineRouteConfigurationReady() bool {
 	return len(rh.updates) == len(rh.cancels)
 }
 
-// Must be called from an xDS Client Callback.
-func (rh *rdsHandler) handleRouteUpdate(routeName string, update rdsWatcherUpdate) {
-	rwu := rh.updates[routeName]
-
-	// Accept the new update if any of the following are true:
-	// 1. we had no valid update data.
-	// 2. the update is valid.
-	// 3. the update error is ResourceNotFound.
-	if rwu.data == nil || update.err == nil || xdsresource.ErrType(update.err) == xdsresource.ErrorTypeResourceNotFound {
-		rwu = update
-	}
-	rh.updates[routeName] = rwu
-	rh.callback(routeName, rwu)
-}
-
 // close() is meant to be called by wrapped listener when the wrapped listener
 // is closed, and it cleans up resources by canceling all the active RDS
 // watches.
@@ -151,7 +135,7 @@ type rdsWatcher struct {
 	canceled bool // eats callbacks if true
 }
 
-func (rw *rdsWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (rw *rdsWatcher) ResourceChanged(update *xdsresource.RouteConfigResourceData, onDone func()) {
 	defer onDone()
 	rw.mu.Lock()
 	if rw.canceled {
@@ -162,10 +146,14 @@ func (rw *rdsWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDo
 	if rw.logger.V(2) {
 		rw.logger.Infof("RDS watch for resource %q received update: %#v", rw.routeName, update.Resource)
 	}
-	rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{data: &update.Resource})
+
+	routeName := rw.routeName
+	rwu := rdsWatcherUpdate{data: &update.Resource}
+	rw.parent.updates[routeName] = rwu
+	rw.parent.callback(routeName, rwu)
 }
 
-func (rw *rdsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (rw *rdsWatcher) ResourceError(err error, onDone func()) {
 	defer onDone()
 	rw.mu.Lock()
 	if rw.canceled {
@@ -174,12 +162,16 @@ func (rw *rdsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
 	}
 	rw.mu.Unlock()
 	if rw.logger.V(2) {
-		rw.logger.Infof("RDS watch for resource %q reported error: %v", rw.routeName, err)
+		rw.logger.Infof("RDS watch for resource %q reported resource error", rw.routeName)
 	}
-	rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{err: err})
+
+	routeName := rw.routeName
+	rwu := rdsWatcherUpdate{err: err}
+	rw.parent.updates[routeName] = rwu
+	rw.parent.callback(routeName, rwu)
 }
 
-func (rw *rdsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (rw *rdsWatcher) AmbientError(err error, onDone func()) {
 	defer onDone()
 	rw.mu.Lock()
 	if rw.canceled {
@@ -187,8 +179,10 @@ func (rw *rdsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
 		return
 	}
 	rw.mu.Unlock()
-	rw.logger.Warningf("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})
+	if rw.logger.V(2) {
+		rw.logger.Infof("RDS watch for resource %q reported ambient error: %v", rw.routeName, err)
+	}
+	routeName := rw.routeName
+	rwu := rw.parent.updates[routeName]
+	rw.parent.callback(routeName, rwu)
 }
diff --git a/xds/internal/xdsclient/authority.go b/xds/internal/xdsclient/authority.go
index 05c3e49b..ec3a7352 100644
--- a/xds/internal/xdsclient/authority.go
+++ b/xds/internal/xdsclient/authority.go
@@ -238,9 +238,15 @@ func (a *authority) propagateConnectivityErrorToAllWatchers(err error) {
 	for _, rType := range a.resources {
 		for _, state := range rType {
 			for watcher := range state.watchers {
-				a.watcherCallbackSerializer.TrySchedule(func(context.Context) {
-					watcher.OnError(xdsresource.NewErrorf(xdsresource.ErrorTypeConnection, "xds: error received from xDS stream: %v", err), func() {})
-				})
+				if state.cache == nil {
+					a.watcherCallbackSerializer.TrySchedule(func(context.Context) {
+						watcher.ResourceError(xdsresource.NewErrorf(xdsresource.ErrorTypeConnection, "xds: error received from xDS stream: %v", err), func() {})
+					})
+				} else {
+					a.watcherCallbackSerializer.TrySchedule(func(context.Context) {
+						watcher.AmbientError(xdsresource.NewErrorf(xdsresource.ErrorTypeConnection, "xds: error received from xDS stream: %v", err), func() {})
+					})
+				}
 			}
 		}
 	}
@@ -369,7 +375,11 @@ func (a *authority) handleADSResourceUpdate(serverConfig *bootstrap.ServerConfig
 				watcher := watcher
 				err := uErr.Err
 				watcherCnt.Add(1)
-				funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.OnError(err, done) })
+				if state.cache == nil {
+					funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.ResourceError(err, done) })
+				} else {
+					funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.AmbientError(err, done) })
+				}
 			}
 			continue
 		}
@@ -396,7 +406,7 @@ func (a *authority) handleADSResourceUpdate(serverConfig *bootstrap.ServerConfig
 				watcher := watcher
 				resource := uErr.Resource
 				watcherCnt.Add(1)
-				funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.OnUpdate(resource, done) })
+				funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.ResourceChanged(resource, done) })
 			}
 		}
 
@@ -454,7 +464,7 @@ func (a *authority) handleADSResourceUpdate(serverConfig *bootstrap.ServerConfig
 			// `ignore_resource_deletion` server feature is enabled through the
 			// bootstrap configuration. If the resource deletion is to be
 			// ignored, the resource is not removed from the cache and the
-			// corresponding OnResourceDoesNotExist() callback is not invoked on
+			// corresponding ResourceError() callback is not invoked on
 			// the watchers.
 			if !state.deletionIgnored {
 				state.deletionIgnored = true
@@ -473,7 +483,9 @@ func (a *authority) handleADSResourceUpdate(serverConfig *bootstrap.ServerConfig
 		for watcher := range state.watchers {
 			watcher := watcher
 			watcherCnt.Add(1)
-			funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.OnResourceDoesNotExist(done) })
+			funcsToSchedule = append(funcsToSchedule, func(context.Context) {
+				watcher.ResourceError(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds: resource %q of type %q has been removed", name, rType.TypeName()), done)
+			})
 		}
 	}
 }
@@ -515,7 +527,9 @@ func (a *authority) handleADSResourceDoesNotExist(rType xdsresource.Type, resour
 	state.md = xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusNotExist}
 	for watcher := range state.watchers {
 		watcher := watcher
-		a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnResourceDoesNotExist(func() {}) })
+		a.watcherCallbackSerializer.TrySchedule(func(context.Context) {
+			watcher.ResourceError(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds: resource %q of type %q does not exist", resourceName, rType.TypeName()), func() {})
+		})
 	}
 }
 
@@ -606,7 +620,7 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 
 		xdsChannel, err := a.xdsChannelToUse()
 		if err != nil {
-			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
+			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.ResourceError(err, func() {}) })
 			return
 		}
 
@@ -647,7 +661,7 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 			// xdsClientSerializer callback. Hence making a copy of the cached
 			// resource here for watchCallbackSerializer.
 			resource := state.cache
-			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnUpdate(resource, func() {}) })
+			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.ResourceChanged(resource, func() {}) })
 		}
 		// If last update was NACK'd, notify the new watcher of error
 		// immediately as well.
@@ -659,12 +673,18 @@ func (a *authority) watchResource(rType xdsresource.Type, resourceName string, w
 			// xdsClientSerializer callback. Hence making a copy of the error
 			// here for watchCallbackSerializer.
 			err := state.md.ErrState.Err
-			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
+			if state.cache == nil {
+				a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.ResourceError(err, func() {}) })
+			} else {
+				a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.AmbientError(err, func() {}) })
+			}
 		}
 		// If the metadata field is updated to indicate that the management
 		// server does not have this resource, notify the new watcher.
 		if state.md.Status == xdsresource.ServiceStatusNotExist {
-			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnResourceDoesNotExist(func() {}) })
+			a.watcherCallbackSerializer.TrySchedule(func(context.Context) {
+				watcher.ResourceError(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds: resource %q of type %q does not exist", resourceName, rType.TypeName()), func() {})
+			})
 		}
 		cleanup = a.unwatchResource(rType, resourceName, watcher)
 	}, func() {
diff --git a/xds/internal/xdsclient/clientimpl_watchers.go b/xds/internal/xdsclient/clientimpl_watchers.go
index c080a0b4..2cce17b0 100644
--- a/xds/internal/xdsclient/clientimpl_watchers.go
+++ b/xds/internal/xdsclient/clientimpl_watchers.go
@@ -33,8 +33,12 @@ type wrappingWatcher struct {
 	nodeID string
 }
 
-func (w *wrappingWatcher) OnError(err error, done xdsresource.OnDoneFunc) {
-	w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %w", w.nodeID, err), done)
+func (w *wrappingWatcher) ResourceError(err error, done func()) {
+	w.ResourceWatcher.ResourceError(fmt.Errorf("[xDS node id: %v]: %w", w.nodeID, err), done)
+}
+
+func (w *wrappingWatcher) AmbientError(err error, done func()) {
+	w.ResourceWatcher.AmbientError(fmt.Errorf("[xDS node id: %v]: %w", w.nodeID, err), done)
 }
 
 // WatchResource uses xDS to discover the resource associated with the provided
@@ -60,8 +64,8 @@ func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string,
 	}
 
 	if err := c.resourceTypes.maybeRegister(rType); err != nil {
-		logger.Warningf("Watch registered for name %q of type %q which is already registered", rType.TypeName(), resourceName)
-		c.serializer.TrySchedule(func(context.Context) { watcher.OnError(err, func() {}) })
+		logger.Warningf("Watch registered for type %q, which is already registered", rType.TypeName())
+		c.serializer.TrySchedule(func(context.Context) { watcher.ResourceError(err, func() {}) })
 		return func() {}
 	}
 
@@ -69,7 +73,7 @@ func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string,
 	a := c.getAuthorityForResource(n)
 	if a == nil {
 		logger.Warningf("Watch registered for name %q of type %q, authority %q is not found", rType.TypeName(), resourceName, n.Authority)
-		watcher.OnError(fmt.Errorf("authority %q not found in bootstrap config for resource %q", n.Authority, resourceName), func() {})
+		watcher.ResourceError(fmt.Errorf("authority %q not found in bootstrap config for resource %q", n.Authority, resourceName), func() {})
 		return func() {}
 	}
 	// The watchResource method on the authority is invoked with n.String()
diff --git a/xds/internal/xdsclient/transport/lrs/lrs_stream.go b/xds/internal/xdsclient/transport/lrs/lrs_stream.go
index e33ac694..7260816b 100644
--- a/xds/internal/xdsclient/transport/lrs/lrs_stream.go
+++ b/xds/internal/xdsclient/transport/lrs/lrs_stream.go
@@ -241,11 +241,11 @@ func (lrs *StreamImpl) recvFirstLoadStatsResponse(stream transport.StreamingCall
 		lrs.logger.Infof("Received first LoadStatsResponse: %s", pretty.ToJSON(resp))
 	}
 
-	internal := resp.GetLoadReportingInterval()
-	if internal.CheckValid() != nil {
+	interval := resp.GetLoadReportingInterval()
+	if err := interval.CheckValid(); err != nil {
 		return nil, 0, fmt.Errorf("lrs: invalid load_reporting_interval: %v", err)
 	}
-	loadReportingInterval := internal.AsDuration()
+	loadReportingInterval := interval.AsDuration()
 
 	clusters := resp.Clusters
 	if resp.SendAllClusters {
diff --git a/xds/internal/xdsclient/xdslbregistry/xdslbregistry.go b/xds/internal/xdsclient/xdslbregistry/xdslbregistry.go
index a7782709..88ee9fab 100644
--- a/xds/internal/xdsclient/xdslbregistry/xdslbregistry.go
+++ b/xds/internal/xdsclient/xdslbregistry/xdslbregistry.go
@@ -69,8 +69,6 @@ func ConvertToServiceConfig(lbPolicy *v3clusterpb.LoadBalancingPolicy, depth int
 		converter := m[policy.GetTypedExtensionConfig().GetTypedConfig().GetTypeUrl()]
 		// "Any entry not in the above list is unsupported and will be skipped."
 		// - A52
-		// This includes Least Request as well, since grpc-go does not support
-		// the Least Request Load Balancing Policy.
 		if converter == nil {
 			continue
 		}
diff --git a/xds/internal/xdsclient/xdsresource/cluster_resource_type.go b/xds/internal/xdsclient/xdsresource/cluster_resource_type.go
index 8e9375fc..3d85c31f 100644
--- a/xds/internal/xdsclient/xdsresource/cluster_resource_type.go
+++ b/xds/internal/xdsclient/xdsresource/cluster_resource_type.go
@@ -108,41 +108,40 @@ func (c *ClusterResourceData) Raw() *anypb.Any {
 }
 
 // ClusterWatcher wraps the callbacks to be invoked for different events
-// corresponding to the cluster resource being watched.
+// corresponding to the cluster resource being watched. gRFC A88 contains an
+// exhaustive list of what method is invoked under what conditions.
 type ClusterWatcher interface {
-	// OnUpdate is invoked to report an update for the resource being watched.
-	OnUpdate(*ClusterResourceData, OnDoneFunc)
-
-	// OnError is invoked under different error conditions including but not
-	// limited to the following:
-	//	- authority mentioned in the resource is not found
-	//	- resource name parsing error
-	//	- resource deserialization error
-	//	- resource validation error
-	//	- ADS stream failure
-	//	- connection failure
-	OnError(error, OnDoneFunc)
-
-	// OnResourceDoesNotExist is invoked for a specific error condition where
-	// the requested resource is not found on the xDS management server.
-	OnResourceDoesNotExist(OnDoneFunc)
+	// ResourceChanged indicates a new version of the resource is available.
+	ResourceChanged(resource *ClusterResourceData, 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())
 }
 
 type delegatingClusterWatcher struct {
 	watcher ClusterWatcher
 }
 
-func (d *delegatingClusterWatcher) OnUpdate(data ResourceData, onDone OnDoneFunc) {
+func (d *delegatingClusterWatcher) ResourceChanged(data ResourceData, onDone func()) {
 	c := data.(*ClusterResourceData)
-	d.watcher.OnUpdate(c, onDone)
+	d.watcher.ResourceChanged(c, onDone)
 }
 
-func (d *delegatingClusterWatcher) OnError(err error, onDone OnDoneFunc) {
-	d.watcher.OnError(err, onDone)
+func (d *delegatingClusterWatcher) ResourceError(err error, onDone func()) {
+	d.watcher.ResourceError(err, onDone)
 }
 
-func (d *delegatingClusterWatcher) OnResourceDoesNotExist(onDone OnDoneFunc) {
-	d.watcher.OnResourceDoesNotExist(onDone)
+func (d *delegatingClusterWatcher) AmbientError(err error, onDone func()) {
+	d.watcher.AmbientError(err, onDone)
 }
 
 // WatchCluster uses xDS to discover the configuration associated with the
diff --git a/xds/internal/xdsclient/xdsresource/endpoints_resource_type.go b/xds/internal/xdsclient/xdsresource/endpoints_resource_type.go
index 94c03d0c..de574dd8 100644
--- a/xds/internal/xdsclient/xdsresource/endpoints_resource_type.go
+++ b/xds/internal/xdsclient/xdsresource/endpoints_resource_type.go
@@ -104,41 +104,40 @@ func (e *EndpointsResourceData) Raw() *anypb.Any {
 }
 
 // EndpointsWatcher wraps the callbacks to be invoked for different
-// events corresponding to the endpoints resource being watched.
+// events corresponding to the endpoints resource being watched. gRFC A88
+// contains an exhaustive list of what method is invoked under what conditions.
 type EndpointsWatcher interface {
-	// OnUpdate is invoked to report an update for the resource being watched.
-	OnUpdate(*EndpointsResourceData, OnDoneFunc)
-
-	// OnError is invoked under different error conditions including but not
-	// limited to the following:
-	//	- authority mentioned in the resource is not found
-	//	- resource name parsing error
-	//	- resource deserialization error
-	//	- resource validation error
-	//	- ADS stream failure
-	//	- connection failure
-	OnError(error, OnDoneFunc)
-
-	// OnResourceDoesNotExist is invoked for a specific error condition where
-	// the requested resource is not found on the xDS management server.
-	OnResourceDoesNotExist(OnDoneFunc)
+	// ResourceChanged indicates a new version of the resource is available.
+	ResourceChanged(resource *EndpointsResourceData, 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())
 }
 
 type delegatingEndpointsWatcher struct {
 	watcher EndpointsWatcher
 }
 
-func (d *delegatingEndpointsWatcher) OnUpdate(data ResourceData, onDone OnDoneFunc) {
+func (d *delegatingEndpointsWatcher) ResourceChanged(data ResourceData, onDone func()) {
 	e := data.(*EndpointsResourceData)
-	d.watcher.OnUpdate(e, onDone)
+	d.watcher.ResourceChanged(e, onDone)
 }
 
-func (d *delegatingEndpointsWatcher) OnError(err error, onDone OnDoneFunc) {
-	d.watcher.OnError(err, onDone)
+func (d *delegatingEndpointsWatcher) ResourceError(err error, onDone func()) {
+	d.watcher.ResourceError(err, onDone)
 }
 
-func (d *delegatingEndpointsWatcher) OnResourceDoesNotExist(onDone OnDoneFunc) {
-	d.watcher.OnResourceDoesNotExist(onDone)
+func (d *delegatingEndpointsWatcher) AmbientError(err error, onDone func()) {
+	d.watcher.AmbientError(err, onDone)
 }
 
 // WatchEndpoints uses xDS to discover the configuration associated with the
diff --git a/xds/internal/xdsclient/xdsresource/listener_resource_type.go b/xds/internal/xdsclient/xdsresource/listener_resource_type.go
index e3ca1134..0f49e6c5 100644
--- a/xds/internal/xdsclient/xdsresource/listener_resource_type.go
+++ b/xds/internal/xdsclient/xdsresource/listener_resource_type.go
@@ -141,41 +141,39 @@ func (l *ListenerResourceData) Raw() *anypb.Any {
 }
 
 // ListenerWatcher wraps the callbacks to be invoked for different
-// events corresponding to the listener resource being watched.
+// events corresponding to the listener resource being watched. gRFC A88
+// contains an exhaustive list of what method is invoked under what conditions.
 type ListenerWatcher interface {
-	// OnUpdate is invoked to report an update for the resource being watched.
-	OnUpdate(*ListenerResourceData, OnDoneFunc)
-
-	// OnError is invoked under different error conditions including but not
-	// limited to the following:
-	//	- authority mentioned in the resource is not found
-	//	- resource name parsing error
-	//	- resource deserialization error
-	//	- resource validation error
-	//	- ADS stream failure
-	//	- connection failure
-	OnError(error, OnDoneFunc)
-
-	// OnResourceDoesNotExist is invoked for a specific error condition where
-	// the requested resource is not found on the xDS management server.
-	OnResourceDoesNotExist(OnDoneFunc)
+	// ResourceChanged indicates a new version of the resource is available.
+	ResourceChanged(resource *ListenerResourceData, 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())
 }
 
 type delegatingListenerWatcher struct {
 	watcher ListenerWatcher
 }
 
-func (d *delegatingListenerWatcher) OnUpdate(data ResourceData, onDone OnDoneFunc) {
+func (d *delegatingListenerWatcher) ResourceChanged(data ResourceData, onDone func()) {
 	l := data.(*ListenerResourceData)
-	d.watcher.OnUpdate(l, onDone)
+	d.watcher.ResourceChanged(l, onDone)
 }
-
-func (d *delegatingListenerWatcher) OnError(err error, onDone OnDoneFunc) {
-	d.watcher.OnError(err, onDone)
+func (d *delegatingListenerWatcher) ResourceError(err error, onDone func()) {
+	d.watcher.ResourceError(err, onDone)
 }
 
-func (d *delegatingListenerWatcher) OnResourceDoesNotExist(onDone OnDoneFunc) {
-	d.watcher.OnResourceDoesNotExist(onDone)
+func (d *delegatingListenerWatcher) AmbientError(err error, onDone func()) {
+	d.watcher.AmbientError(err, onDone)
 }
 
 // WatchListener uses xDS to discover the configuration associated with the
diff --git a/xds/internal/xdsclient/xdsresource/resource_type.go b/xds/internal/xdsclient/xdsresource/resource_type.go
index e14f56f7..c22c5a6a 100644
--- a/xds/internal/xdsclient/xdsresource/resource_type.go
+++ b/xds/internal/xdsclient/xdsresource/resource_type.go
@@ -52,33 +52,30 @@ type Producer interface {
 	WatchResource(rType Type, resourceName string, watcher ResourceWatcher) (cancel func())
 }
 
-// OnDoneFunc is a function to be invoked by watcher implementations upon
-// completing the processing of a callback from the xDS client. Failure to
-// invoke this callback prevents the xDS client from reading further messages
-// from the xDS server.
-type OnDoneFunc func()
-
-// ResourceWatcher wraps the callbacks to be invoked for different events
-// corresponding to the resource being watched.
+// 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, registration of those new watchers 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 {
-	// OnUpdate is invoked to report an update for the resource being watched.
-	// The ResourceData parameter needs to be type asserted to the appropriate
-	// type for the resource being watched.
-	OnUpdate(ResourceData, OnDoneFunc)
-
-	// OnError is invoked under different error conditions including but not
-	// limited to the following:
-	//	- authority mentioned in the resource is not found
-	//	- resource name parsing error
-	//	- resource deserialization error
-	//	- resource validation error
-	//	- ADS stream failure
-	//	- connection failure
-	OnError(error, OnDoneFunc)
-
-	// OnResourceDoesNotExist is invoked for a specific error condition where
-	// the requested resource is not found on the xDS management server.
-	OnResourceDoesNotExist(OnDoneFunc)
+	// 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())
 }
 
 // TODO: Once the implementation is complete, rename this interface as
diff --git a/xds/internal/xdsclient/xdsresource/route_config_resource_type.go b/xds/internal/xdsclient/xdsresource/route_config_resource_type.go
index 98ac3132..c292b1b8 100644
--- a/xds/internal/xdsclient/xdsresource/route_config_resource_type.go
+++ b/xds/internal/xdsclient/xdsresource/route_config_resource_type.go
@@ -105,41 +105,41 @@ func (r *RouteConfigResourceData) Raw() *anypb.Any {
 }
 
 // RouteConfigWatcher wraps the callbacks to be invoked for different
-// events corresponding to the route configuration resource being watched.
+// events corresponding to the route configuration resource being watched. gRFC
+// A88 contains an exhaustive list of what method is invoked under what
+// conditions.
 type RouteConfigWatcher interface {
-	// OnUpdate is invoked to report an update for the resource being watched.
-	OnUpdate(*RouteConfigResourceData, OnDoneFunc)
-
-	// OnError is invoked under different error conditions including but not
-	// limited to the following:
-	//	- authority mentioned in the resource is not found
-	//	- resource name parsing error
-	//	- resource deserialization error
-	//	- resource validation error
-	//	- ADS stream failure
-	//	- connection failure
-	OnError(error, OnDoneFunc)
-
-	// OnResourceDoesNotExist is invoked for a specific error condition where
-	// the requested resource is not found on the xDS management server.
-	OnResourceDoesNotExist(OnDoneFunc)
+	// ResourceChanged indicates a new version of the resource is available.
+	ResourceChanged(resource *RouteConfigResourceData, 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())
 }
 
 type delegatingRouteConfigWatcher struct {
 	watcher RouteConfigWatcher
 }
 
-func (d *delegatingRouteConfigWatcher) OnUpdate(data ResourceData, onDone OnDoneFunc) {
+func (d *delegatingRouteConfigWatcher) ResourceChanged(data ResourceData, onDone func()) {
 	rc := data.(*RouteConfigResourceData)
-	d.watcher.OnUpdate(rc, onDone)
+	d.watcher.ResourceChanged(rc, onDone)
 }
 
-func (d *delegatingRouteConfigWatcher) OnError(err error, onDone OnDoneFunc) {
-	d.watcher.OnError(err, onDone)
+func (d *delegatingRouteConfigWatcher) ResourceError(err error, onDone func()) {
+	d.watcher.ResourceError(err, onDone)
 }
 
-func (d *delegatingRouteConfigWatcher) OnResourceDoesNotExist(onDone OnDoneFunc) {
-	d.watcher.OnResourceDoesNotExist(onDone)
+func (d *delegatingRouteConfigWatcher) AmbientError(err error, onDone func()) {
+	d.watcher.AmbientError(err, onDone)
 }
 
 // WatchRouteConfig uses xDS to discover the configuration associated with the



Test patch:
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
index a876fcb0..160783fc 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
@@ -41,6 +41,7 @@ import (
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/pickfirst"
 	"google.golang.org/grpc/internal/testutils/stats"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/status"
@@ -1424,6 +1425,85 @@ func (s) TestPickFirstLeaf_HealthUpdates(t *testing.T) {
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 }
 
+// Tests the case where an address update received by the pick_first LB policy
+// differs in metadata which should be ignored by the LB policy. In this case,
+// the test verifies that new connections are not created when the address
+// update only changes the metadata.
+func (s) TestPickFirstLeaf_AddressUpdateWithMetadata(t *testing.T) {
+	dialer := testutils.NewBlockingDialer()
+	dopts := []grpc.DialOption{
+		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%s":{}}]}`, pickfirstleaf.Name)),
+		grpc.WithContextDialer(dialer.DialContext),
+	}
+	cc, r, backends := setupPickFirstLeaf(t, 2, dopts...)
+
+	// Add a metadata to the addresses before pushing them to the pick_first LB
+	// policy through the manual resolver.
+	addrs := backends.resolverAddrs()
+	for i := range addrs {
+		addrs[i].Metadata = &metadata.MD{
+			"test-metadata-1": []string{fmt.Sprintf("%d", i)},
+		}
+	}
+	r.UpdateState(resolver.State{Addresses: addrs})
+
+	// Ensure that RPCs succeed to the expected backend.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create holds for each backend. This will be used to verify the connection
+	// is not re-established.
+	holds := backends.holds(dialer)
+
+	// Add metadata to the addresses before pushing them to the pick_first LB
+	// policy through the manual resolver. Leave the order of the addresses
+	// unchanged.
+	for i := range addrs {
+		addrs[i].Metadata = &metadata.MD{
+			"test-metadata-2": []string{fmt.Sprintf("%d", i)},
+		}
+	}
+	r.UpdateState(resolver.State{Addresses: addrs})
+
+	// Ensure that no new connection is established.
+	for i := range holds {
+		sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+		defer sCancel()
+		if holds[i].Wait(sCtx) {
+			t.Fatalf("Unexpected connection attempt to backend: %s", addrs[i])
+		}
+	}
+
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
+		t.Fatal(err)
+	}
+
+	// Add metadata to the addresses before pushing them to the pick_first LB
+	// policy through the manual resolver. Reverse of the order of addresses.
+	for i := range addrs {
+		addrs[i].Metadata = &metadata.MD{
+			"test-metadata-3": []string{fmt.Sprintf("%d", i)},
+		}
+	}
+	addrs[0], addrs[1] = addrs[1], addrs[0]
+	r.UpdateState(resolver.State{Addresses: addrs})
+
+	// Ensure that no new connection is established.
+	for i := range holds {
+		sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+		defer sCancel()
+		if holds[i].Wait(sCtx) {
+			t.Fatalf("Unexpected connection attempt to backend: %s", addrs[i])
+		}
+	}
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[1]); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // healthListenerCapturingCCWrapper is used to capture the health listener so
 // that health updates can be mocked for testing.
 type healthListenerCapturingCCWrapper struct {
diff --git a/internal/xds/bootstrap/tlscreds/bundle_ext_test.go b/internal/xds/bootstrap/tlscreds/bundle_ext_test.go
index 7cefffba..4c03fe94 100644
--- a/internal/xds/bootstrap/tlscreds/bundle_ext_test.go
+++ b/internal/xds/bootstrap/tlscreds/bundle_ext_test.go
@@ -58,7 +58,7 @@ func (s) TestValidTlsBuilder(t *testing.T) {
 	caCert := testdata.Path("x509/server_ca_cert.pem")
 	clientCert := testdata.Path("x509/client1_cert.pem")
 	clientKey := testdata.Path("x509/client1_key.pem")
-	clientSpiffeBundle := testdata.Path("spiffe_end2end/client_spiffe.json")
+	clientSpiffeBundle := testdata.Path("spiffe_end2end/client_spiffebundle.json")
 	tests := []struct {
 		name string
 		jd   string
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
index 809ff2c2..3a6c52ba 100644
--- a/stats/opentelemetry/e2e_test.go
+++ b/stats/opentelemetry/e2e_test.go
@@ -204,21 +204,28 @@ func validateTraces(t *testing.T, spans tracetest.SpanStubs, wantSpanInfos []tra
 
 		for _, span := range spans {
 			switch {
-			case span.Name == "grpc.testing.TestService.UnaryCall":
+			case span.Name == "Sent.grpc.testing.TestService.UnaryCall":
 				isUnary = true
 				if span.SpanKind == oteltrace.SpanKindClient {
 					unaryClient = &span
-				} else {
+				}
+			case span.Name == "Recv.grpc.testing.TestService.UnaryCall":
+				isUnary = true
+				if span.SpanKind == oteltrace.SpanKindServer {
 					unaryServer = &span
 				}
 			case span.Name == "Attempt.grpc.testing.TestService.UnaryCall":
 				isUnary = true
 				unaryAttempt = &span
-			case span.Name == "grpc.testing.TestService.FullDuplexCall":
+
+			case span.Name == "Sent.grpc.testing.TestService.FullDuplexCall":
 				isStream = true
 				if span.SpanKind == oteltrace.SpanKindClient {
 					streamClient = &span
-				} else {
+				}
+			case span.Name == "Recv.grpc.testing.TestService.FullDuplexCall":
+				isStream = true
+				if span.SpanKind == oteltrace.SpanKindServer {
 					streamServer = &span
 				}
 			case span.Name == "Attempt.grpc.testing.TestService.FullDuplexCall":
@@ -850,7 +857,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 
 	wantSpanInfos := []traceSpanInfo{
 		{
-			name:     "grpc.testing.TestService.UnaryCall",
+			name:     "Recv.grpc.testing.TestService.UnaryCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -872,7 +879,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -889,7 +896,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 					},
 				},
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -930,7 +937,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -947,7 +954,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 					},
 				},
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -966,13 +973,13 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 			},
 		},
 		{
-			name:       "grpc.testing.TestService.UnaryCall",
+			name:       "Sent.grpc.testing.TestService.UnaryCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
 		},
 		{
-			name:     "grpc.testing.TestService.FullDuplexCall",
+			name:     "Recv.grpc.testing.TestService.FullDuplexCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -995,7 +1002,7 @@ func (s) TestMetricsAndTracesOptionEnabled(t *testing.T) {
 			events: nil,
 		},
 		{
-			name:       "grpc.testing.TestService.FullDuplexCall",
+			name:       "Sent.grpc.testing.TestService.FullDuplexCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
@@ -1074,7 +1081,7 @@ func (s) TestSpan(t *testing.T) {
 
 	wantSpanInfos := []traceSpanInfo{
 		{
-			name:     "grpc.testing.TestService.UnaryCall",
+			name:     "Recv.grpc.testing.TestService.UnaryCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -1096,7 +1103,7 @@ func (s) TestSpan(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1106,14 +1113,10 @@ func (s) TestSpan(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1123,10 +1126,6 @@ func (s) TestSpan(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 			},
@@ -1154,7 +1153,7 @@ func (s) TestSpan(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1164,14 +1163,10 @@ func (s) TestSpan(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1181,22 +1176,18 @@ func (s) TestSpan(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 			},
 		},
 		{
-			name:       "grpc.testing.TestService.UnaryCall",
+			name:       "Sent.grpc.testing.TestService.UnaryCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
 		},
 		{
-			name:     "grpc.testing.TestService.FullDuplexCall",
+			name:     "Recv.grpc.testing.TestService.FullDuplexCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -1219,7 +1210,7 @@ func (s) TestSpan(t *testing.T) {
 			events: nil,
 		},
 		{
-			name:       "grpc.testing.TestService.FullDuplexCall",
+			name:       "Sent.grpc.testing.TestService.FullDuplexCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
@@ -1300,7 +1291,7 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 
 	wantSpanInfos := []traceSpanInfo{
 		{
-			name:     "grpc.testing.TestService.UnaryCall",
+			name:     "Recv.grpc.testing.TestService.UnaryCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -1322,7 +1313,7 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1332,14 +1323,10 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1349,10 +1336,6 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 			},
@@ -1380,7 +1363,7 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 			},
 			events: []trace.Event{
 				{
-					Name: "Outbound compressed message",
+					Name: "Outbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1390,14 +1373,10 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 				{
-					Name: "Inbound compressed message",
+					Name: "Inbound message",
 					Attributes: []attribute.KeyValue{
 						{
 							Key:   "sequence-number",
@@ -1407,22 +1386,18 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 							Key:   "message-size",
 							Value: attribute.IntValue(10006),
 						},
-						{
-							Key:   "message-size-compressed",
-							Value: attribute.IntValue(10006),
-						},
 					},
 				},
 			},
 		},
 		{
-			name:       "grpc.testing.TestService.UnaryCall",
+			name:       "Sent.grpc.testing.TestService.UnaryCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
 		},
 		{
-			name:     "grpc.testing.TestService.FullDuplexCall",
+			name:     "Recv.grpc.testing.TestService.FullDuplexCall",
 			spanKind: oteltrace.SpanKindServer.String(),
 			attributes: []attribute.KeyValue{
 				{
@@ -1445,7 +1420,7 @@ func (s) TestSpan_WithW3CContextPropagator(t *testing.T) {
 			events: nil,
 		},
 		{
-			name:       "grpc.testing.TestService.FullDuplexCall",
+			name:       "Sent.grpc.testing.TestService.FullDuplexCall",
 			spanKind:   oteltrace.SpanKindClient.String(),
 			attributes: nil,
 			events:     nil,
@@ -1601,7 +1576,7 @@ func (s) TestTraceSpan_WithRetriesAndNameResolutionDelay(t *testing.T) {
 				_, err := client.UnaryCall(ctx, &testpb.SimpleRequest{})
 				return err
 			},
-			spanName: "grpc.testing.TestService.UnaryCall",
+			spanName: "Sent.grpc.testing.TestService.UnaryCall",
 		},
 		{
 			name: "streaming",
@@ -1645,7 +1620,7 @@ func (s) TestTraceSpan_WithRetriesAndNameResolutionDelay(t *testing.T) {
 				}
 				return nil
 			},
-			spanName: "grpc.testing.TestService.FullDuplexCall",
+			spanName: "Sent.grpc.testing.TestService.FullDuplexCall",
 		},
 	}
 
diff --git a/xds/csds/csds_e2e_test.go b/xds/csds/csds_e2e_test.go
index cae9d22f..a9654b20 100644
--- a/xds/csds/csds_e2e_test.go
+++ b/xds/csds/csds_e2e_test.go
@@ -71,49 +71,49 @@ func Test(t *testing.T) {
 
 type nopListenerWatcher struct{}
 
-func (nopListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (nopListenerWatcher) ResourceChanged(_ *xdsresource.ListenerResourceData, onDone func()) {
 	onDone()
 }
-func (nopListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (nopListenerWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (nopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (nopListenerWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
 type nopRouteConfigWatcher struct{}
 
-func (nopRouteConfigWatcher) OnUpdate(_ *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (nopRouteConfigWatcher) ResourceChanged(_ *xdsresource.RouteConfigResourceData, onDone func()) {
 	onDone()
 }
-func (nopRouteConfigWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (nopRouteConfigWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (nopRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (nopRouteConfigWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
 type nopClusterWatcher struct{}
 
-func (nopClusterWatcher) OnUpdate(_ *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
+func (nopClusterWatcher) ResourceChanged(_ *xdsresource.ClusterResourceData, onDone func()) {
 	onDone()
 }
-func (nopClusterWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (nopClusterWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (nopClusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (nopClusterWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
 type nopEndpointsWatcher struct{}
 
-func (nopEndpointsWatcher) OnUpdate(_ *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
+func (nopEndpointsWatcher) ResourceChanged(_ *xdsresource.EndpointsResourceData, onDone func()) {
 	onDone()
 }
-func (nopEndpointsWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (nopEndpointsWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (nopEndpointsWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
@@ -127,31 +127,31 @@ func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc)
 // for ADS stream level flow control), and was causing CSDS to not receive any
 // updates from the xDS client.
 type blockingListenerWatcher struct {
-	testCtxDone <-chan struct{}             // Closed when the test is done.
-	onDoneCh    chan xdsresource.OnDoneFunc // Channel to write the onDone callback to.
+	testCtxDone <-chan struct{} // Closed when the test is done.
+	onDoneCh    chan func()     // Channel to write the onDone callback to.
 }
 
 func newBlockingListenerWatcher(testCtxDone <-chan struct{}) *blockingListenerWatcher {
 	return &blockingListenerWatcher{
 		testCtxDone: testCtxDone,
-		onDoneCh:    make(chan xdsresource.OnDoneFunc, 1),
+		onDoneCh:    make(chan func(), 1),
 	}
 }
 
-func (w *blockingListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (w *blockingListenerWatcher) ResourceChanged(_ *xdsresource.ListenerResourceData, onDone func()) {
 	writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
 }
-func (w *blockingListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (w *blockingListenerWatcher) ResourceError(_ error, onDone func()) {
 	writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
 }
-func (w *blockingListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (w *blockingListenerWatcher) AmbientError(_ error, onDone func()) {
 	writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
 }
 
 // writeOnDone attempts to write the onDone callback on the onDone channel. It
 // returns when it can successfully write to the channel or when the test is
 // done, which is signalled by testCtxDone being closed.
-func writeOnDone(testCtxDone <-chan struct{}, onDoneCh chan xdsresource.OnDoneFunc, onDone xdsresource.OnDoneFunc) {
+func writeOnDone(testCtxDone <-chan struct{}, onDoneCh chan func(), onDone func()) {
 	select {
 	case <-testCtxDone:
 	case onDoneCh <- onDone:
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
index f76e9991..78d8e118 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -1003,7 +1003,7 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
 
 	// Ensure RPC fails with Unavailable status code and the error message is
 	// meaningful and contains the xDS node ID.
-	wantErr := fmt.Sprintf("cluster %q not found", clusterName)
+	wantErr := fmt.Sprintf("resource %q of type %q has been removed", clusterName, "ClusterResource")
 	_, err := client.EmptyCall(ctx, &testpb.Empty{})
 	if err := verifyRPCError(err, codes.Unavailable, wantErr, nodeID); err != nil {
 		t.Fatal(err)
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
index b9eed392..517915dd 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
@@ -279,7 +279,7 @@ func (s) TestErrorFromParentLB_ResourceNotFound(t *testing.T) {
 	}
 
 	// Ensure that RPCs start to fail with expected error.
-	wantErr := fmt.Sprintf("cluster %q not found", clusterName)
+	wantErr := fmt.Sprintf("resource %q of type %q has been removed", clusterName, "ClusterResource")
 	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
 		sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 		defer sCancel()
diff --git a/xds/internal/balancer/ringhash/config_test.go b/xds/internal/balancer/ringhash/config_test.go
index 9588a898..b3772e45 100644
--- a/xds/internal/balancer/ringhash/config_test.go
+++ b/xds/internal/balancer/ringhash/config_test.go
@@ -118,6 +118,16 @@ func (s) TestParseConfig(t *testing.T) {
 				RequestHashHeader: "x-foo",
 			},
 		},
+		{
+			name:                "request metadata key set with uppercase letters",
+			js:                  `{"requestHashHeader": "x-FOO"}`,
+			requestHeaderEnvVar: true,
+			want: &LBConfig{
+				MinRingSize:       defaultMinSize,
+				MaxRingSize:       defaultMaxSize,
+				RequestHashHeader: "x-foo",
+			},
+		},
 		{
 			name:                "invalid request hash header",
 			js:                  `{"requestHashHeader": "!invalid"}`,
diff --git a/xds/internal/resolver/xds_resolver_test.go b/xds/internal/resolver/xds_resolver_test.go
index 00107e58..641fc584 100644
--- a/xds/internal/resolver/xds_resolver_test.go
+++ b/xds/internal/resolver/xds_resolver_test.go
@@ -288,13 +288,12 @@ func (s) TestResolverCloseClosesXDSClient(t *testing.T) {
 	}
 }
 
-// Tests the case where a resource returned by the management server is NACKed
-// by the xDS client, which then returns an update containing an error to the
-// resolver. Verifies that the update is propagated to the ClientConn by the
-// resolver. It also tests the cases where the resolver gets a good update
-// subsequently, and another error after the good update. The test also verifies
-// that these are propagated to the ClientConn.
-func (s) TestResolverBadServiceUpdate(t *testing.T) {
+// Tests the case where a resource, not present in cache, returned by the
+// management server is NACKed by the xDS client, which then returns an update
+// containing a resource error to the resolver. It tests the case where the
+// resolver gets an error update without any previous good update. The test
+// also verifies that these are propagated to the ClientConn.
+func (s) TestResolverBadServiceUpdate_NACKedWithoutCache(t *testing.T) {
 	// Spin up an xDS management server for the test.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -319,12 +318,30 @@ func (s) TestResolverBadServiceUpdate(t *testing.T) {
 	}
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, []*v3listenerpb.Listener{lis}, nil)
 
-	// 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"
-	if err := waitForErrorFromResolver(ctx, errCh, wantErr, nodeID); err != nil {
+	// Build the resolver and expect an error update from it. Since the
+	// resource is not cached, it should be received as resource error.
+	_, errCh, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
+	if err := waitForErrorFromResolver(ctx, errCh, "no RouteSpecifier", nodeID); err != nil {
 		t.Fatal(err)
 	}
+}
+
+// Tests the case where a resource, present in cache, returned by the
+// management server is NACKed by the xDS client, which then returns
+// an update containing an ambient error to the resolver. Verifies that the
+// update is propagated to the ClientConn by the resolver. It tests the
+// case where the resolver gets a good update first, and an error
+// after the good update. The test also verifies that these are propagated to
+// the ClientConn and that RPC succeeds as expected after receiving good update
+// as well as ambient error.
+func (s) TestResolverBadServiceUpdate_NACKedWithCache(t *testing.T) {
+	// Spin up an xDS management server for the test.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	nodeID := uuid.New().String()
+	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
+
+	stateCh, errCh, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Configure good listener and route configuration resources on the
 	// management server.
@@ -333,14 +350,44 @@ func (s) TestResolverBadServiceUpdate(t *testing.T) {
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
 
 	// Expect a good update from the resolver.
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+
+	// "Make an RPC" by invoking the config selector.
+	_, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
+	if err != nil {
+		t.Fatalf("cs.SelectConfig(): %v", err)
+	}
+
+	// Configure a listener resource that is expected to be NACKed because it
+	// does not contain the `RouteSpecifier` field in the HTTPConnectionManager.
+	hcm := testutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
+		HttpFilters: []*v3httppb.HttpFilter{e2e.HTTPFilter("router", &v3routerpb.Router{})},
+	})
+	lis := &v3listenerpb.Listener{
+		Name:        defaultTestServiceName,
+		ApiListener: &v3listenerpb.ApiListener{ApiListener: hcm},
+		FilterChains: []*v3listenerpb.FilterChain{{
+			Name: "filter-chain-name",
+			Filters: []*v3listenerpb.Filter{{
+				Name:       wellknown.HTTPConnectionManager,
+				ConfigType: &v3listenerpb.Filter_TypedConfig{TypedConfig: hcm},
+			}},
+		}},
+	}
 
-	// Configure another bad resource on the management server and expect an
-	// error update from the resolver.
+	// Expect an error update from the resolver. Since the resource is cached,
+	// it should be received as an ambient error.
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, []*v3listenerpb.Listener{lis}, nil)
-	if err := waitForErrorFromResolver(ctx, errCh, wantErr, nodeID); err != nil {
+	if err := waitForErrorFromResolver(ctx, errCh, "no RouteSpecifier", nodeID); err != nil {
 		t.Fatal(err)
 	}
+
+	// "Make an RPC" by invoking the config selector which should succeed by
+	// continuing to use the previously cached resource.
+	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
+	if err != nil {
+		t.Fatalf("cs.SelectConfig(): %v", err)
+	}
 }
 
 // TestResolverGoodServiceUpdate tests the case where the resource returned by
@@ -543,7 +590,7 @@ 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 := verifyResolverError(err, codes.Unavailable, "no valid clusters", nodeID); err != nil {
+	if err := verifyResolverError(err, codes.Unavailable, "has been removed", nodeID); err != nil {
 		t.Fatal(err)
 	}
 
@@ -650,7 +697,7 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 
 	// "Make another RPC" by invoking the config selector.
 	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
-	if err := verifyResolverError(err, codes.Unavailable, "no valid clusters", nodeID); err != nil {
+	if err := verifyResolverError(err, codes.Unavailable, "has been removed", nodeID); err != nil {
 		t.Fatal(err)
 	}
 
@@ -674,7 +721,7 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 	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 {
+		if err := verifyResolverError(err, codes.Unavailable, "has been removed", nodeID); err != nil {
 			t.Fatal(err)
 		}
 	}
diff --git a/xds/internal/testutils/resource_watcher.go b/xds/internal/testutils/resource_watcher.go
index dae72e2a..e8f7830a 100644
--- a/xds/internal/testutils/resource_watcher.go
+++ b/xds/internal/testutils/resource_watcher.go
@@ -29,50 +29,55 @@ import "google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 type TestResourceWatcher struct {
 	// UpdateCh is the channel on which xDS client updates are delivered.
 	UpdateCh chan *xdsresource.ResourceData
-	// ErrorCh is the channel on which errors from the xDS client are delivered.
-	ErrorCh chan error
-	// ResourceDoesNotExistCh is the channel used to indicate calls to OnResourceDoesNotExist
-	ResourceDoesNotExistCh chan struct{}
+	// AmbientErrorCh is the channel on which ambient errors from the xDS
+	// client are delivered.
+	AmbientErrorCh chan error
+	// ResourceErrorCh is the channel on which resource errors from the xDS
+	// client are delivered.
+	ResourceErrorCh chan struct{}
 }
 
-// OnUpdate is invoked by the xDS client to report the latest update on the resource
-// being watched.
-func (w *TestResourceWatcher) OnUpdate(data xdsresource.ResourceData, onDone xdsresource.OnDoneFunc) {
+// ResourceChanged is invoked by the xDS client to report the latest update.
+func (w *TestResourceWatcher) ResourceChanged(data xdsresource.ResourceData, onDone func()) {
 	defer onDone()
 	select {
 	case <-w.UpdateCh:
 	default:
 	}
 	w.UpdateCh <- &data
+
 }
 
-// OnError is invoked by the xDS client to report the latest error.
-func (w *TestResourceWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+// ResourceError is invoked by the xDS client to report the latest error to
+// stop watching the resource.
+func (w *TestResourceWatcher) ResourceError(err error, onDone func()) {
 	defer onDone()
 	select {
-	case <-w.ErrorCh:
+	case <-w.ResourceErrorCh:
+	case <-w.AmbientErrorCh:
 	default:
 	}
-	w.ErrorCh <- err
+	w.AmbientErrorCh <- err
+	w.ResourceErrorCh <- struct{}{}
 }
 
-// OnResourceDoesNotExist is used by the xDS client to report that the resource
-// being watched no longer exists.
-func (w *TestResourceWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+// AmbientError is invoked by the xDS client to report the latest ambient
+// error.
+func (w *TestResourceWatcher) AmbientError(err error, onDone func()) {
 	defer onDone()
 	select {
-	case <-w.ResourceDoesNotExistCh:
+	case <-w.AmbientErrorCh:
 	default:
 	}
-	w.ResourceDoesNotExistCh <- struct{}{}
+	w.AmbientErrorCh <- err
 }
 
 // NewTestResourceWatcher returns a TestResourceWatcher to watch for resources
 // via the xDS client.
 func NewTestResourceWatcher() *TestResourceWatcher {
 	return &TestResourceWatcher{
-		UpdateCh:               make(chan *xdsresource.ResourceData, 1),
-		ErrorCh:                make(chan error, 1),
-		ResourceDoesNotExistCh: make(chan struct{}, 1),
+		UpdateCh:        make(chan *xdsresource.ResourceData, 1),
+		AmbientErrorCh:  make(chan error, 1),
+		ResourceErrorCh: make(chan struct{}, 1),
 	}
 }
diff --git a/xds/internal/xdsclient/metrics_test.go b/xds/internal/xdsclient/metrics_test.go
index fddd42f4..369f7216 100644
--- a/xds/internal/xdsclient/metrics_test.go
+++ b/xds/internal/xdsclient/metrics_test.go
@@ -36,15 +36,15 @@ import (
 
 type noopListenerWatcher struct{}
 
-func (noopListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) ResourceChanged(_ *xdsresource.ListenerResourceData, onDone func()) {
 	onDone()
 }
 
-func (noopListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
 
-func (noopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/ads_stream_flow_control_test.go b/xds/internal/xdsclient/tests/ads_stream_flow_control_test.go
index 037042ba..ec1e3cef 100644
--- a/xds/internal/xdsclient/tests/ads_stream_flow_control_test.go
+++ b/xds/internal/xdsclient/tests/ads_stream_flow_control_test.go
@@ -45,22 +45,22 @@ import (
 // DoneNotifier passed to the callback available to the test, thereby enabling
 // the test to block this watcher for as long as required.
 type blockingListenerWatcher struct {
-	doneNotifierCh chan xdsresource.OnDoneFunc // DoneNotifier passed to the callback.
-	updateCh       chan struct{}               // Written to when an update is received.
-	errorCh        chan struct{}               // Written to when an error is received.
-	notFoundCh     chan struct{}               // Written to when the resource is not found.
+	doneNotifierCh chan func()   // DoneNotifier passed to the callback.
+	updateCh       chan struct{} // Written to when an update is received.
+	ambientErrCh   chan struct{} // Written to when an ambient error is received.
+	resourceErrCh  chan struct{} // Written to when a resource error is received.
 }
 
 func newBLockingListenerWatcher() *blockingListenerWatcher {
 	return &blockingListenerWatcher{
-		doneNotifierCh: make(chan xdsresource.OnDoneFunc, 1),
+		doneNotifierCh: make(chan func(), 1),
 		updateCh:       make(chan struct{}, 1),
-		errorCh:        make(chan struct{}, 1),
-		notFoundCh:     make(chan struct{}, 1),
+		ambientErrCh:   make(chan struct{}, 1),
+		resourceErrCh:  make(chan struct{}, 1),
 	}
 }
 
-func (lw *blockingListenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, done xdsresource.OnDoneFunc) {
+func (lw *blockingListenerWatcher) ResourceChanged(update *xdsresource.ListenerResourceData, done func()) {
 	// Notify receipt of the update.
 	select {
 	case lw.updateCh <- struct{}{}:
@@ -73,10 +73,10 @@ func (lw *blockingListenerWatcher) OnUpdate(update *xdsresource.ListenerResource
 	}
 }
 
-func (lw *blockingListenerWatcher) OnError(err error, done xdsresource.OnDoneFunc) {
+func (lw *blockingListenerWatcher) ResourceError(err error, done func()) {
 	// Notify receipt of an error.
 	select {
-	case lw.errorCh <- struct{}{}:
+	case lw.resourceErrCh <- struct{}{}:
 	default:
 	}
 
@@ -86,10 +86,10 @@ func (lw *blockingListenerWatcher) OnError(err error, done xdsresource.OnDoneFun
 	}
 }
 
-func (lw *blockingListenerWatcher) OnResourceDoesNotExist(done xdsresource.OnDoneFunc) {
-	// Notify receipt of resource not found.
+func (lw *blockingListenerWatcher) AmbientError(err error, done func()) {
+	// Notify receipt of an error.
 	select {
-	case lw.notFoundCh <- struct{}{}:
+	case lw.ambientErrCh <- struct{}{}:
 	default:
 	}
 
@@ -396,7 +396,7 @@ func (s) TestADSFlowControl_ResourceUpdates_MultipleResources(t *testing.T) {
 	// guaranteed. So, we select on both of them and unblock the first watcher
 	// whose callback is invoked.
 	var otherWatcherUpdateCh chan struct{}
-	var otherWatcherDoneCh chan xdsresource.OnDoneFunc
+	var otherWatcherDoneCh chan func()
 	select {
 	case <-watcher1.updateCh:
 		onDone := <-watcher1.doneNotifierCh
@@ -438,9 +438,10 @@ func (s) TestADSFlowControl_ResourceUpdates_MultipleResources(t *testing.T) {
 }
 
 // Test ADS stream flow control with a single resource that is expected to be
-// NACKed by the xDS client and the watcher's OnError() callback is expected to
-// be invoked. Verifies that no further reads are attempted until the error is
-// completely processed by the watcher.
+// NACKed by the xDS client and the watcher's ResourceError() callback is
+// expected to be invoked because resource is not cached. Verifies that no
+// further reads are attempted until the error is completely processed by the
+// watcher.
 func (s) TestADSFlowControl_ResourceErrors(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -490,9 +491,9 @@ func (s) TestADSFlowControl_ResourceErrors(t *testing.T) {
 		t.Fatalf("Timed out waiting for ADS stream to be read from")
 	}
 
-	// Wait for the error to reach the watcher.
+	// Wait for the resource error to reach the watcher.
 	select {
-	case <-watcher.errorCh:
+	case <-watcher.resourceErrCh:
 	case <-ctx.Done():
 		t.Fatalf("Timed out waiting for error to reach watcher")
 	}
@@ -518,7 +519,7 @@ func (s) TestADSFlowControl_ResourceErrors(t *testing.T) {
 }
 
 // Test ADS stream flow control with a single resource that is deleted from the
-// management server and therefore the watcher's OnResourceDoesNotExist()
+// management server and therefore the watcher's ResourceError()
 // callback is expected to be invoked. Verifies that no further reads are
 // attempted until the callback is completely handled by the watcher.
 func (s) TestADSFlowControl_ResourceDoesNotExist(t *testing.T) {
@@ -598,7 +599,7 @@ func (s) TestADSFlowControl_ResourceDoesNotExist(t *testing.T) {
 
 	// Wait for the resource not found callback to be invoked.
 	select {
-	case <-watcher.notFoundCh:
+	case <-watcher.resourceErrCh:
 	case <-ctx.Done():
 		t.Fatalf("Timed out waiting for resource not found callback to be invoked on the watcher")
 	}
diff --git a/xds/internal/xdsclient/tests/authority_test.go b/xds/internal/xdsclient/tests/authority_test.go
index a7382278..187fb4d1 100644
--- a/xds/internal/xdsclient/tests/authority_test.go
+++ b/xds/internal/xdsclient/tests/authority_test.go
@@ -307,7 +307,7 @@ func (s) TestAuthority_Fallback(t *testing.T) {
 	// Ensure that the connectivity error callback is not called.
 	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 	defer sCancel()
-	if v, err := watcher.errCh.Receive(sCtx); err != context.DeadlineExceeded {
+	if v, err := watcher.ambientErrCh.Receive(sCtx); err != context.DeadlineExceeded {
 		t.Fatalf("Error callback on the watcher with error:  %v", v.(error))
 	}
 
@@ -330,7 +330,7 @@ func (s) TestAuthority_Fallback(t *testing.T) {
 	secondaryLis.Close()
 
 	// Ensure that the connectivity error callback is called.
-	if _, err := watcher.errCh.Receive(ctx); err != nil {
+	if _, err := watcher.ambientErrCh.Receive(ctx); err != nil {
 		t.Fatal("Timeout when waiting for error callback on the watcher")
 	}
 }
@@ -338,38 +338,38 @@ func (s) TestAuthority_Fallback(t *testing.T) {
 // TODO: Get rid of the clusterWatcher type in cds_watchers_test.go and use this
 // one instead. Also, rename this to clusterWatcher as part of that refactor.
 type clusterWatcherV2 struct {
-	updateCh           *testutils.Channel // Messages of type xdsresource.ClusterUpdate
-	errCh              *testutils.Channel // Messages of type error
-	resourceNotFoundCh *testutils.Channel // Messages of type error
+	updateCh      *testutils.Channel // Messages of type xdsresource.ClusterUpdate
+	ambientErrCh  *testutils.Channel // Messages of type ambient error
+	resourceErrCh *testutils.Channel // Messages of type resource error
 }
 
 func newClusterWatcherV2() *clusterWatcherV2 {
 	return &clusterWatcherV2{
-		updateCh:           testutils.NewChannel(),
-		errCh:              testutils.NewChannel(),
-		resourceNotFoundCh: testutils.NewChannel(),
+		updateCh:      testutils.NewChannel(),
+		ambientErrCh:  testutils.NewChannel(),
+		resourceErrCh: testutils.NewChannel(),
 	}
 }
 
-func (cw *clusterWatcherV2) OnUpdate(update *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcherV2) ResourceChanged(update *xdsresource.ClusterResourceData, onDone func()) {
 	cw.updateCh.Send(update.Resource)
 	onDone()
 }
 
-func (cw *clusterWatcherV2) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcherV2) AmbientError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
 	// here simplifies tests that want access to the most recently received
 	// error.
-	cw.errCh.Replace(err)
+	cw.ambientErrCh.Replace(err)
 	onDone()
 }
 
-func (cw *clusterWatcherV2) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcherV2) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
 	// here simplifies tests that want access to the most recently received
 	// error.
-	cw.resourceNotFoundCh.Replace(xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response"))
+	cw.resourceErrCh.Replace(err)
 	onDone()
 }
diff --git a/xds/internal/xdsclient/tests/cds_watchers_test.go b/xds/internal/xdsclient/tests/cds_watchers_test.go
index 7cd14a16..82d7a06b 100644
--- a/xds/internal/xdsclient/tests/cds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/cds_watchers_test.go
@@ -44,13 +44,13 @@ import (
 
 type noopClusterWatcher struct{}
 
-func (noopClusterWatcher) OnUpdate(update *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
+func (noopClusterWatcher) ResourceChanged(_ *xdsresource.ClusterResourceData, onDone func()) {
 	onDone()
 }
-func (noopClusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (noopClusterWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (noopClusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (noopClusterWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
@@ -67,22 +67,22 @@ func newClusterWatcher() *clusterWatcher {
 	return &clusterWatcher{updateCh: testutils.NewChannel()}
 }
 
-func (cw *clusterWatcher) OnUpdate(update *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcher) ResourceChanged(update *xdsresource.ClusterResourceData, onDone func()) {
 	cw.updateCh.Send(clusterUpdateErrTuple{update: update.Resource})
 	onDone()
 }
 
-func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (cw *clusterWatcher) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
-	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// here and in AmbientError() simplifies tests which will have
 	// access to the most recently received error.
 	cw.updateCh.Replace(clusterUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	cw.updateCh.Replace(clusterUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response")})
+func (cw *clusterWatcher) AmbientError(err error, onDone func()) {
+	cw.updateCh.Replace(clusterUpdateErrTuple{err: err})
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/eds_watchers_test.go b/xds/internal/xdsclient/tests/eds_watchers_test.go
index fd393991..fcb7de4c 100644
--- a/xds/internal/xdsclient/tests/eds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/eds_watchers_test.go
@@ -53,13 +53,13 @@ const (
 
 type noopEndpointsWatcher struct{}
 
-func (noopEndpointsWatcher) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
+func (noopEndpointsWatcher) ResourceChanged(_ *xdsresource.EndpointsResourceData, onDone func()) {
 	onDone()
 }
-func (noopEndpointsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (noopEndpointsWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (noopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (noopEndpointsWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
@@ -76,22 +76,22 @@ func newEndpointsWatcher() *endpointsWatcher {
 	return &endpointsWatcher{updateCh: testutils.NewChannel()}
 }
 
-func (ew *endpointsWatcher) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
+func (ew *endpointsWatcher) ResourceChanged(update *xdsresource.EndpointsResourceData, onDone func()) {
 	ew.updateCh.Send(endpointsUpdateErrTuple{update: update.Resource})
 	onDone()
 }
 
-func (ew *endpointsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (ew *endpointsWatcher) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
-	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// here and in AmbientError() simplifies tests which will have
 	// access to the most recently received error.
 	ew.updateCh.Replace(endpointsUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (ew *endpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	ew.updateCh.Replace(endpointsUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Endpoints not found in received response")})
+func (ew *endpointsWatcher) AmbientError(err error, onDone func()) {
+	ew.updateCh.Replace(endpointsUpdateErrTuple{err: err})
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/lds_watchers_test.go b/xds/internal/xdsclient/tests/lds_watchers_test.go
index 333fb1e5..f75b572a 100644
--- a/xds/internal/xdsclient/tests/lds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/lds_watchers_test.go
@@ -48,13 +48,13 @@ import (
 
 type noopListenerWatcher struct{}
 
-func (noopListenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) ResourceChanged(_ *xdsresource.ListenerResourceData, onDone func()) {
 	onDone()
 }
-func (noopListenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (noopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (noopListenerWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
@@ -71,22 +71,22 @@ func newListenerWatcher() *listenerWatcher {
 	return &listenerWatcher{updateCh: testutils.NewChannel()}
 }
 
-func (lw *listenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (lw *listenerWatcher) ResourceChanged(update *xdsresource.ListenerResourceData, onDone func()) {
 	lw.updateCh.Send(listenerUpdateErrTuple{update: update.Resource})
 	onDone()
 }
 
-func (lw *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (lw *listenerWatcher) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
-	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// here and in AmbientError() simplifies tests which will have
 	// access to the most recently received error.
 	lw.updateCh.Replace(listenerUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (lw *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	lw.updateCh.Replace(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
+func (lw *listenerWatcher) AmbientError(err error, onDone func()) {
+	lw.updateCh.Replace(listenerUpdateErrTuple{err: err})
 	onDone()
 }
 
@@ -100,18 +100,18 @@ func newListenerWatcherMultiple(size int) *listenerWatcherMultiple {
 	return &listenerWatcherMultiple{updateCh: testutils.NewChannelWithSize(size)}
 }
 
-func (lw *listenerWatcherMultiple) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
+func (lw *listenerWatcherMultiple) ResourceChanged(update *xdsresource.ListenerResourceData, onDone func()) {
 	lw.updateCh.Send(listenerUpdateErrTuple{update: update.Resource})
 	onDone()
 }
 
-func (lw *listenerWatcherMultiple) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (lw *listenerWatcherMultiple) ResourceError(err error, onDone func()) {
 	lw.updateCh.Send(listenerUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (lw *listenerWatcherMultiple) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	lw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
+func (lw *listenerWatcherMultiple) AmbientError(err error, onDone func()) {
+	lw.updateCh.Send(listenerUpdateErrTuple{err: err})
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/misc_watchers_test.go b/xds/internal/xdsclient/tests/misc_watchers_test.go
index a9b8b61c..f448a343 100644
--- a/xds/internal/xdsclient/tests/misc_watchers_test.go
+++ b/xds/internal/xdsclient/tests/misc_watchers_test.go
@@ -72,7 +72,7 @@ func newTestRouteConfigWatcher(client xdsclient.XDSClient, name1, name2 string)
 	}
 }
 
-func (rw *testRouteConfigWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (rw *testRouteConfigWatcher) ResourceChanged(update *xdsresource.RouteConfigResourceData, onDone func()) {
 	rw.updateCh.Send(routeConfigUpdateErrTuple{update: update.Resource})
 
 	rw.cancel1 = xdsresource.WatchRouteConfig(rw.client, rw.name1, rw.rcw1)
@@ -80,17 +80,17 @@ func (rw *testRouteConfigWatcher) OnUpdate(update *xdsresource.RouteConfigResour
 	onDone()
 }
 
-func (rw *testRouteConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (rw *testRouteConfigWatcher) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
-	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// here and in AmbientError() simplifies tests which will have
 	// access to the most recently received error.
 	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (rw *testRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
+func (rw *testRouteConfigWatcher) AmbientError(err error, onDone func()) {
+	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: err})
 	onDone()
 }
 
@@ -322,7 +322,7 @@ func (s) TestNodeProtoSentOnlyInFirstRequest(t *testing.T) {
 	select {
 	case <-ctx.Done():
 		t.Fatal("Timeout when waiting for the connection error to be propagated to the watcher")
-	case <-watcher.ErrorCh:
+	case <-watcher.AmbientErrorCh:
 	}
 
 	// Restart the management server.
@@ -427,9 +427,9 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
 		case <-sCtx.Done():
 		case <-watcher.UpdateCh:
 			t.Fatal("Unexpected resource update")
-		case <-watcher.ErrorCh:
+		case <-watcher.AmbientErrorCh:
 			t.Fatal("Unexpected resource error")
-		case <-watcher.ResourceDoesNotExistCh:
+		case <-watcher.ResourceErrorCh:
 			t.Fatal("Unexpected resource does not exist")
 		}
 
@@ -437,7 +437,7 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
 		select {
 		case <-ctx.Done():
 			t.Fatal("Timeout when waiting for error callback to be invoked")
-		case err := <-watcher.ErrorCh:
+		case err := <-watcher.AmbientErrorCh:
 			if err == nil || !strings.Contains(err.Error(), nodeID) {
 				t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
 			}
@@ -452,7 +452,7 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
 		select {
 		case <-ctx.Done():
 			t.Fatal("Timeout when waiting for error callback to be invoked")
-		case err := <-watcher.ErrorCh:
+		case err := <-watcher.AmbientErrorCh:
 			if err == nil || !strings.Contains(err.Error(), nodeID) {
 				t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
 			}
@@ -500,7 +500,7 @@ func (s) TestWatchErrorsContainNodeID_ChannelCreationFailure(t *testing.T) {
 	select {
 	case <-ctx.Done():
 		t.Fatal("Timeout when waiting for error callback to be invoked")
-	case err := <-watcher.ErrorCh:
+	case err := <-watcher.AmbientErrorCh:
 		if err == nil || !strings.Contains(err.Error(), nodeID) {
 			t.Fatalf("Unexpected error: %v, want error with node ID: %q", err, nodeID)
 		}
diff --git a/xds/internal/xdsclient/tests/rds_watchers_test.go b/xds/internal/xdsclient/tests/rds_watchers_test.go
index 6209e333..2086c018 100644
--- a/xds/internal/xdsclient/tests/rds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/rds_watchers_test.go
@@ -43,13 +43,13 @@ import (
 
 type noopRouteConfigWatcher struct{}
 
-func (noopRouteConfigWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (noopRouteConfigWatcher) ResourceChanged(_ *xdsresource.RouteConfigResourceData, onDone func()) {
 	onDone()
 }
-func (noopRouteConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (noopRouteConfigWatcher) ResourceError(_ error, onDone func()) {
 	onDone()
 }
-func (noopRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
+func (noopRouteConfigWatcher) AmbientError(_ error, onDone func()) {
 	onDone()
 }
 
@@ -66,22 +66,22 @@ func newRouteConfigWatcher() *routeConfigWatcher {
 	return &routeConfigWatcher{updateCh: testutils.NewChannel()}
 }
 
-func (rw *routeConfigWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
+func (rw *routeConfigWatcher) ResourceChanged(update *xdsresource.RouteConfigResourceData, onDone func()) {
 	rw.updateCh.Send(routeConfigUpdateErrTuple{update: update.Resource})
 	onDone()
 }
 
-func (rw *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
+func (rw *routeConfigWatcher) ResourceError(err error, onDone func()) {
 	// When used with a go-control-plane management server that continuously
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
-	// here and in OnResourceDoesNotExist() simplifies tests which will have
+	// here and in AmbientError() simplifies tests which will have
 	// access to the most recently received error.
 	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: err})
 	onDone()
 }
 
-func (rw *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
+func (rw *routeConfigWatcher) AmbientError(err error, onDone func()) {
+	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: err})
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/resource_update_test.go b/xds/internal/xdsclient/tests/resource_update_test.go
index faea52e6..a9fce90c 100644
--- a/xds/internal/xdsclient/tests/resource_update_test.go
+++ b/xds/internal/xdsclient/tests/resource_update_test.go
@@ -161,7 +161,7 @@ func (s) TestHandleListenerResponseFromManagementServer(t *testing.T) {
 					Value:   []byte{1, 2, 3, 4},
 				}},
 			},
-			wantErr: "Listener not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ListenerResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.listener.v3.Listener",
@@ -177,7 +177,7 @@ func (s) TestHandleListenerResponseFromManagementServer(t *testing.T) {
 				TypeUrl:     "type.googleapis.com/envoy.config.listener.v3.Listener",
 				VersionInfo: "1",
 			},
-			wantErr: "Listener not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ListenerResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.listener.v3.Listener",
@@ -194,7 +194,7 @@ func (s) TestHandleListenerResponseFromManagementServer(t *testing.T) {
 				VersionInfo: "1",
 				Resources:   []*anypb.Any{testutils.MarshalAny(t, &v3routepb.RouteConfiguration{})},
 			},
-			wantErr: "Listener not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ListenerResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.listener.v3.Listener",
@@ -422,7 +422,7 @@ func (s) TestHandleRouteConfigResponseFromManagementServer(t *testing.T) {
 					Value:   []byte{1, 2, 3, 4},
 				}},
 			},
-			wantErr: "RouteConfiguration not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "RouteConfigResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
@@ -438,7 +438,7 @@ func (s) TestHandleRouteConfigResponseFromManagementServer(t *testing.T) {
 				TypeUrl:     "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
 				VersionInfo: "1",
 			},
-			wantErr: "RouteConfiguration not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "RouteConfigResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
@@ -455,7 +455,7 @@ func (s) TestHandleRouteConfigResponseFromManagementServer(t *testing.T) {
 				VersionInfo: "1",
 				Resources:   []*anypb.Any{testutils.MarshalAny(t, &v3clusterpb.Cluster{})},
 			},
-			wantErr: "RouteConfiguration not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "RouteConfigResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
@@ -675,7 +675,7 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
 					Value:   []byte{1, 2, 3, 4},
 				}},
 			},
-			wantErr: "Cluster not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ClusterResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.cluster.v3.Cluster",
@@ -691,7 +691,7 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
 				TypeUrl:     "type.googleapis.com/envoy.config.cluster.v3.Cluster",
 				VersionInfo: "1",
 			},
-			wantErr: "Cluster not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ClusterResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.cluster.v3.Cluster",
@@ -708,7 +708,7 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
 				VersionInfo: "1",
 				Resources:   []*anypb.Any{testutils.MarshalAny(t, &v3endpointpb.ClusterLoadAssignment{})},
 			},
-			wantErr: "Cluster not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "ClusterResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.cluster.v3.Cluster",
@@ -986,7 +986,7 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) {
 					Value:   []byte{1, 2, 3, 4},
 				}},
 			},
-			wantErr: "Endpoints not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "EndpointsResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
@@ -1002,7 +1002,7 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) {
 				TypeUrl:     "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
 				VersionInfo: "1",
 			},
-			wantErr: "Endpoints not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "EndpointsResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
@@ -1019,7 +1019,7 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) {
 				VersionInfo: "1",
 				Resources:   []*anypb.Any{testutils.MarshalAny(t, &v3listenerpb.Listener{})},
 			},
-			wantErr: "Endpoints not found in received response",
+			wantErr: fmt.Sprintf("xds: resource %q of type %q does not exist", resourceName1, "EndpointsResource"),
 			wantGenericXDSConfig: []*v3statuspb.ClientConfig_GenericXdsConfig{
 				{
 					TypeUrl:      "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
diff --git a/xds/internal/xdsclient/xdslbregistry/xdslbregistry_test.go b/xds/internal/xdsclient/xdslbregistry/xdslbregistry_test.go
index d3a7bef0..edb455b9 100644
--- a/xds/internal/xdsclient/xdslbregistry/xdslbregistry_test.go
+++ b/xds/internal/xdsclient/xdslbregistry/xdslbregistry_test.go
@@ -45,6 +45,7 @@ import (
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3leastrequestpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/least_request/v3"
+	v3maglevpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/maglev/v3"
 	v3pickfirstpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/pick_first/v3"
 	v3ringhashpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/ring_hash/v3"
 	v3roundrobinpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/round_robin/v3"
@@ -382,7 +383,7 @@ func (s) TestConvertToServiceConfigFailure(t *testing.T) {
 					{
 						TypedExtensionConfig: &v3corepb.TypedExtensionConfig{
 							// Not supported by gRPC-Go.
-							TypedConfig: testutils.MarshalAny(t, &v3leastrequestpb.LeastRequest{}),
+							TypedConfig: testutils.MarshalAny(t, &v3maglevpb.Maglev{}),
 						},
 					},
 				},
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go b/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
index ad393bd6..33c70ecf 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
@@ -107,7 +107,7 @@ func (s) TestValidateCluster_Failure(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "non-round-robin-or-ring-hash-lb-policy",
+			name: "not-supported-lb-policy",
 			cluster: &v3clusterpb.Cluster{
 				ClusterDiscoveryType: &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_EDS},
 				EdsClusterConfig: &v3clusterpb.Cluster_EdsClusterConfig{
@@ -117,7 +117,7 @@ func (s) TestValidateCluster_Failure(t *testing.T) {
 						},
 					},
 				},
-				LbPolicy: v3clusterpb.Cluster_LEAST_REQUEST,
+				LbPolicy: v3clusterpb.Cluster_MAGLEV,
 			},
 			wantErr: true,
 		},
diff --git a/xds/server_ext_test.go b/xds/server_ext_test.go
index 80f33a0b..81ead009 100644
--- a/xds/server_ext_test.go
+++ b/xds/server_ext_test.go
@@ -84,8 +84,11 @@ type servingModeChangeHandler struct {
 	logger interface {
 		Logf(format string, args ...any)
 	}
-	modeCh chan connectivity.ServingMode
-	errCh  chan error
+	currentMode connectivity.ServingMode
+	currentErr  error
+	mu          sync.Mutex
+	modeCh      chan connectivity.ServingMode
+	errCh       chan error
 }
 
 func newServingModeChangeHandler(t *testing.T) *servingModeChangeHandler {
@@ -97,11 +100,24 @@ func newServingModeChangeHandler(t *testing.T) *servingModeChangeHandler {
 }
 
 func (m *servingModeChangeHandler) modeChangeCallback(addr net.Addr, args xds.ServingModeChangeArgs) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	// Suppress pushing duplicate mode change and error if the mode is staying
+	// in NOT_SERVING and the error is the same.
+	//
+	// TODO(purnesh42h): Should we move this check to listener wrapper? This
+	// shouldn't happen in practice a lot. But we never know what kind of
+	// management servers users run.
+	if m.currentMode == args.Mode && m.currentMode == connectivity.ServingModeNotServing && m.currentErr.Error() == args.Err.Error() {
+		return
+	}
 	m.logger.Logf("Serving mode for listener %q changed to %q, err: %v", addr.String(), args.Mode, args.Err)
 	m.modeCh <- args.Mode
+	m.currentMode = args.Mode
 	if args.Err != nil {
 		m.errCh <- args.Err
 	}
+	m.currentErr = args.Err
 }
 
 // createStubServer creates a new xDS-enabled gRPC server and returns a
diff --git a/xds/server_security_ext_test.go b/xds/server_security_ext_test.go
index 28685b02..992d8ddd 100644
--- a/xds/server_security_ext_test.go
+++ b/xds/server_security_ext_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/google/uuid"
 	"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/testutils"
@@ -128,7 +129,8 @@ func (s) TestServer_Security_NoCertProvidersInBootstrap_Success(t *testing.T) {
 // client is expected to NACK this resource because the certificate provider
 // 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.
+// if the xDS-enabled gRPC server receives resource error causing mode change,
+// it does not enter "serving" mode.
 func (s) TestServer_Security_NoCertificateProvidersInBootstrap_Failure(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -214,14 +216,16 @@ func (s) TestServer_Security_NoCertificateProvidersInBootstrap_Failure(t *testin
 		t.Fatal("Timeout when waiting for an NACK from the xDS client for the LDS response")
 	}
 
-	// Wait a short duration and ensure that the server does not enter "serving"
-	// mode.
+	// Wait a short duration and ensure that if the server receive mode change
+	// it does not enter "serving" mode.
 	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")
+	case stateCh := <-modeChangeHandler.modeCh:
+		if stateCh == connectivity.ServingModeServing {
+			t.Fatal("Server entered serving mode before the route config was received")
+		}
 	}
 
 	// Create a client that uses insecure creds and verify that RPCs don't
@@ -245,7 +249,8 @@ func (s) TestServer_Security_NoCertificateProvidersInBootstrap_Failure(t *testin
 //     certificate provider instance
 //
 // 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.
+// second listener receive a resource error which cause the server mode change
+// but never moves to "serving" mode.
 func (s) TestServer_Security_WithValidAndInvalidSecurityConfiguration(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -448,12 +453,14 @@ func (s) TestServer_Security_WithValidAndInvalidSecurityConfiguration(t *testing
 		t.Fatal("Timeout when waiting for an NACK from the xDS client for the LDS response")
 	}
 
-	// Wait a short duration and ensure that the server does not enter "serving"
-	// mode.
+	// Wait a short duration and ensure that if the server receives mode change
+	// it does not enter "serving" mode.
 	select {
 	case <-time.After(2 * defaultTestShortTimeout):
 	case <-modeChangeHandler2.modeCh:
-		t.Fatal("Server changed to serving mode when not expected to")
+		if modeChangeHandler2.currentMode == connectivity.ServingModeServing {
+			t.Fatal("Server changed to serving mode when not expected to")
+		}
 	}
 
 	// Create a client that uses insecure creds and verify that RPCs don't
diff --git a/xds/server_test.go b/xds/server_test.go
index 51579333..205f2646 100644
--- a/xds/server_test.go
+++ b/xds/server_test.go
@@ -655,15 +655,13 @@ func (s) TestHandleListenerUpdate_ErrorUpdate(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Also make sure that no serving mode updates are received. The serving
-	// mode does not change until the server comes to the conclusion that the
-	// requested resource is not present in the management server. This happens
-	// when the watch timer expires or when the resource is explicitly deleted
-	// by the management server.
+	// Also make sure that serving mode updates are received. The serving
+	// mode changes to NOT_SERVING. This happens because watcher received a
+	// resource error for the invalid resource from the server.
 	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
 	defer sCancel()
-	if _, err := modeChangeCh.Receive(sCtx); err != context.DeadlineExceeded {
-		t.Fatal("Serving mode changed received when none expected")
+	if _, err := modeChangeCh.Receive(sCtx); err == context.DeadlineExceeded {
+		t.Fatal("Serving mode did not change when expected to change")
 	}
 }
 
@@ -697,10 +695,23 @@ func (s) TestServeAndCloseDoNotRace(t *testing.T) {
 	// Generate bootstrap contents up front for all servers.
 	bootstrapContents := generateBootstrapContents(t, uuid.NewString(), nonExistentManagementServer)
 
+	// Override the default ServingModeCallback with a noop function because the
+	// serverURI is invalid which will result in xDS channel creation failure
+	// while registering the watch for listener resource. This will trigger
+	// resource error notifications for the invalid listener resource leading
+	// to service mode change to "not serving" each time.
+	//
+	// Even if the the server is currently NOT_SERVING, in the case (where we
+	// are NOT_SERVING and the new mode is also NOT_SERVING), the update is not
+	// suppressed as:
+	//   1. the error may have change
+	//   2. it provides a timestamp of the last backoff attempt
+	noopModeChangeCallback := func(_ net.Addr, _ ServingModeChangeArgs) {}
+
 	wg := sync.WaitGroup{}
 	wg.Add(200)
 	for i := 0; i < 100; i++ {
-		server, err := NewGRPCServer(BootstrapContentsForTesting(bootstrapContents))
+		server, err := NewGRPCServer(BootstrapContentsForTesting(bootstrapContents), ServingModeCallback(noopModeChangeCallback))
 		if err != nil {
 			t.Fatalf("Failed to create an xDS enabled gRPC server: %v", err)
 		}