xds: Fail CDS cluster when EDS update has no usable endpoints

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

@@ -156,8 +156,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
       if (addresses.isEmpty()) {
         Status status = Status.UNAVAILABLE
             .withDescription("No usable endpoint from cluster: " + clusterName);
-        delegate.handleNameResolutionError(status);
-        return status;
+        return fail(status);
       }
       Object gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
           lbRegistry.getProvider(PRIORITY_POLICY_NAME),
@@ -223,6 +222,7 @@ public void shutdown() {
   @CheckReturnValue // don't forget to return up the stack after the fail call
   private Status fail(Status error) {
     delegate.shutdown();
+    delegate = new GracefulSwitchLoadBalancer(cdsLbHelper);
     addBackendServicePickDetailsLabel = false;
     helper.updateBalancingState(
         TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));

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

@@ -31,6 +31,7 @@
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -62,6 +63,7 @@
 import io.grpc.HttpConnectProxiedSocketAddress;
 import io.grpc.InternalEquivalentAddressGroup;
 import io.grpc.LoadBalancer;
+import io.grpc.LoadBalancer.FixedResultPicker;
 import io.grpc.LoadBalancer.Helper;
 import io.grpc.LoadBalancer.PickDetailsConsumer;
 import io.grpc.LoadBalancer.PickResult;
@@ -887,6 +889,57 @@ public void handleEdsResource_noHealthyEndpoint() {
     assertPicker(pickerCaptor.getValue(), expectedStatus, null);
   }
 
+  @Test
+  public void handleEdsResource_allEndpointsRemovedAfterReady_failsAndRecovers() {
+    ClusterLoadAssignment clusterLoadAssignment = ClusterLoadAssignment.newBuilder()
+        .setClusterName(EDS_SERVICE_NAME)
+        .addEndpoints(LocalityLbEndpoints.newBuilder()
+          .setLoadBalancingWeight(UInt32Value.of(100))
+          .setLocality(LOCALITY1)
+          .addLbEndpoints(newSocketLbEndpoint("127.0.0.1", 8000)))
+        .build();
+    controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, ImmutableMap.of(
+        EDS_SERVICE_NAME, clusterLoadAssignment));
+    startXdsDepManager();
+
+    FakeLoadBalancer childBalancer = Iterables.getOnlyElement(childBalancers);
+    childBalancer.helper.updateBalancingState(
+        ConnectivityState.READY, new FixedResultPicker(PickResult.withNoResult()));
+    verify(helper).updateBalancingState(eq(ConnectivityState.READY), any());
+
+    // An empty ClusterLoadAssignment must not leave the previous READY picker in place
+    controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, ImmutableMap.of(
+        EDS_SERVICE_NAME,
+        ClusterLoadAssignment.newBuilder().setClusterName(EDS_SERVICE_NAME).build()));
+
+    assertThat(childBalancer.shutdown).isTrue();
+    assertThat(childBalancers).isEmpty();
+    verify(helper).updateBalancingState(
+        eq(ConnectivityState.TRANSIENT_FAILURE), pickerCaptor.capture());
+    Status expectedStatus = Status.UNAVAILABLE
+        .withDescription("No usable endpoint from cluster: " + CLUSTER);
+    assertPicker(pickerCaptor.getValue(), expectedStatus, null);
+
+    // Endpoints come back: a new child policy is created
+    controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, ImmutableMap.of(
+        EDS_SERVICE_NAME, clusterLoadAssignment.toBuilder()
+          .setEndpoints(0, clusterLoadAssignment.getEndpoints(0).toBuilder()
+            .setLbEndpoints(0, newSocketLbEndpoint("127.0.0.2", 8000)))
+          .build()));
+
+    FakeLoadBalancer newChildBalancer = Iterables.getOnlyElement(childBalancers);
+    assertThat(newChildBalancer).isNotSameInstanceAs(childBalancer);
+    assertThat(newChildBalancer.shutdown).isFalse();
+    assertAddressesEqual(
+        Arrays.asList(new EquivalentAddressGroup(newInetSocketAddress("127.0.0.2", 8000))),
+        newChildBalancer.addresses);
+
+    // The new child's READY state reaches the channel, replacing the error picker
+    newChildBalancer.helper.updateBalancingState(
+        ConnectivityState.READY, new FixedResultPicker(PickResult.withNoResult()));
+    verify(helper, times(2)).updateBalancingState(eq(ConnectivityState.READY), any());
+  }
+
   @Test
   public void onlyLogicalDnsCluster_endpointsResolved() {
     boolean originalVal = LoadStatsManager2.isEnabledOrcaLrsPropagation;