Change getOrCreateControlPlaneClient to throw an exception rather tha…

…n return a null to better match the name.

Do some cleanup

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

@@ -42,6 +42,7 @@
 import io.grpc.xds.client.Bootstrapper.AuthorityInfo;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
+import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -146,6 +147,7 @@ public void run() {
               lrsClient.stopLoadReporting();
             }
             cleanUpResourceTimers(null);
+            activatedCpClients.clear();
           }
         });
   }
@@ -199,19 +201,6 @@ public Collection<String> getSubscribedResources(
     return retVal.isEmpty() ? null : retVal;
   }
 
-  private boolean hasSubscribers(String authority) {
-    for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> map
-        : resourceSubscribers.values()) {
-      for (ResourceSubscriber<? extends ResourceUpdate> subscriber : map.values()) {
-        if (subscriber != null && Objects.equals(subscriber.authority, authority)) {
-          return true;
-        }
-      }
-    }
-
-    return false;
-  }
-
   // As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
   // ResourceTypes that do not have subscribers does not show up in the snapshot keys.
   @Override
@@ -286,16 +275,15 @@ public void run() {
   /**
    * Gets a ControlPlaneClient for the subscriber's authority, creating one if necessary.
    * If there already was an active CPC for this authority, and it is different from the one
-   * identified, then do fallback to it.
+   * identified, then do fallback to the identified one (cpcToUse).
    *
-   * @return identified CPC or {@code null} if there are no valid ServerInfos associated with the
-   *     subscriber's authority
+   * @return identified CPC or {@code null} (if there are no valid ServerInfos associated with the
+   *     subscriber's authority or CPC's for all are in backoff), and whether did a fallback.
    */
-  @Nullable
   private <T extends ResourceUpdate> CpcWithFallbackState manageControlPlaneClient(
       ResourceSubscriber<T> subscriber) {
 
-    ControlPlaneClient cpcToUse = null;
+    ControlPlaneClient cpcToUse;
     boolean didFallback = false;
     try {
       cpcToUse = getOrCreateControlPlaneClient(subscriber.authority);
@@ -304,27 +292,28 @@ private <T extends ResourceUpdate> CpcWithFallbackState manageControlPlaneClient
 
       subscriber.onError(
           Status.INVALID_ARGUMENT.withDescription(subscriber.errorDescription), null);
+      return new CpcWithFallbackState(null, false);
+    } catch (IOException e) {
+      logger.log(XdsLogLevel.WARNING,
+          "Could not create a control plane client for authority {0}",
+          subscriber.authority);
+      return new CpcWithFallbackState(null, false);
     }
 
-    if (cpcToUse != null) {
-      if (!cpcToUse.isConnected()) {
-        cpcToUse.connect(); // Make sure we are at least trying to connect
-      }
-
-      ControlPlaneClient activeCpClient = getActiveCpc(subscriber.authority);
-      if (cpcToUse != activeCpClient) {
-        if (activeCpClient != null) {
-          didFallback = fallBackToCpc(cpcToUse, subscriber.authority, activeCpClient);
-        }
+    if (!cpcToUse.isConnected()) {
+      cpcToUse.connect(); // Make sure we are at least trying to connect
+    }
 
-        addCpcToAuthority(subscriber.authority, cpcToUse); // makes it active
+    ControlPlaneClient activeCpClient = getActiveCpc(subscriber.authority);
+    if (cpcToUse != activeCpClient) {
+      if (activeCpClient != null) {
+        didFallback = fallBackToCpc(cpcToUse, subscriber.authority, activeCpClient);
       }
-      cpcToUse.adjustResourceSubscription(subscriber.type);
-    } else {
-      logger.log(XdsLogLevel.WARNING,
-          "Could not create a control plane client for authority {0}",
-          subscriber.authority);
+
+      addCpcToAuthority(subscriber.authority, cpcToUse); // makes it active
     }
+    cpcToUse.adjustResourceSubscription(subscriber.type);
+
     return new CpcWithFallbackState(cpcToUse, didFallback);
   }
 
@@ -454,7 +443,7 @@ private void cleanUpResourceTimers(ControlPlaneClient cpcForThisStream) {
     }
   }
 
-  private ControlPlaneClient getOrCreateControlPlaneClient(String authority) {
+  private ControlPlaneClient getOrCreateControlPlaneClient(String authority) throws IOException {
     // Optimize for the common case of a working ads stream already exists for the authority
     ControlPlaneClient activeCpc = getActiveCpc(authority);
     if (activeCpc != null && !activeCpc.isInBackoff()) {
@@ -463,7 +452,7 @@ private ControlPlaneClient getOrCreateControlPlaneClient(String authority) {
 
     ImmutableList<ServerInfo> serverInfos = getServerInfos(authority);
     if (serverInfos == null) {
-      return null;
+      throw new IllegalArgumentException("No xds servers found for authority " + authority);
     }
 
     for (ServerInfo serverInfo : serverInfos) {
@@ -474,8 +463,8 @@ private ControlPlaneClient getOrCreateControlPlaneClient(String authority) {
       return cpc;
     }
 
-    // Everything existed and is in backoff so return null
-    return null;
+    // Everything existed and is in backoff so throw
+    throw new IOException("All xds transports for authority " + authority + " are in backoff");
   }
 
   private ControlPlaneClient getOrCreateControlPlaneClient(ServerInfo serverInfo) {
@@ -546,21 +535,6 @@ private ImmutableList<ServerInfo> getServerInfos(String authority) {
     }
   }
 
-  private List<String> getAuthoritiesForServerInfo(ServerInfo serverInfo) {
-    List<String> authorities = new ArrayList<>();
-    for (Map.Entry<String, AuthorityInfo> entry : bootstrapInfo.authorities().entrySet()) {
-      if (entry.getValue().xdsServers().contains(serverInfo)) {
-        authorities.add(entry.getKey());
-      }
-    }
-
-    if (bootstrapInfo.servers().contains(serverInfo)) {
-      authorities.add(null);
-    }
-
-    return authorities;
-  }
-
   @SuppressWarnings("unchecked")
   private <T extends ResourceUpdate> void handleResourceUpdate(
       XdsResourceType.Args args, List<Any> resources, XdsResourceType<T> xdsResourceType,