Fix fallback to pass all tests

Author: larry-safran

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

@@ -137,14 +137,14 @@ static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
     private XdsClient xdsClient;
     @GuardedBy("lock")
     private int refCount;
-    private final Constructor backoffProviderConstructor;
+    private final Constructor<? extends BackoffPolicy.Provider> backoffProviderConstructor;
 
     @VisibleForTesting
     RefCountedXdsClientObjectPool(BootstrapInfo bootstrapInfo, String target,
                                   Class<? extends BackoffPolicy.Provider> rawBackoffProviderClass) {
       this.bootstrapInfo = checkNotNull(bootstrapInfo);
       this.target = target;
-      Class backoffProviderClass = rawBackoffProviderClass != null
+      Class<? extends BackoffPolicy.Provider> backoffProviderClass = rawBackoffProviderClass != null
           ? rawBackoffProviderClass
           : ExponentialBackoffPolicy.Provider.class;
       try {

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

@@ -99,7 +99,6 @@ final class ControlPlaneClient {
       SynchronizationContext syncContext,
       BackoffPolicy.Provider backoffPolicyProvider,
       Supplier<Stopwatch> stopwatchSupplier,
-      XdsClient xdsClient, // Has been replaced by xdsResponseHandler
       MessagePrettyPrinter messagePrinter) {
     this.serverInfo = checkNotNull(serverInfo, "serverInfo");
     this.xdsTransport = checkNotNull(xdsTransport, "xdsTransport");
@@ -152,10 +151,13 @@ void adjustResourceSubscription(XdsResourceType<?> resourceType, String authorit
     }
     if (adsStream == null) {
       startRpcStream();
+      // when the stream becomes ready, it will send the discovery requests
+      return;
     }
+
     Collection<String> resources =
         resourceStore.getSubscribedResources(serverInfo, resourceType, authority);
-    if (resources != null) {
+    if (resources != null && !resources.isEmpty()) {
       adsStream.sendDiscoveryRequest(resourceType, resources);
     }
   }
@@ -244,13 +246,16 @@ void sendDiscoveryRequests(String authority) {
         new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());
     for (XdsResourceType<?> type : subscribedResourceTypes) {
       Collection<String> resources =
-          type.updateInPlaceOnFallback()
-          ? resourceStore.getAllResources(type, authority)
-          : resourceStore.getSubscribedResources(serverInfo, type, authority);
+          resourceStore.getSubscribedResources(serverInfo, type, authority);
       if (resources != null && !resources.isEmpty()) {
         adsStream.sendDiscoveryRequest(type, resources);
       }
+    }
+  }
 
+  public <T extends XdsClient.ResourceUpdate> void removeNonceForType(XdsResourceType<T> type) {
+    if (!shutdown && adsStream != null) {
+      adsStream.respNonces.remove(type);
     }
   }
 
@@ -278,6 +283,7 @@ XdsResourceType<?> fromTypeUrl(String typeUrl) {
   private class AdsStream implements EventHandler<DiscoveryResponse> {
     private boolean responseReceived;
     private boolean closed;
+    private boolean hasGoneReady = false;
     // Response nonce for the most recently received discovery responses of each resource type.
     // Client initiated requests start response nonce with empty string.
     // Nonce in each response is echoed back in the following ACK/NACK request. It is
@@ -337,6 +343,11 @@ final void sendDiscoveryRequest(XdsResourceType<?> type, Collection<String> reso
 
     @Override
     public void onReady() {
+      if (shutdown || closed || hasGoneReady) {
+        return;
+      }
+
+      hasGoneReady = true;
       syncContext.execute(ControlPlaneClient.this::readyHandler);
     }
 
@@ -439,7 +450,11 @@ private void cleanUp() {
 
   private long scheduleRpcRetry() {
     long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
+    if (retryBackoffPolicy == null) {
+      retryBackoffPolicy = backoffPolicyProvider.get();
+    }
     long delayNanos = Math.max(0, retryBackoffPolicy.nextBackoffNanos() - elapsed);
+
     rpcRetryTimer =
         syncContext.schedule(new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);
     return delayNanos;

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

@@ -417,7 +417,7 @@ void handleResourceResponse(
      * Is expected do manage the ControlPlanClients and cache updates associated with
      * Moving to or from a fallback server.
      *
-     * Must be synchronized.
+     * <p>Must be synchronized.
      */
     void handleStreamReady(ServerInfo serverInfo);
   }
@@ -434,17 +434,15 @@ public interface ResourceStore {
     @Nullable
     Collection<String> getSubscribedResources(ServerInfo serverInfo,
                                               XdsResourceType<? extends ResourceUpdate> type);
-    default Collection<String> getSubscribedResources(
-        ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type, String authority) {
-      return getSubscribedResources(serverInfo, type);
-    };
 
-    Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
-
-    Collection<String> getAllResources(XdsResourceType<?> type);
-    default Collection<String> getAllResources(XdsResourceType<?> type, String authority) {
-      return getAllResources(type);
-    }
+    /**
+     * Like {@link #getSubscribedResources(ServerInfo, XdsResourceType)}, but limits the results to
+     * those matching the given authority.
+     */
+    @Nullable
+    Collection<String> getSubscribedResources(
+        ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type, String authority);
 
+    Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
   }
 }

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

@@ -82,16 +82,20 @@ public void uncaughtException(Thread t, Throwable e) {
         }
       });
 
-  private final Map<ServerInfo, LoadStatsManager2> loadStatsManagerMap =
-      new HashMap<>();
-  final Map<ServerInfo, LoadReportClient> serverLrsClientMap =
-      new HashMap<>();
+  private final Map<ServerInfo, LoadStatsManager2> loadStatsManagerMap = new HashMap<>();
+  final Map<ServerInfo, LoadReportClient> serverLrsClientMap = new HashMap<>();
+  /** Map of authority to its active control plane client (affected by xds fallback). */
+  private final Map<String, ControlPlaneClient> activeCpClients = new HashMap<>();
 
   private final Map<ServerInfo, ControlPlaneClient> serverCpClientMap = new HashMap<>();
+
+  /** Maps resource type to the corresponding map of subscribers (keyed by subscriber name). */
   private final Map<XdsResourceType<? extends ResourceUpdate>,
       Map<String, ResourceSubscriber<? extends ResourceUpdate>>>
       resourceSubscribers = new HashMap<>();
+  /** Maps typeUrl to the corresponding XdsResourceType. */
   private final Map<String, XdsResourceType<?>> subscribedResourceTypeUrls = new HashMap<>();
+
   private final XdsTransportFactory xdsTransportFactory;
   private final Bootstrapper.BootstrapInfo bootstrapInfo;
   private final ScheduledExecutorService timeService;
@@ -103,10 +107,6 @@ public void uncaughtException(Thread t, Throwable e) {
   private final XdsLogger logger;
   private volatile boolean isShutdown;
   private final MessagePrettyPrinter messagePrinter;
-  /**
-   * Map of authority to its active control plane client.
-   */
-  private Map<String, ControlPlaneClient> activeCpClients = new HashMap<>();
 
   public XdsClientImpl(
       XdsTransportFactory xdsTransportFactory,
@@ -172,7 +172,7 @@ public Collection<String> getSubscribedResources(
   @Override
   public Collection<String> getSubscribedResources(
       ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type, String authority) {
-    return getSubscribedResources(serverInfo, type, authority, false);
+    return getSubscribedResources(serverInfo, type, authority, true);
   }
 
   private Collection<String> getSubscribedResources(ServerInfo serverInfo, XdsResourceType<?
@@ -183,44 +183,21 @@ private Collection<String> getSubscribedResources(ServerInfo serverInfo, XdsReso
     ControlPlaneClient controlPlaneClient = serverCpClientMap.get(serverInfo);
     for (String key : resources.keySet()) {
       ResourceSubscriber<? extends ResourceUpdate> resource = resources.get(key);
-      if (resource.controlPlaneClient.equals(controlPlaneClient) &&
-          (!useAuthority || Objects.equals(authority, resource.authority))) {
+      if (resource.controlPlaneClient.equals(controlPlaneClient)
+          && (!useAuthority || Objects.equals(authority, resource.authority))) {
         builder.add(key);
       }
     }
     Collection<String> retVal = builder.build();
     return retVal.isEmpty() ? null : retVal;
   }
 
-  @Override
-  public Collection<String> getAllResources(XdsResourceType<?> type) {
-    if (!resourceSubscribers.containsKey(type)) {
-      return null;
-    }
-    return ImmutableList.copyOf(resourceSubscribers.get(type).keySet());
-  }
-
-  @Override
-  public Collection<String> getAllResources(XdsResourceType<?> type, String authority) {
-    if (!resourceSubscribers.containsKey(type)) {
-      return null;
-    }
-
-    List<String> matchingResources = new ArrayList<>();
-    for (ResourceSubscriber<?> resource : resourceSubscribers.get(type).values()) {
-      if (Objects.equals(authority, resource.authority)) {
-        matchingResources.add(resource.resource);
-      }
-    }
-    return matchingResources;
-  }
-
 
   // As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
   // ResourceTypes that do not have subscribers does not show up in the snapshot keys.
   @Override
   public ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
-     getSubscribedResourcesMetadataSnapshot() {
+      getSubscribedResourcesMetadataSnapshot() {
     final SettableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>> future =
         SettableFuture.create();
     syncContext.execute(new Runnable() {
@@ -290,17 +267,22 @@ public void run() {
         if (subscriber == null) {
           logger.log(XdsLogLevel.WARNING, "double cancel of resource watch for {0}:{1}",
               type.typeName(), resourceName);
+          return;
         }
         subscriber.removeWatcher(watcher);
         if (!subscriber.isWatched()) {
           subscriber.cancelResourceWatch();
           resourceSubscribers.get(type).remove(resourceName);
-          if (subscriber.controlPlaneClient != null) {
-            subscriber.controlPlaneClient.adjustResourceSubscription(type, subscriber.authority);
+          ControlPlaneClient controlPlaneClient = subscriber.controlPlaneClient;
+          if (controlPlaneClient != null) {
+            controlPlaneClient.adjustResourceSubscription(type, subscriber.authority);
           }
           if (resourceSubscribers.get(type).isEmpty()) {
             resourceSubscribers.remove(type);
             subscribedResourceTypeUrls.remove(type.typeUrl());
+            if (controlPlaneClient != null) {
+              controlPlaneClient.removeNonceForType(type);
+            }
           }
         }
       }
@@ -432,7 +414,6 @@ private ControlPlaneClient getOrCreateControlPlaneClient(ServerInfo serverInfo)
         syncContext,
         backoffPolicyProvider,
         stopwatchSupplier,
-        this,
         messagePrinter
     );
 
@@ -1015,8 +996,8 @@ private void internalHandleStreamReady(
         return;
       }
 
-      if (activeCpClient.isReady() &&
-          compareCpClients(activeCpClient, controlPlaneClient, authority) < 0) {
+      if (activeCpClient.isReady()
+          && compareCpClients(activeCpClient, controlPlaneClient, authority) < 0) {
         logger.log(XdsLogLevel.INFO, "Ignoring stream restart for lower priority server {0}",
             serverInfo.target());
         return;
@@ -1027,7 +1008,7 @@ private void internalHandleStreamReady(
           resourceSubscribers.values()) {
         for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
           if (subscriber.controlPlaneClient == controlPlaneClient
-              && subscriber.authority == authority) {
+              && Objects.equals(subscriber.authority, authority)) {
             subscriber.restartTimer();
           }
         }
@@ -1036,9 +1017,10 @@ private void internalHandleStreamReady(
       if (activeCpClient != controlPlaneClient) {
         activeCpClients.put(authority, controlPlaneClient);
         updateRootResources(controlPlaneClient, authority, false);
-        controlPlaneClient.sendDiscoveryRequests(authority);
       }
 
+      controlPlaneClient.sendDiscoveryRequests(authority);
+
       // Shutdown any lower priority control plane clients.
       Iterator<ControlPlaneClient> iterator = serverCpClientMap.values().iterator();
       while (iterator.hasNext()) {
@@ -1049,7 +1031,6 @@ private void internalHandleStreamReady(
         }
       }
     }
-
   }
 
   private int compareCpClients(ControlPlaneClient base, ControlPlaneClient other,
@@ -1059,10 +1040,13 @@ private int compareCpClients(ControlPlaneClient base, ControlPlaneClient other,
     }
 
     ImmutableList<ServerInfo> serverInfos = getServerInfos(authority);
-    if (!serverInfos.contains(base.getServerInfo()) || !serverInfos.contains(other.getServerInfo())) {
+    ServerInfo baseServerInfo = base.getServerInfo();
+    ServerInfo otherServerInfo = other.getServerInfo();
+
+    if (!serverInfos.contains(baseServerInfo) || !serverInfos.contains(otherServerInfo)) {
       return -100; // At least one of them isn't serving this authority
     }
 
-    return serverInfos.indexOf(base.getServerInfo()) - serverInfos.indexOf(other.getServerInfo());
+    return serverInfos.indexOf(baseServerInfo) - serverInfos.indexOf(otherServerInfo);
   }
 }

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

@@ -455,15 +455,18 @@ public Collection<String> getSubscribedResources(ServerInfo serverInfo,
       return null;
     }
 
+    @Nullable
     @Override
-    public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {
-      return ImmutableMap.of();
+    public Collection<String> getSubscribedResources(
+        ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type, String authority) {
+      return null;
     }
 
     @Override
-    public Collection<String> getAllResources(XdsResourceType<?> type) {
-      return null;
+    public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {
+      return ImmutableMap.of();
     }
+
   }
 
   private static class FakeXdsClientPoolFactory implements XdsClientPoolFactory {

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

@@ -69,6 +69,7 @@ public class XdsClientFallbackTest {
         public void onChanged(XdsListenerResource.LdsUpdate update) {
           log.info("LDS update: " + update);
         }
+
         @Override
         public void onError(Status error) {
           log.info("LDS update error: " + error.getDescription());
@@ -109,7 +110,7 @@ public void setUp() throws XdsInitializationException {
 
     SharedXdsClientPoolProvider clientPoolProvider = new SharedXdsClientPoolProvider();
     clientPoolProvider.setBootstrapOverride(defaultBootstrapOverride());
-//    clientPoolProvider.setBackoffProviderClass(MinimalBackoffPolicyProvider.class);
+    //    clientPoolProvider.setBackoffProviderClass(MinimalBackoffPolicyProvider.class);
     xdsClientPool = clientPoolProvider.getOrCreate(DUMMY_TARGET);
   }