Finished fixing rebase

Author: larry-safran

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

@@ -164,6 +164,8 @@ void adjustResourceSubscription(XdsResourceType<?> resourceType) {
       resources = Collections.emptyList();
     }
     adsStream.sendDiscoveryRequest(resourceType, resources);
+    resourceStore.startMissingResourceTimers(resources, resourceType);
+
     if (resources.isEmpty()) {
       // The resource type no longer has subscribing resources; clean up references to it
       versions.remove(resourceType);

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

@@ -447,5 +447,12 @@ Collection<String> getSubscribedResources(
         ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type);
 
     Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
+
+    /**
+     * For any of the subscribers to one of the specified resources, if there isn't a result or
+     * an existing timer for the resource, start a timer for the resource.
+     */
+    void startMissingResourceTimers(Collection<String> resourceNames,
+                                    XdsResourceType<?> resourceType);
   }
 }

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

@@ -201,6 +201,20 @@ public Collection<String> getSubscribedResources(
     return retVal.isEmpty() ? null : retVal;
   }
 
+  @Override
+  public void startMissingResourceTimers(Collection<String> resourceNames,
+                                         XdsResourceType<?> resourceType) {
+    Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap =
+        resourceSubscribers.get(resourceType);
+
+    for (String resourceName : resourceNames) {
+      ResourceSubscriber<?> subscriber = subscriberMap.get(resourceName);
+      if (subscriber.respTimer == null && !subscriber.hasResult()) {
+        subscriber.restartTimer();
+      }
+    }
+  }
+
   // 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

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

@@ -503,6 +503,12 @@ public Collection<String> getSubscribedResources(
       return null;
     }
 
+    @Override
+    public void startMissingResourceTimers(Collection<String> resourceNames,
+                                           XdsResourceType<?> resourceType) {
+      // do nothing
+    }
+
     @Override
     public Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl() {
       return ImmutableMap.of();

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

@@ -3725,8 +3725,7 @@ public void streamClosedAndRetryRaceWithAddRemoveWatchers() {
   }
 
   @Test
-  public void streamClosedAndRetryRestartsResourceInitialFetchTimerForUnresolvedResources()
-      throws InterruptedException {
+  public void streamClosedAndRetryRestartsResourceInitialFetchTimerForUnresolvedResources() {
     xdsClient.watchXdsResource(XdsListenerResource.getInstance(), LDS_RESOURCE, ldsResourceWatcher);
     xdsClient.watchXdsResource(XdsRouteConfigureResource.getInstance(), RDS_RESOURCE,
         rdsResourceWatcher);
@@ -3762,7 +3761,7 @@ public void streamClosedAndRetryRestartsResourceInitialFetchTimerForUnresolvedRe
     verify(ldsResourceWatcher, never()).onError(errorCaptor.capture());
     verify(rdsResourceWatcher, never()).onError(errorCaptor.capture());
     verify(cdsResourceWatcher, never()).onError(errorCaptor.capture());
-
+    verify(edsResourceWatcher, never()).onError(errorCaptor.capture());
     // Check metric data.
     callback_ReportServerConnection();
     verifyServerConnection(4, true, xdsServerInfo.target());