Repository: grpc/grpc-go

Issue:
Flaky test RingHash_RequestHashKeyConnecting and RingHash_RequestHashKeyRandom
On master, `TestRingHash_RequestHashKeyRandom` and `TestRingHash_RequestHashKeyConnecting` occasionally fail with randomness assertions being out of tolerance:

```
--- FAIL: Test (13.38s)
    --- FAIL: Test/RingHash_RequestHashKeyRandom (1.31s)
        tlogger.go:116: INFO server.go:690 [core] [Server #479]Server created  (t=+209.593µs)
        stubserver.go:300: Started test service backend at "127.0.0.1:39929"
        tlogger.go:116: INFO server.go:690 [core] [Server #480]Server created  (t=+412.302µs)
        stubserver.go:300: Started test service backend at "127.0.0.1:39907"
        tlogger.go:116: INFO server.go:886 [core] [Server #479 ListenSocket #481]ListenSocket created  (t=+590.035µs)
        tlogger.go:116: INFO server.go:886 [core] [Server #480 ListenSocket #482]ListenSocket created  (t=+653.674µs)
        tlogger.go:116: INFO server.go:690 [core] [Server #483]Server created  (t=+721.251µs)
        stubserver.go:300: Started test service backend at "127.0.0.1:40283"
        tlogger.go:116: INFO server.go:886 [core] [Server #483 ListenSocket #484]ListenSocket created  (t=+844.863µs)
        tlogger.go:116: INFO server.go:690 [core] [Server #485]Server created  (t=+998.701µs)
        stubserver.go:300: Started test service backend at "127.0.0.1:35865"
        tlogger.go:116: INFO clientconn.go:1687 [core] original dial target is: "whatever:///test.server"  (t=+1.064985ms)
        tlogger.go:116: INFO clientconn.go:333 [core] [Channel #486]Channel created  (t=+1.159131ms)
        tlogger.go:116: INFO clientconn.go:207 [core] [Channel #486]parsed dial target is: resolver.Target{URL:url.URL{Scheme:"whatever", Opaque:"", User:(*url.Userinfo)(nil), Host:"", Path:"/test.server", RawPath:"", OmitHost:false, ForceQuery:false, RawQuery:"", Fragment:"", RawFragment:""}}  (t=+1.21165ms)
        tlogger.go:116: INFO clientconn.go:208 [core] [Channel #486]Channel authority set to "test.server"  (t=+1.241035ms)
        tlogger.go:116: INFO resolver_wrapper.go:210 [core] [Channel #486]Resolver state updated: {
              "Addresses": null,
              "Endpoints": [
                {
                  "Addresses": [
                    {
                      "Addr": "127.0.0.1:39929",
                      "ServerName": "",
                      "Attributes": null,
                      "BalancerAttributes": null,
                      "Metadata": null
                    }
                  ],
                  "Attributes": null
                },
                {
                  "Addresses": [
                    {
                      "Addr": "127.0.0.1:39907",
                      "ServerName": "",
                      "Attributes": null,
                      "BalancerAttributes": null,
                      "Metadata": null
                    }
                  ],
                  "Attributes": null
                },
                {
                  "Addresses": [
                    {
                      "Addr": "127.0.0.1:40283",
                      "ServerName": "",
                      "Attributes": null,
                      "BalancerAttributes": null,
                      "Metadata": null
                    }
                  ],
                  "Attributes": null
                },
                {
                  "Addresses": [
                    {
                      "Addr": "127.0.0.1:35865",
                      "ServerName": "",
                      "Attributes": null,
                      "BalancerAttributes": null,
                      "Metadata": null
                    }
                  ],
                  "Attributes": null
                }
              ],
              "ServiceConfig": null,
              "Attributes": null
            } ()  (t=+1.460025ms)
        tlogger.go:116: INFO balancer_wrapper.go:122 [core] [Channel #486]Channel switches to new LB policy "ring_hash_experimental"  (t=+1.739018ms)
        tlogger.go:116: INFO ringhash.go:65 [xds] [ring-hash-lb 0xc000dacaa0] Created  (t=+1.779744ms)
        tlogger.go:116: INFO clientconn.go:333 [core] [Channel #486]Channel exiting idle mode  (t=+3.846074ms)
        tlogger.go:116: INFO server.go:886 [core] [Server #485 ListenSocket #487]ListenSocket created  (t=+4.057881ms)
        tlogger.go:116: INFO clientconn.go:563 [core] [Channel #486]Channel Connectivity change to CONNECTING  (t=+4.771919ms)
        tlogger.go:116: INFO clientconn.go:866 [core] [Channel #486 SubChannel #488]Subchannel created  (t=+4.861477ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #488]Subchannel Connectivity change to CONNECTING  (t=+5.020745ms)
        tlogger.go:116: INFO clientconn.go:1352 [core] [Channel #486 SubChannel #488]Subchannel picks a new address "127.0.0.1:39907" to connect  (t=+5.063185ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #488]Subchannel Connectivity change to READY  (t=+5.885786ms)
        tlogger.go:116: INFO clientconn.go:563 [core] [Channel #486]Channel Connectivity change to READY  (t=+6.012803ms)
        tlogger.go:116: INFO clientconn.go:866 [core] [Channel #486 SubChannel #491]Subchannel created  (t=+6.189284ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #491]Subchannel Connectivity change to CONNECTING  (t=+6.249387ms)
        tlogger.go:116: INFO clientconn.go:1352 [core] [Channel #486 SubChannel #491]Subchannel picks a new address "127.0.0.1:40283" to connect  (t=+6.291596ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #491]Subchannel Connectivity change to READY  (t=+7.392037ms)
        tlogger.go:116: INFO clientconn.go:866 [core] [Channel #486 SubChannel #494]Subchannel created  (t=+8.935378ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #494]Subchannel Connectivity change to CONNECTING  (t=+8.994719ms)
        tlogger.go:116: INFO clientconn.go:1352 [core] [Channel #486 SubChannel #494]Subchannel picks a new address "127.0.0.1:39929" to connect  (t=+9.033913ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #494]Subchannel Connectivity change to READY  (t=+9.707534ms)
        tlogger.go:116: INFO clientconn.go:866 [core] [Channel #486 SubChannel #497]Subchannel created  (t=+12.064019ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #497]Subchannel Connectivity change to CONNECTING  (t=+12.11826ms)
        tlogger.go:116: INFO clientconn.go:1352 [core] [Channel #486 SubChannel #497]Subchannel picks a new address "127.0.0.1:35865" to connect  (t=+12.155931ms)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #497]Subchannel Connectivity change to READY  (t=+12.721721ms)
        ringhash_balancer_test.go:2783: Fraction of RPCs to backend 127.0.0.1:39907: got 0.30469565217391303, want 0.25 (margin: +-0.05)
        tlogger.go:116: INFO clientconn.go:563 [core] [Channel #486]Channel Connectivity change to SHUTDOWN  (t=+1.253461573s)
        tlogger.go:116: INFO resolver_wrapper.go:111 [core] [Channel #486]Closing the name resolver  (t=+1.253520283s)
        tlogger.go:116: INFO balancer_wrapper.go:160 [core] [Channel #486]ccBalancerWrapper: closing  (t=+1.25359367s)
        tlogger.go:116: INFO ringhash.go:300 [xds] [ring-hash-lb 0xc000dacaa0] Shutdown  (t=+1.253660916s)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #494]Subchannel Connectivity change to SHUTDOWN  (t=+1.253863936s)
        tlogger.go:116: INFO clientconn.go:1560 [core] [Channel #486 SubChannel #494]Subchannel deleted  (t=+1.253931794s)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #488]Subchannel Connectivity change to SHUTDOWN  (t=+1.254014138s)
        tlogger.go:116: INFO clientconn.go:1560 [core] [Channel #486 SubChannel #488]Subchannel deleted  (t=+1.254075012s)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #491]Subchannel Connectivity change to SHUTDOWN  (t=+1.254265288s)
        tlogger.go:116: INFO clientconn.go:1560 [core] [Channel #486 SubChannel #491]Subchannel deleted  (t=+1.25433[229](https://github.com/grpc/grpc-go/actions/runs/14410175562/job/40416016501?pr=8243#step:8:230)4s)
        tlogger.go:116: INFO clientconn.go:1233 [core] [Channel #486 SubChannel #497]Subchannel Connectivity change to SHUTDOWN  (t=+1.254402175s)
        tlogger.go:116: INFO clientconn.go:1560 [core] [Channel #486 SubChannel #497]Subchannel deleted  (t=+1.254460344s)
        tlogger.go:116: INFO clientconn.go:333 [core] [Channel #486]Channel deleted  (t=+1.25476809s)
        tlogger.go:116: INFO server.go:822 [core] [Server #485 ListenSocket #487]ListenSocket deleted  (t=+1.254905828s)
        tlogger.go:116: INFO server.go:822 [core] [Server #483 ListenSocket #484]ListenSocket deleted  (t=+1.255077019s)
        tlogger.go:116: INFO server.go:822 [core] [Server #480 ListenSocket #482]ListenSocket deleted  (t=+1.255[234](https://github.com/grpc/grpc-go/actions/runs/14410175562/job/40416016501?pr=8243#step:8:235)734s)
        tlogger.go:116: INFO server.go:822 [core] [Server #479 ListenSocket #481]ListenSocket deleted  (t=+1.255384936s)
FAIL
FAIL	google.golang.org/grpc/xds/internal/balancer/ringhash/e2e	26.426s
```

We need to figure out whether:
1. the number of RPC is too low to accurately assert the behavior
2. the distribution is not actually random.

Hints from issue comments:

Solution patch:
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/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/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/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/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)
 		}