address comments

Author: shivaspeaks

xds/src/main/java/io/grpc/xds/client/XdsClient.java

@@ -140,6 +140,13 @@ public interface ResourceUpdate {}
 
   /**
    * Watcher interface for a single requested xDS resource.
+   *
+   * <p>Note that we expect that the implementer to:
+   * - Comply with the guarantee to not generate certain statuses by the library:
+   *   https://grpc.github.io/grpc/core/md_doc_statuscodes.html. If the code needs to be
+   *   propagated to the channel, override it with {@link io.grpc.Status.Code#UNAVAILABLE}.
+   * - Keep {@link Status} description in one form or another, as it contains valuable debugging
+   *   information.
    */
   @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10862")
   public interface ResourceWatcher<T extends ResourceUpdate> {

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -293,7 +293,6 @@ public void run() {
   private <T extends ResourceUpdate> CpcWithFallbackState manageControlPlaneClient(
       ResourceSubscriber<T> subscriber) {
 
-    ControlPlaneClient activeCpc = getActiveCpc(subscriber.authority);
     ControlPlaneClient cpcToUse;
     boolean didFallback = false;
     try {
@@ -313,16 +312,17 @@ private <T extends ResourceUpdate> CpcWithFallbackState manageControlPlaneClient
       return new CpcWithFallbackState(null, false);
     }
 
-    if (cpcToUse != activeCpc) {
+    ControlPlaneClient activeCpClient = getActiveCpc(subscriber.authority);
+    if (cpcToUse != activeCpClient) {
       addCpcToAuthority(subscriber.authority, cpcToUse);
-      if (activeCpc != null) {
+      if (activeCpClient != null) {
         didFallback = cpcToUse != null && !cpcToUse.isInError();
         if (didFallback) {
           logger.log(XdsLogLevel.INFO, "Falling back to XDS server {0}",
               cpcToUse.getServerInfo().target());
         } else {
           logger.log(XdsLogLevel.WARNING, "No working fallback XDS Servers found from {0}",
-              activeCpc.getServerInfo().target());
+              activeCpClient.getServerInfo().target());
         }
       }
     }
@@ -847,7 +847,7 @@ void onData(ParsedResource<T> parsedResource, String version, long updateTime,
           processingTracker.startTask();
           executor.execute(() -> {
             try {
-              watcher.onResourceChanged(update); // Call the new method
+              watcher.onResourceChanged(update);
             } finally {
               processingTracker.onComplete();
             }
@@ -1033,10 +1033,6 @@ public void handleStreamClosed(Status status, boolean shouldTryFallback) {
           if (!authoritiesForClosedCpc.contains(subscriber.authority)) {
             continue;
           }
-          if (subscriber.hasResult()) {
-            subscriber.onError(status, null); // This will become an onAmbientError
-            continue;
-          }
 
           // try to fallback to lower priority control plane client
           if (shouldTryFallback && manageControlPlaneClient(subscriber).didFallback) {