xds: Improve shutdown handling of XdsDepManager

The most important change here is to handle subscribeToCluster() calls
after shutdown(), and preventing the internal state from being heavily
confused as the assumption is there are no watchers after shutdown().

ClusterSubscription.closed isn't strictly necessary, but I don't want
the code to depend on double-deregistration being safe.
maybePublishConfig() isn't being called after shutdown(), but adding the
protection avoids a class of bugs that would cause channel panic.

xds/src/main/java/io/grpc/xds/XdsDependencyManager.java

@@ -93,13 +93,15 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
 
   @Override
   public Closeable subscribeToCluster(String clusterName) {
-
     checkNotNull(clusterName, "clusterName");
     ClusterSubscription subscription = new ClusterSubscription(clusterName);
 
     syncContext.execute(() -> {
+      if (getWatchers(XdsListenerResource.getInstance()).isEmpty()) {
+        subscription.closed = true;
+        return; // shutdown() called
+      }
       addClusterWatcher(clusterName, subscription, 1);
-      maybePublishConfig();
     });
 
     return subscription;
@@ -207,10 +209,14 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
     checkNotNull(subscription, "subscription");
     String clusterName = subscription.getClusterName();
     syncContext.execute(() -> {
+      if (subscription.closed) {
+        return;
+      }
+      subscription.closed = true;
       XdsWatcherBase<?> cdsWatcher =
           resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
       if (cdsWatcher == null) {
-        return; // already released while waiting for the syncContext
+        return; // shutdown() called
       }
       cancelClusterWatcherTree((CdsWatcher) cdsWatcher, subscription);
       maybePublishConfig();
@@ -257,6 +263,9 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
    */
   private void maybePublishConfig() {
     syncContext.throwIfNotInThisSynchronizationContext();
+    if (getWatchers(XdsListenerResource.getInstance()).isEmpty()) {
+      return; // shutdown() called
+    }
     boolean waitingOnResource = resourceWatchers.values().stream()
         .flatMap(typeWatchers -> typeWatchers.watchers.values().stream())
         .anyMatch(XdsWatcherBase::missingResult);
@@ -293,6 +302,11 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
       routeSource = ((LdsWatcher) ldsWatcher).getRouteSource();
     }
 
+    if (routeSource == null) {
+      return StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
+          "Bug: No route source found for listener " + dataPlaneAuthority));
+    }
+
     StatusOr<RdsUpdate> statusOrRdsUpdate = routeSource.getRdsUpdate();
     if (!statusOrRdsUpdate.hasValue()) {
       return StatusOr.fromStatus(statusOrRdsUpdate.getStatus());
@@ -557,14 +571,15 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
     void onUpdate(StatusOr<XdsConfig> config);
   }
 
-  private class ClusterSubscription implements Closeable {
+  private final class ClusterSubscription implements Closeable {
-    String clusterName;
+    private final String clusterName;
+    boolean closed; // Accessed from syncContext
 
     public ClusterSubscription(String clusterName) {
-      this.clusterName = clusterName;
+      this.clusterName = checkNotNull(clusterName, "clusterName");
     }
 
-    public String getClusterName() {
+    String getClusterName() {
       return clusterName;
     }
 

xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java

@@ -848,6 +848,22 @@ public class XdsDependencyManagerTest {
     });
   }
 
+  @Test
+  public void subscribeToClusterAfterShutdown() throws Exception {
+    XdsTestUtils.setAdsConfig(controlPlaneService, serverName, "RDS", "CDS", "EDS",
+        ENDPOINT_HOSTNAME, ENDPOINT_PORT);
+
+    InOrder inOrder = Mockito.inOrder(xdsConfigWatcher);
+    xdsDependencyManager = new XdsDependencyManager(xdsClient, xdsConfigWatcher, syncContext,
+        serverName, serverName, nameResolverArgs, scheduler);
+    inOrder.verify(xdsConfigWatcher).onUpdate(any());
+    xdsDependencyManager.shutdown();
+
+    Closeable subscription = xdsDependencyManager.subscribeToCluster("CDS");
+    inOrder.verify(xdsConfigWatcher, never()).onUpdate(any());
+    subscription.close();
+  }
+
   private Listener buildInlineClientListener(String rdsName, String clusterName) {
     return XdsTestUtils.buildInlineClientListener(rdsName, clusterName, serverName);
   }