core: PickFirstLB should not return a subchannel during CONNECTING

This is a follow-up from noticing a breakage in gRPC-LB because it
wasn't checking if the connectivity state changed even if the picker was
identical.

lb_serverStatusCodeConversion() has been misleading since 42e1829
("xds: Do RLS fallback policy eagar start"). At that point, the
subchannel it marked as READY was for the default target's policy, not
the policy for wilderness. However, since old PF policy provided a
subchannel when CONNECTING, everything was "fine", but RLS would
mistakenly count toward target_picks.

This demonstrates that RLS target_picks has been broken since it was
introduced for PF, as PF relied on the caller to avoid the picker when
it was CONNECTING. This may have been hard to notice in production, as
the metrics become correct as soon as the connection is established, so
as long as you use the channel for a while, the duplicate counting would
become a small percentage of the overall amount.

Author: ejona86

core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.java

@@ -86,8 +86,7 @@ public void onSubchannelState(ConnectivityStateInfo stateInfo) {
 
       // The channel state does not get updated when doing name resolving today, so for the moment
       // let LB report CONNECTION and call subchannel.requestConnection() immediately.
-      updateBalancingState(
-          CONNECTING, new FixedResultPicker(PickResult.withSubchannel(subchannel)));
+      updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
       subchannel.requestConnection();
     } else {
       subchannel.updateAddresses(servers);

core/src/test/java/io/grpc/internal/PickFirstLoadBalancerTest.java

@@ -219,7 +219,7 @@ public void refreshNameResolutionAfterSubchannelConnectionBroken() {
     inOrder.verify(mockSubchannel).start(stateListenerCaptor.capture());
     SubchannelStateListener stateListener = stateListenerCaptor.getValue();
     inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    assertSame(mockSubchannel, pickerCaptor.getValue().pickSubchannel(mockArgs).getSubchannel());
+    assertThat(pickerCaptor.getValue().pickSubchannel(mockArgs).hasResult()).isFalse();
     inOrder.verify(mockSubchannel).requestConnection();
 
     stateListener.onSubchannelState(ConnectivityStateInfo.forNonError(CONNECTING));
@@ -278,7 +278,7 @@ public void pickAfterResolvedAndChanged() throws Exception {
     assertThat(args.getAddresses()).isEqualTo(servers);
     inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
     verify(mockSubchannel).requestConnection();
-    assertEquals(mockSubchannel, pickerCaptor.getValue().pickSubchannel(mockArgs).getSubchannel());
+    assertThat(pickerCaptor.getValue().pickSubchannel(mockArgs).hasResult()).isFalse();
 
     loadBalancer.acceptResolvedAddresses(
         ResolvedAddresses.newBuilder().setAddresses(newServers).setAttributes(affinity).build());
@@ -300,7 +300,7 @@ public void pickAfterStateChangeAfterResolution() throws Exception {
     verify(mockSubchannel).start(stateListenerCaptor.capture());
     SubchannelStateListener stateListener = stateListenerCaptor.getValue();
     verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    Subchannel subchannel = pickerCaptor.getValue().pickSubchannel(mockArgs).getSubchannel();
+    assertThat(pickerCaptor.getValue().pickSubchannel(mockArgs).hasResult()).isFalse();
     reset(mockHelper);
     when(mockHelper.getSynchronizationContext()).thenReturn(syncContext);
 
@@ -317,7 +317,7 @@ public void pickAfterStateChangeAfterResolution() throws Exception {
 
     stateListener.onSubchannelState(ConnectivityStateInfo.forNonError(READY));
     inOrder.verify(mockHelper).updateBalancingState(eq(READY), pickerCaptor.capture());
-    assertEquals(subchannel, pickerCaptor.getValue().pickSubchannel(mockArgs).getSubchannel());
+    assertEquals(mockSubchannel, pickerCaptor.getValue().pickSubchannel(mockArgs).getSubchannel());
 
     verify(mockHelper, atLeast(0)).getSynchronizationContext();  // Don't care
     verifyNoMoreInteractions(mockHelper);
@@ -405,8 +405,7 @@ public void nameResolutionSuccessAfterError() throws Exception {
     inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
     verify(mockSubchannel).requestConnection();
 
-    assertEquals(mockSubchannel, pickerCaptor.getValue().pickSubchannel(mockArgs)
-        .getSubchannel());
+    assertThat(pickerCaptor.getValue().pickSubchannel(mockArgs).hasResult()).isFalse();
 
     assertEquals(pickerCaptor.getValue().pickSubchannel(mockArgs),
         pickerCaptor.getValue().pickSubchannel(mockArgs));

rls/src/test/java/io/grpc/rls/RlsLoadBalancerTest.java

@@ -72,7 +72,6 @@
 import io.grpc.inprocess.InProcessServerBuilder;
 import io.grpc.internal.FakeClock;
 import io.grpc.internal.JsonParser;
-import io.grpc.internal.PickFirstLoadBalancerProvider;
 import io.grpc.internal.PickSubchannelArgsImpl;
 import io.grpc.internal.testing.StreamRecorder;
 import io.grpc.lookup.v1.RouteLookupServiceGrpc;
@@ -212,12 +211,14 @@ public void lb_serverStatusCodeConversion() throws Exception {
         throw new RuntimeException(e);
       }
     });
+    assertThat(subchannels.poll()).isNotNull(); // default target
+    assertThat(subchannels.poll()).isNull();
+    // Warm-up pick; will be queued
     InOrder inOrder = inOrder(helper);
     inOrder.verify(helper)
         .updateBalancingState(eq(ConnectivityState.CONNECTING), pickerCaptor.capture());
     SubchannelPicker picker = pickerCaptor.getValue();
     PickSubchannelArgs fakeSearchMethodArgs = newPickSubchannelArgs(fakeSearchMethod);
-    // Warm-up pick; will be queued
     PickResult res = picker.pickSubchannel(fakeSearchMethodArgs);
     assertThat(res.getStatus().isOk()).isTrue();
     assertThat(res.getSubchannel()).isNull();
@@ -230,8 +231,7 @@ public void lb_serverStatusCodeConversion() throws Exception {
     subchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     res = picker.pickSubchannel(fakeSearchMethodArgs);
     assertThat(res.getStatus().getCode()).isEqualTo(Status.Code.OK);
-    int expectedTimes = PickFirstLoadBalancerProvider.isEnabledNewPickFirst() ? 1 : 2;
-    verifyLongCounterAdd("grpc.lb.rls.target_picks", expectedTimes, 1, "wilderness", "complete");
+    verifyLongCounterAdd("grpc.lb.rls.target_picks", 1, 1, "wilderness", "complete");
 
     // Check on conversion
     Throwable cause = new Throwable("cause");
@@ -284,8 +284,7 @@ public void lb_working_withDefaultTarget_rlsResponding() throws Exception {
     res = picker.pickSubchannel(searchSubchannelArgs);
     assertThat(subchannelIsReady(res.getSubchannel())).isTrue();
     assertThat(res.getSubchannel()).isSameInstanceAs(searchSubchannel);
-    int expectedTimes = PickFirstLoadBalancerProvider.isEnabledNewPickFirst() ? 1 : 2;
-    verifyLongCounterAdd("grpc.lb.rls.target_picks", expectedTimes, 1, "wilderness", "complete");
+    verifyLongCounterAdd("grpc.lb.rls.target_picks", 1, 1, "wilderness", "complete");
 
     // rescue should be pending status although the overall channel state is READY
     res = picker.pickSubchannel(rescueSubchannelArgs);
@@ -431,7 +430,7 @@ public void lb_working_withDefaultTarget_noRlsResponse() throws Exception {
     inOrder.verify(helper).getMetricRecorder();
     inOrder.verify(helper).getChannelTarget();
     inOrder.verifyNoMoreInteractions();
-    int times = PickFirstLoadBalancerProvider.isEnabledNewPickFirst() ? 1 : 2;
+    int times = 1;
     verifyLongCounterAdd("grpc.lb.rls.default_target_picks", times, 1,
         "defaultTarget", "complete");
 
@@ -536,8 +535,7 @@ public void lb_working_withoutDefaultTarget() throws Exception {
     res = picker.pickSubchannel(newPickSubchannelArgs(fakeSearchMethod));
     assertThat(res.getStatus().isOk()).isFalse();
     assertThat(subchannelIsReady(res.getSubchannel())).isFalse();
-    int expectedTimes = PickFirstLoadBalancerProvider.isEnabledNewPickFirst() ? 1 : 2;
-    verifyLongCounterAdd("grpc.lb.rls.target_picks", expectedTimes, 1, "wilderness", "complete");
+    verifyLongCounterAdd("grpc.lb.rls.target_picks", 1, 1, "wilderness", "complete");
 
     res = picker.pickSubchannel(newPickSubchannelArgs(fakeRescueMethod));
     assertThat(subchannelIsReady(res.getSubchannel())).isTrue();