xds:Cleanup to reduce test flakiness (#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.

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

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