xds: better error handling to avoid RPC hangup

This change will fail application RPC immediately if XdsClient encounters any error instead of retrying or getting to fallback silently.

There could be optimization if the channel is currently READY while XdsClient stream just closed due to connection error, in which case we could still be using the current available subchannels while retrying, but this requires the LB knows the semantics of error status from the XdsClient. This optimization is not worth the effort for now.

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

@@ -502,6 +502,7 @@ final class LookasideLb extends LoadBalancer {
     @Override
     public void onError(Status error) {
       channelLogger.log(ChannelLogLevel.ERROR, "EDS load balancer received an error: {0}",  error);
+      lookasideLbHelper.updateBalancingState(TRANSIENT_FAILURE, new ErrorPicker(error));
       endpointUpdateCallback.onError();
     }
   }

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

@@ -28,6 +28,7 @@ import io.grpc.Status;
 import io.grpc.SynchronizationContext.ScheduledHandle;
 import io.grpc.util.ForwardingLoadBalancerHelper;
 import io.grpc.xds.LookasideLb.EndpointUpdateCallback;
+import io.grpc.xds.XdsSubchannelPickers.ErrorPicker;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.CheckForNull;
 import javax.annotation.Nullable;
@@ -118,6 +119,10 @@ final class XdsLoadBalancer2 extends LoadBalancer {
 
         @Override
         public void run() {
+          helper.updateBalancingState(
+              ConnectivityState.TRANSIENT_FAILURE,
+              new ErrorPicker(Status.UNAVAILABLE.withDescription(
+                  "Channel is not ready when timeout for entering fallback mode happens")));
           useFallbackPolicy();
         }
       }

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

@@ -787,8 +787,11 @@ public class LookasideLbTest {
   public void verifyRpcErrorPropagation() {
     lookasideLb.handleResolvedAddresses(defaultResolvedAddress);
 
+    verify(helper, never()).updateBalancingState(
+        eq(TRANSIENT_FAILURE), any(SubchannelPicker.class));
     verify(edsUpdateCallback, never()).onError();
     serverResponseWriter.onError(new RuntimeException());
+    verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), any(SubchannelPicker.class));
     verify(edsUpdateCallback).onError();
   }
 

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

@@ -19,6 +19,9 @@ package io.grpc.xds;
 import static com.google.common.truth.Truth.assertThat;
 import static io.grpc.ConnectivityState.CONNECTING;
 import static io.grpc.ConnectivityState.READY;
+import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
@@ -144,7 +147,9 @@ public class XdsLoadBalancer2Test {
     fakeClock.forwardTime(9, TimeUnit.SECONDS);
     edsUpdateCallback.onWorking();
     verifyNotInFallbackMode();
+
     fakeClock.forwardTime(1, TimeUnit.SECONDS);
+    verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), any(SubchannelPicker.class));
     verifyInFallbackMode();
 
     SubchannelPicker subchannelPicker = mock(SubchannelPicker.class);