TransportSet shutdown() also shuts down the pending transport.

Previously TransportSet.shutdown() only shuts down the active transport,
which means a transport will not be shutdown if it's not ready yet. This
issue was introduced by grpc#1494 that postponed the assignment of the
active transport till transport ready.

Author: zhangkun83

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

@@ -102,6 +102,13 @@ final class TransportSet implements WithLogId {
   private final Collection<ManagedClientTransport> transports =
       new ArrayList<ManagedClientTransport>();
 
+  /**
+   * The to-be active transport, which is not ready yet.
+   */
+  @GuardedBy("lock")
+  @Nullable
+  private ManagedClientTransport pendingTransport;
+
   private final LoadBalancer<ClientTransport> loadBalancer;
 
   @GuardedBy("lock")
@@ -194,6 +201,7 @@ private void startNewTransport(DelayedClientTransport delayedTransport) {
       log.log(Level.FINE, "[{0}] Created {1} for {2}",
           new Object[] {getLogId(), transport.getLogId(), address});
     }
+    pendingTransport = transport;
     transports.add(transport);
     transport.start(new TransportListener(transport, delayedTransport, address));
   }
@@ -257,6 +265,7 @@ public ClientTransport get() {
    */
   final void shutdown() {
     ManagedClientTransport savedActiveTransport;
+    ManagedClientTransport savedPendingTransport;
     boolean runCallback = false;
     synchronized (lock) {
       if (shutdown) {
@@ -265,6 +274,7 @@ final void shutdown() {
       // Transition to SHUTDOWN
       shutdown = true;
       savedActiveTransport = activeTransport;
+      savedPendingTransport = pendingTransport;
       activeTransport = null;
       if (transports.isEmpty()) {
         runCallback = true;
@@ -274,6 +284,9 @@ final void shutdown() {
     if (savedActiveTransport != null) {
       savedActiveTransport.shutdown();
     }
+    if (savedPendingTransport != null) {
+      savedPendingTransport.shutdown();
+    }
     if (runCallback) {
       callback.onTerminated();
     }
@@ -358,7 +371,9 @@ public void transportReady() {
               "Unexpected non-null activeTransport");
         } else if (activeTransport == delayedTransport) {
           // Transition to READY
+          Preconditions.checkState(pendingTransport == transport, "transport mismatch");
           activeTransport = transport;
+          pendingTransport = null;
         }
       }
       delayedTransport.setTransport(transport);

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

@@ -498,6 +498,22 @@ public void shutdownBeforeTransportCreatedWithoutPendingStream() throws Exceptio
     assertEquals(0, transports.size());
   }
 
+  @Test
+  public void shutdownBeforeTransportReady() throws Exception {
+    SocketAddress addr = mock(SocketAddress.class);
+    createTransportSet(addr);
+
+    ClientTransport pick = transportSet.obtainActiveTransport();
+    MockClientTransportInfo transportInfo = transports.poll();
+    assertNotSame(transportInfo.transport, pick);
+
+    // Shutdown the TransportSet before the pending transport is ready
+    transportSet.shutdown();
+
+    // The transport should've been shut down even though it's not the active transport yet.
+    verify(transportInfo.transport).shutdown();
+  }
+
   @Test
   public void obtainTransportAfterShutdown() throws Exception {
     SocketAddress addr = mock(SocketAddress.class);