xds: Improve ClusterImpl's FakeSubchannel to verify state changes

The main goal was to make sure subchannels went CONNECTING only after a
connection was requested (since the test doesn't transition to
CONNECTING from TF). That helps guarantee that the test is using the
expected subchannel.

The missing ClusterImplLB.requestConnection() doesn't actually matter
much, as cluster manager doesn't propagate connection requests.

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

@@ -163,6 +163,13 @@ public void handleNameResolutionError(Status error) {
     }
   }
 
+  @Override
+  public void requestConnection() {
+    if (childSwitchLb != null) {
+      childSwitchLb.requestConnection();
+    }
+  }
+
   @Override
   public void shutdown() {
     if (dropStats != null) {

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

@@ -18,6 +18,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -280,6 +281,7 @@ public void pick_addsLocalityLabel() {
     FakeLoadBalancer leafBalancer = Iterables.getOnlyElement(downstreamBalancers);
     leafBalancer.createSubChannel();
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     assertThat(currentState).isEqualTo(ConnectivityState.READY);
@@ -309,6 +311,7 @@ public void recordLoadStats() {
     FakeLoadBalancer leafBalancer = Iterables.getOnlyElement(downstreamBalancers);
     Subchannel subchannel = leafBalancer.createSubChannel();
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     assertThat(currentState).isEqualTo(ConnectivityState.READY);
@@ -407,6 +410,7 @@ public void pickFirstLoadReport_onUpdateAddress() {
     // Leaf balancer is created by Pick First. Get FakeSubchannel created to update attributes
     // A real subchannel would get these attributes from the connected address's EAG locality.
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     assertThat(currentState).isEqualTo(ConnectivityState.READY);
@@ -490,6 +494,7 @@ public void dropRpcsWithRespectToLbConfigDropCategories() {
         .isEqualTo(endpoint.getAddresses());
     leafBalancer.createSubChannel();
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
 
@@ -571,6 +576,7 @@ private void subtest_maxConcurrentRequests_appliedByLbConfig(boolean enableCircu
         .isEqualTo(endpoint.getAddresses());
     leafBalancer.createSubChannel();
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     assertThat(currentState).isEqualTo(ConnectivityState.READY);
@@ -665,6 +671,7 @@ private void subtest_maxConcurrentRequests_appliedWithDefaultValue(
         .isEqualTo(endpoint.getAddresses());
     leafBalancer.createSubChannel();
     FakeSubchannel fakeSubchannel = helper.subchannels.poll();
+    fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
     fakeSubchannel.setConnectedEagIndex(0);
     fakeSubchannel.updateState(ConnectivityStateInfo.forNonError(ConnectivityState.READY));
     assertThat(currentState).isEqualTo(ConnectivityState.READY);
@@ -943,6 +950,7 @@ Subchannel createSubChannel() {
               new FixedResultPicker(PickResult.withSubchannel(subchannel)));
         }
       });
+      subchannel.requestConnection();
       return subchannel;
     }
   }
@@ -989,6 +997,8 @@ private static final class FakeSubchannel extends Subchannel {
     private final Attributes attrs;
     private SubchannelStateListener listener;
     private Attributes connectedAttributes;
+    private ConnectivityStateInfo state = ConnectivityStateInfo.forNonError(ConnectivityState.IDLE);
+    private boolean connectionRequested;
 
     private FakeSubchannel(List<EquivalentAddressGroup> eags, Attributes attrs) {
       this.eags = eags;
@@ -1006,6 +1016,9 @@ public void shutdown() {
 
     @Override
     public void requestConnection() {
+      if (state.getState() == ConnectivityState.IDLE) {
+        this.connectionRequested = true;
+      }
     }
 
     @Override
@@ -1028,6 +1041,26 @@ public Attributes getConnectedAddressAttributes() {
     }
 
     public void updateState(ConnectivityStateInfo newState) {
+      switch (newState.getState()) {
+        case IDLE:
+          assertThat(state.getState()).isEqualTo(ConnectivityState.READY);
+          break;
+        case CONNECTING:
+          assertThat(state.getState())
+              .isIn(Arrays.asList(ConnectivityState.IDLE, ConnectivityState.TRANSIENT_FAILURE));
+          if (state.getState() == ConnectivityState.IDLE) {
+            assertWithMessage("Connection requested").that(this.connectionRequested).isTrue();
+            this.connectionRequested = false;
+          }
+          break;
+        case READY:
+        case TRANSIENT_FAILURE:
+          assertThat(state.getState()).isEqualTo(ConnectivityState.CONNECTING);
+          break;
+        default:
+          break;
+      }
+      this.state = newState;
       listener.onSubchannelState(newState);
     }