xds: Update and add tests for proactive connection

Update existing unit tests and add new test cases to validate the
proactive connection behavior. Verification counts in several tests
have been adjusted to reflect that the balancer now initiates
connections immediately upon subchannel failure.

Author: becomeStar

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

@@ -255,7 +255,7 @@ public void aggregateSubchannelStates_connectingReadyIdleFailure() {
       inOrder.verify(helper).refreshNameResolution();
       inOrder.verify(helper).updateBalancingState(eq(CONNECTING), any());
     }
-    verifyConnection(0);
+    verifyConnection(1);
   }
 
   private void verifyConnection(int times) {
@@ -537,7 +537,7 @@ public void pickWithRandomHash_firstSubchannelInTransientFailure_remainingSubcha
     // Bring one subchannel to TRANSIENT_FAILURE.
     deliverSubchannelUnreachable(getSubChannel(servers.get(0)));
     verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(1);
 
     // Pick subchannel with random hash does trigger connection by walking the ring
     // and choosing the first (at most one) IDLE subchannel along the way.
@@ -583,7 +583,7 @@ public void skipFailingHosts_pickNextNonFailingHost() {
         getSubChannel(servers.get(0)),
         ConnectivityStateInfo.forTransientFailure(
             Status.UNAVAILABLE.withDescription("unreachable")));
-    verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
+    verify(helper, atLeastOnce()).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
 
     PickResult result = pickerCaptor.getValue().pickSubchannel(args);
     assertThat(result.getStatus().isOk()).isTrue();
@@ -649,7 +649,7 @@ public void skipFailingHosts_firstTwoHostsFailed_pickNextFirstReady() {
         ConnectivityStateInfo.forTransientFailure(
             Status.PERMISSION_DENIED.withDescription("permission denied")));
     verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(2);
     PickResult result = pickerCaptor.getValue().pickSubchannel(args); // activate last subchannel
     assertThat(result.getStatus().isOk()).isTrue();
     int expectedCount = PickFirstLoadBalancerProvider.isEnabledNewPickFirst() ? 0 : 1;
@@ -721,7 +721,7 @@ public void allSubchannelsInTransientFailure() {
     }
     verify(helper, atLeastOnce())
         .updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(2);
 
     // Picking subchannel triggers connection. RPC hash hits server0.
     PickSubchannelArgs args = getDefaultPickSubchannelArgsForServer(0);
@@ -740,12 +740,13 @@ public void firstSubchannelIdle() {
     List<EquivalentAddressGroup> servers = createWeightedServerAddrs(1, 1, 1);
     initializeLbSubchannels(config, servers);
 
-    // Go to TF does nothing, though PF will try to reconnect after backoff
+    // As per gRFC A61, entering TF triggers a proactive connection attempt
+    // on an IDLE subchannel because no other subchannel is currently CONNECTING.
     deliverSubchannelState(getSubchannel(servers, 1),
         ConnectivityStateInfo.forTransientFailure(
         Status.UNAVAILABLE.withDescription("unreachable")));
     verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(1);
 
     // Picking subchannel triggers connection. RPC hash hits server0.
     PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
@@ -796,7 +797,7 @@ public void firstSubchannelFailure() {
         ConnectivityStateInfo.forTransientFailure(
             Status.UNAVAILABLE.withDescription("unreachable")));
     verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(1);
 
     // Per GRFC A61 Picking subchannel should no longer request connections that were failing
     PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
@@ -805,8 +806,6 @@ public void firstSubchannelFailure() {
     assertThat(result.getStatus().isOk()).isTrue();
     assertThat(result.getSubchannel()).isNull();
     verify(subchannelList.get(0), never()).requestConnection(); // In TF
-    verify(subchannelList.get(1)).requestConnection();
-    verify(subchannelList.get(2), never()).requestConnection(); // Not one of the first 2
   }
 
   @Test
@@ -824,7 +823,7 @@ public void secondSubchannelConnecting() {
 
     Subchannel firstSubchannel = getSubchannel(servers, 0);
     deliverSubchannelUnreachable(firstSubchannel);
-    verifyConnection(0);
+    verifyConnection(1);
 
     deliverSubchannelState(getSubchannel(servers, 2), CSI_CONNECTING);
     verify(helper, times(2)).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
@@ -833,7 +832,7 @@ public void secondSubchannelConnecting() {
     // Picking subchannel when idle triggers connection.
     deliverSubchannelState(getSubchannel(servers, 2),
         ConnectivityStateInfo.forNonError(IDLE));
-    verifyConnection(0);
+    verifyConnection(1);
     PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
     PickResult result = pickerCaptor.getValue().pickSubchannel(args);
     assertThat(result.getStatus().isOk()).isTrue();
@@ -857,7 +856,7 @@ public void secondSubchannelFailure() {
     deliverSubchannelUnreachable(firstSubchannel);
     deliverSubchannelUnreachable(getSubchannel(servers, 2));
     verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(2);
 
     // Picking subchannel triggers connection.
     PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
@@ -887,7 +886,7 @@ public void thirdSubchannelConnecting() {
     deliverSubchannelState(getSubchannel(servers, 1), CSI_CONNECTING);
     verify(helper, atLeastOnce())
         .updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(2);
 
     // Picking subchannel should not trigger connection per gRFC A61.
     PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
@@ -909,7 +908,7 @@ public void stickyTransientFailure() {
     deliverSubchannelUnreachable(firstSubchannel);
 
     verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
-    verifyConnection(0);
+    verifyConnection(1);
 
     reset(helper);
     deliverSubchannelState(firstSubchannel, ConnectivityStateInfo.forNonError(IDLE));
@@ -1127,6 +1126,39 @@ public void config_equalsTester() {
         .testEquals();
   }
 
+  @Test
+  public void tfWithoutConnectingChild_triggersIdleChildConnection() {
+    RingHashConfig config = new RingHashConfig(10, 100, "");
+    List<EquivalentAddressGroup> servers = createWeightedServerAddrs(1, 1);
+
+    initializeLbSubchannels(config, servers);
+
+    Subchannel tfSubchannel = getSubchannel(servers, 0);
+    Subchannel idleSubchannel = getSubchannel(servers, 1);
+
+    deliverSubchannelUnreachable(tfSubchannel);
+
+    Subchannel requested = connectionRequestedQueue.poll();
+    assertThat(requested).isSameInstanceAs(idleSubchannel);
+    assertThat(connectionRequestedQueue.poll()).isNull();
+  }
+
+  @Test
+  public void tfWithReadyChild_doesNotTriggerIdleChildConnection() {
+    RingHashConfig config = new RingHashConfig(10, 100, "");
+    List<EquivalentAddressGroup> servers = createWeightedServerAddrs(1, 1, 1);
+
+    initializeLbSubchannels(config, servers);
+
+    Subchannel tfSubchannel = getSubchannel(servers, 0);
+    Subchannel readySubchannel = getSubchannel(servers, 1);
+
+    deliverSubchannelState(readySubchannel, ConnectivityStateInfo.forNonError(READY));
+    deliverSubchannelUnreachable(tfSubchannel);
+
+    assertThat(connectionRequestedQueue.poll()).isNull();
+  }
+
   private List<Subchannel> initializeLbSubchannels(RingHashConfig config,
       List<EquivalentAddressGroup> servers, InitializationFlags... initFlags) {