xds:fix cancel xds watcher accidentally removes the url (v1.52 backport) (#9810)

Fix a bug. When any of the subscribers for a resource has the last watcher cancelled, the bug will accidentally remove that resource type from the map, which make xds stream not accepting response update for that resource type entirely (pass through, no ACK/NACK will send).

Author: YifeiZhuang

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

@@ -328,13 +328,13 @@ public void run() {
         if (!subscriber.isWatched()) {
           subscriber.cancelResourceWatch();
           resourceSubscribers.get(type).remove(resourceName);
-          subscribedResourceTypeUrls.remove(type.typeUrl());
-          subscribedResourceTypeUrls.remove(type.typeUrlV2());
           if (subscriber.xdsChannel != null) {
             subscriber.xdsChannel.adjustResourceSubscription(type);
           }
           if (resourceSubscribers.get(type).isEmpty()) {
             resourceSubscribers.remove(type);
+            subscribedResourceTypeUrls.remove(type.typeUrl());
+            subscribedResourceTypeUrls.remove(type.typeUrlV2());
           }
         }
       }

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

@@ -921,6 +921,35 @@ public void ldsResourceUpdated() {
     assertThat(channelForEmptyAuthority).isNull();
   }
 
+  @Test
+  public void cancelResourceWatcherNotRemoveUrlSubscribers() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsListenerResource.getInstance(), LDS_RESOURCE,
+        ldsResourceWatcher);
+    verifyResourceMetadataRequested(LDS, LDS_RESOURCE);
+
+    // Initial LDS response.
+    call.sendResponse(LDS, testListenerVhosts, VERSION_1, "0000");
+    call.verifyRequest(LDS, LDS_RESOURCE, VERSION_1, "0000", NODE);
+    verify(ldsResourceWatcher).onChanged(ldsUpdateCaptor.capture());
+    verifyGoldenListenerVhosts(ldsUpdateCaptor.getValue());
+    verifyResourceMetadataAcked(LDS, LDS_RESOURCE, testListenerVhosts, VERSION_1, TIME_INCREMENT);
+
+    xdsClient.watchXdsResource(XdsListenerResource.getInstance(),
+        LDS_RESOURCE + "1", ldsResourceWatcher);
+    xdsClient.cancelXdsResourceWatch(XdsListenerResource.getInstance(), LDS_RESOURCE + "1",
+        ldsResourceWatcher);
+
+    // Updated LDS response.
+    Any testListenerVhosts2 = Any.pack(mf.buildListenerWithApiListener(LDS_RESOURCE,
+        mf.buildRouteConfiguration("new", mf.buildOpaqueVirtualHosts(VHOST_SIZE))));
+    call.sendResponse(LDS, testListenerVhosts2, VERSION_2, "0001");
+    call.verifyRequest(LDS, LDS_RESOURCE, VERSION_2, "0001", NODE);
+    verify(ldsResourceWatcher).onChanged(ldsUpdateCaptor.capture());
+    verifyGoldenListenerVhosts(ldsUpdateCaptor.getValue());
+    verifyResourceMetadataAcked(LDS, LDS_RESOURCE, testListenerVhosts2, VERSION_2,
+        TIME_INCREMENT * 2);
+  }
+
   @Test
   public void ldsResourceUpdated_withXdstpResourceName() {
     BootstrapperImpl.enableFederation = true;