core: handle closed transport factory in InternalSubchannel without panic

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

@@ -273,10 +273,20 @@ private void startNewTransport() {
     TransportLogger transportLogger = new TransportLogger();
     // In case the transport logs in the constructor, use the subchannel logId
     transportLogger.logId = getLogId();
-    ConnectionClientTransport transport =
-        new CallTracingTransport(
-            transportFactory
-                .newClientTransport(address, options, transportLogger), callsTracer);
+    ConnectionClientTransport rawTransport;
+    try {
+      rawTransport =
+          transportFactory.newClientTransport(address, options, transportLogger);
+    } catch (IllegalStateException e) {
+      channelLogger.log(
+          ChannelLogLevel.WARNING, "Transport factory is closed, shutting down subchannel", e);
+      shutdown(
+          Status.UNAVAILABLE
+              .withDescription("Transport factory is closed")
+              .withCause(e));
+      return;
+    }
+    ConnectionClientTransport transport = new CallTracingTransport(rawTransport, callsTracer);
     transportLogger.logId = transport.getLogId();
     channelz.addClientSocket(transport);
     pendingTransport = transport;

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

@@ -1671,6 +1671,21 @@ public void subchannelStateChanges_backendServiceFallsBackToResolutionResultAttr
     );
   }
 
+  @Test
+  public void transportFactoryClosed_shutsDownSubchannelSafely() {
+    SocketAddress addr = mock(SocketAddress.class);
+    createInternalSubchannel(addr);
+    assertEquals(IDLE, internalSubchannel.getState());
+
+    when(mockTransportFactory.newClientTransport(any(), any(), any()))
+        .thenThrow(new IllegalStateException("The transport factory is closed."));
+
+    assertNull(internalSubchannel.obtainActiveTransport());
+    assertExactCallbackInvokes(
+        "onStateChange:CONNECTING", "onStateChange:SHUTDOWN", "onTerminated");
+    assertEquals(SHUTDOWN, internalSubchannel.getState());
+  }
+
   private void assertNoCallbackInvoke() {
     while (fakeExecutor.runDueTasks() > 0) {}
     assertEquals(0, callbackInvokes.size());