Detect transport executors with no remaining threads

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -499,6 +499,29 @@ public Runnable start(Listener listener) {
       outboundFlow = new OutboundFlowController(this, frameWriter);
     }
     final CountDownLatch latch = new CountDownLatch(1);
+    // This runs con-concurrently with handshake and works as a hack checking enough threads are
+    // available to start the transport.
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        long waitStartTime = System.nanoTime();
+        synchronized (lock) {
+          try {
+            lock.wait(100);
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
+        }
+        long waitEndTime = System.nanoTime();
+        System.out.println(waitEndTime - waitStartTime);
+        if (waitEndTime - waitStartTime >= 100000000) { // never got notified
+          startGoAway(0, ErrorCode.INTERNAL_ERROR, Status.UNAVAILABLE
+              .withDescription("Timed out waiting for second handshake thread. "
+                  + "The transport executor pool may have run out of threads"));
+        }
+      }
+    });
+
     // Connecting in the serializingExecutor, so that some stream operations like synStream
     // will be executed after connected.
     serializingExecutor.execute(new Runnable() {
@@ -581,6 +604,7 @@ sslSocketFactory, hostnameVerifier, sock, getOverridenHost(), getOverridenPort()
           if (sslSession != null) {
             securityInfo = new InternalChannelz.Security(new InternalChannelz.Tls(sslSession));
           }
+          lock.notify();
         }
       }
     });

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientTransportTest.java

@@ -247,6 +247,19 @@ public void testToString() throws Exception {
     assertTrue("Unexpected: " + s, s.contains(address.toString()));
   }
 
+  @Test
+  public void testTransportExecutorWithTooFewThreads() throws Exception {
+    ExecutorService fixedPoolExecutor = Executors.newFixedThreadPool(1);
+    channelBuilder.transportExecutor(fixedPoolExecutor);
+    initTransport();
+    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
+    verify(transportListener, timeout(TIME_OUT_MS)).transportShutdown(statusCaptor.capture());
+    Status capturedStatus = statusCaptor.getValue();
+    assertEquals("Timed out waiting for second handshake thread. "
+        + "The transport executor pool may have run out of threads",
+        capturedStatus.getDescription());
+  }
+
   /**
    * Test logging is functioning correctly for client received Http/2 frames. Not intended to test
    * actual frame content being logged.