xds:Cleanup to reduce test flakiness

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

@@ -149,6 +149,7 @@
       throwIfParentContextsNotEmpty(watcher);
     }
 
+    watcher.cancelled = true;
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
@@ -597,6 +598,8 @@
       implements ResourceWatcher<T> {
     private final XdsResourceType<T> type;
     private final String resourceName;
+    boolean cancelled;
+
     @Nullable
     private StatusOr<T> data;
 
@@ -693,6 +696,10 @@
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
+
       handleDoesNotExist(resourceName);
       xdsConfigWatcher.onResourceDoesNotExist(toContextString());
     }
@@ -752,6 +759,9 @@
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
       handleDoesNotExist(checkNotNull(resourceName, "resourceName"));
       xdsConfigWatcher.onResourceDoesNotExist(toContextString());
     }
@@ -836,6 +846,9 @@
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
       handleDoesNotExist(checkNotNull(resourceName, "resourceName"));
       maybePublishConfig();
     }
@@ -857,6 +870,9 @@
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
       handleDoesNotExist(checkNotNull(resourceName, "resourceName"));
       maybePublishConfig();
     }

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

@@ -88,6 +88,7 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatcher;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Captor;
 import org.mockito.InOrder;
@@ -696,9 +697,9 @@ public void testChangeAggCluster() {
     controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, edsMap);
 
     // Verify that the config is updated as expected
-    inOrder.verify(xdsConfigWatcher, timeout(1000)).onUpdate(xdsConfigCaptor.capture());
-    XdsConfig config = xdsConfigCaptor.getValue();
-    assertThat(config.getClusters().keySet()).containsExactly("root", "clusterA21", "clusterA22");
+    ClusterNameMatcher nameMatcher
+        = new ClusterNameMatcher(Arrays.asList("root", "clusterA21", "clusterA22"));
+    inOrder.verify(xdsConfigWatcher, timeout(1000)).onUpdate(argThat(nameMatcher));
   }
 
   private Listener buildInlineClientListener(String rdsName, String clusterName) {
@@ -789,4 +790,21 @@ void deliverLdsUpdate(long httpMaxStreamDurationNano,
       });
     }
   }
+
+  static class ClusterNameMatcher implements ArgumentMatcher<XdsConfig> {
+    private final List<String> expectedNames;
+
+    ClusterNameMatcher(List<String> expectedNames) {
+      this.expectedNames = expectedNames;
+    }
+
+    @Override
+    public boolean matches(XdsConfig xdsConfig) {
+      if (xdsConfig == null || xdsConfig.getClusters() == null) {
+        return false;
+      }
+      return xdsConfig.getClusters().size() == expectedNames.size()
+          && xdsConfig.getClusters().keySet().containsAll(expectedNames);
+    }
+  }
 }