Fixed hang from not getting interrupted.

Added some 'final' to class definitions.

Author: larry-safran

stub/src/main/java/io/grpc/stub/ClientCalls.java

@@ -235,7 +235,7 @@ public static <ReqT, RespT> Iterator<RespT> blockingServerStreamingCall(
    * @throws StatusException if the write to the server failed
    */
   @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10918")
-  public static <ReqT, RespT> BlockingClientCall<?, RespT> blockingV2ServerStreamingCall(
+  public static <ReqT, RespT> BlockingClientCall<ReqT, RespT> blockingV2ServerStreamingCall(
       Channel channel, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, ReqT req)
       throws InterruptedException, StatusException {
     BlockingClientCall<ReqT, RespT> call =
@@ -436,7 +436,7 @@ private abstract static class StartableListener<T> extends ClientCall.Listener<T
     abstract void onStart();
   }
 
-  private static class CallToStreamObserverAdapter<ReqT>
+  private static final class CallToStreamObserverAdapter<ReqT>
       extends ClientCallStreamObserver<ReqT> {
     private boolean frozen;
     private final ClientCall<ReqT, ?> call;
@@ -787,7 +787,7 @@ void onStart() {
   }
 
   @SuppressWarnings("serial")
-  static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runnable>
+  private static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runnable>
       implements Executor {
     private static final Logger log = Logger.getLogger(ThreadlessExecutor.class.getName());
 
@@ -804,12 +804,14 @@ static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runnable>
      * Must only be called by one thread at a time.
      */
     public void waitAndDrain() throws InterruptedException {
+      throwIfInterrupted();
       Runnable runnable = poll();
       if (runnable == null) {
         waiter = Thread.currentThread();
         try {
           while ((runnable = poll()) == null) {
             LockSupport.park(this);
+            throwIfInterrupted();
           }
         } finally {
           waiter = null;
@@ -820,6 +822,12 @@ public void waitAndDrain() throws InterruptedException {
       } while ((runnable = poll()) != null);
     }
 
+    private static void throwIfInterrupted() throws InterruptedException {
+      if (Thread.interrupted()) {
+        throw new InterruptedException();
+      }
+    }
+
     /**
      * Called after final call to {@link #waitAndDrain()}, from same thread.
      */
@@ -857,7 +865,7 @@ static final class ThreadSafeThreadlessExecutor extends ConcurrentLinkedQueue<Ru
     private static final Logger log =
         Logger.getLogger(ThreadSafeThreadlessExecutor.class.getName());
 
-    private Lock waiterLock = new ReentrantLock();
+    private final Lock waiterLock = new ReentrantLock();
     private final Condition waiterCondition = waiterLock.newCondition();
 
     // Non private to avoid synthetic class

stub/src/test/java/io/grpc/stub/ClientCallsTest.java

@@ -971,8 +971,8 @@ public <ReqT,RespT> ClientCall<ReqT, RespT> interceptCall(
         }
 
         @Override public void halfClose() {
-          Thread.currentThread().interrupt();
           super.halfClose();
+          Thread.currentThread().interrupt();
         }
       };
     }