xds:Cleanup to reduce test flakiness (grpc#11895)

* don't process resourceDoesNotExist for watchers that have been cancelled.

* Change test to use an ArgumentMatcher instead of expecting that only the final result will be sent since depending on timing there may be configs sent for clusters being removed with their entries as errors.

Author: larry-safran

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

@@ -149,6 +149,7 @@ private <T extends ResourceUpdate> void cancelWatcher(XdsWatcherBase<T> watcher)
       throwIfParentContextsNotEmpty(watcher);
     }
 
+    watcher.cancelled = true;
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
@@ -597,6 +598,8 @@ private abstract static class XdsWatcherBase<T extends ResourceUpdate>
       implements ResourceWatcher<T> {
     private final XdsResourceType<T> type;
     private final String resourceName;
+    boolean cancelled;
+
     @Nullable
     private StatusOr<T> data;
 
@@ -693,6 +696,10 @@ public void onError(Status error) {
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
+
       handleDoesNotExist(resourceName);
       xdsConfigWatcher.onResourceDoesNotExist(toContextString());
     }
@@ -752,6 +759,9 @@ public void onError(Status error) {
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
       handleDoesNotExist(checkNotNull(resourceName, "resourceName"));
       xdsConfigWatcher.onResourceDoesNotExist(toContextString());
     }
@@ -836,6 +846,9 @@ public void onChanged(XdsClusterResource.CdsUpdate update) {
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
+      if (cancelled) {
+        return;
+      }
       handleDoesNotExist(checkNotNull(resourceName, "resourceName"));
       maybePublishConfig();
     }
@@ -857,6 +870,9 @@ public void onChanged(XdsEndpointResource.EdsUpdate update) {
 
     @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);
+    }
+  }
 }