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.

Author: ejona86

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

@@ -93,13 +93,15 @@ public static String toContextStr(String typeName, String resourceName) {
 
   @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 @@ private void releaseSubscription(ClusterSubscription subscription) {
     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 @@ private void cancelClusterWatcherTree(CdsWatcher root, Object parentContext) {
    */
   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 @@ StatusOr<XdsConfig> buildUpdate() {
       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 @@ public interface XdsConfigWatcher {
     void onUpdate(StatusOr<XdsConfig> config);
   }
 
-  private class ClusterSubscription implements Closeable {
-    String clusterName;
+  private final class ClusterSubscription implements Closeable {
+    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 void edsUpdateAfterShutdown() {
     });
   }
 
+  @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);
   }