Add more checking for fallback being enabled

Author: larry-safran

xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java

@@ -16,7 +16,6 @@
 
 package io.grpc.xds.client;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
@@ -444,7 +443,8 @@ private void handleRpcStreamClosed(Status status) {
                 ? "ADS stream closed with status {0}: {1}. Cause: {2}"
                 : "ADS stream failed with status {0}: {1}. Cause: {2}";
         logger.log(
-            XdsLogLevel.ERROR, errorMsg, status.getCode(), status.getDescription(), status.getCause());
+            XdsLogLevel.ERROR, errorMsg, status.getCode(), status.getDescription(),
+            status.getCause());
       }
       closed = true;
       xdsResponseHandler.handleStreamClosed(status);

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

@@ -642,7 +642,7 @@ private boolean fallBackToCpc(ControlPlaneClient fallbackCpc, String authority,
   }
 
   private void doFallbackIfNecessary(ControlPlaneClient cpc, String authority) {
-    if (cpc == null) {
+    if (cpc == null || !BootstrapperImpl.isEnabledXdsFallback()) {
       return;
     }
 
@@ -668,17 +668,14 @@ private void setCpcForAuthority(String authority, ControlPlaneClient cpc) {
         continue;
       }
       for (ResourceSubscriber<?> subscriber : subscriberTypeEntry.getValue().values()) {
-        if (!Objects.equals(subscriber.authority, authority)) {
+        if (subscriber.controlPlaneClient == cpc
+            || !Objects.equals(subscriber.authority, authority)) {
           continue;
         }
-        if (subscriber.controlPlaneClient != cpc) {
-          subscriber.data = null;
-          subscriber.absent = false;
-          subscriber.controlPlaneClient = cpc;
-          subscriber.restartTimer();
-        } else {
-          subscriber.controlPlaneClient = cpc;
-        }
+        subscriber.data = null;
+        subscriber.absent = false;
+        subscriber.controlPlaneClient = cpc;
+        subscriber.restartTimer();
       }
     }
 
@@ -1090,6 +1087,11 @@ && compareCpClients(activeCpClient, controlPlaneClient, authority) < 0) {
       restartMatchingSubscriberTimers(controlPlaneClient, authority);
       controlPlaneClient.sendDiscoveryRequests(authority);
 
+      // If fallback isn't enabled, we're done.
+      if (!BootstrapperImpl.isEnabledXdsFallback()) {
+        return;
+      }
+
       // Shutdown any lower priority control plane clients.
       Iterator<ControlPlaneClient> iterator = serverCpClientMap.values().iterator();
       while (iterator.hasNext()) {

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

@@ -87,7 +87,7 @@ public class ControlPlaneRule extends TestWatcher {
   private XdsTestControlPlaneService controlPlaneService;
   private XdsTestLoadReportingService loadReportingService;
   private XdsNameResolverProvider nameResolverProvider;
-  private int port;
+  private final int port;
 
   public ControlPlaneRule() {
     this(0);
@@ -219,7 +219,7 @@ void setCdsConfig(Cluster cluster) {
     setCdsConfig(CLUSTER_NAME, cluster);
   }
 
-    void setCdsConfig(String clusterName, Cluster cluster) {
+  void setCdsConfig(String clusterName, Cluster cluster) {
     getService().setXdsConfig(ADS_TYPE_URL_CDS,
         ImmutableMap.<String, Message>of(clusterName, cluster));
   }
@@ -316,6 +316,7 @@ static Listener buildClientListener(String name) {
   static Listener buildClientListener(String name, String identifier) {
     return buildClientListener(name, identifier, RDS_NAME);
   }
+
   static Listener buildClientListener(String name, String identifier, String rdsName) {
     HttpFilter httpFilter = HttpFilter.newBuilder()
         .setName(identifier)

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

@@ -103,8 +103,26 @@ public void onResourceDoesNotExist(String resourceName) {
   @Mock
   private XdsClient.ResourceWatcher<XdsRouteConfigureResource.RdsUpdate> rdsWatcher2;
 
-  @Mock
-  private XdsClient.ResourceWatcher<XdsClusterResource.CdsUpdate> cdsWatcher;
+  private XdsClient.ResourceWatcher<XdsClusterResource.CdsUpdate> raalCdsWatcher =
+      new XdsClient.ResourceWatcher<XdsClusterResource.CdsUpdate>() {
+
+        @Override
+        public void onChanged(XdsClusterResource.CdsUpdate update) {
+          log.log(Level.FINE, "CDS update: " + update);
+        }
+
+        @Override
+        public void onError(Status error) {
+          log.log(Level.FINE, "CDS update error: " + error.getDescription());
+        }
+
+        @Override
+        public void onResourceDoesNotExist(String resourceName) {
+          log.log(Level.FINE, "CDS resource does not exist: " + resourceName);
+        }
+      };
+  private XdsClient.ResourceWatcher<XdsClusterResource.CdsUpdate> cdsWatcher =
+      mock(XdsClient.ResourceWatcher.class, delegatesTo(raalCdsWatcher));
   @Mock
   private XdsClient.ResourceWatcher<XdsClusterResource.CdsUpdate> cdsWatcher2;