From 90caf3941d71fe1b77725a41262feb0e149e05f5 Mon Sep 17 00:00:00 2001
From: Larry Safran <lsafran@google.com>
Date: Mon, 16 Sep 2024 15:03:41 -0700
Subject: [PATCH] Fixed hang from not getting interrupted. Added some 'final'
 to class definitions.

---
 stub/src/main/java/io/grpc/stub/ClientCalls.java | 16 ++++++++++++----
 .../test/java/io/grpc/stub/ClientCallsTest.java  |  2 +-
 2 files changed, 13 insertions(+), 5 deletions(-)

diff --git a/stub/src/main/java/io/grpc/stub/ClientCalls.java b/stub/src/main/java/io/grpc/stub/ClientCalls.java
index 79fab9425bb..2bc5e1f785f 100644
--- a/stub/src/main/java/io/grpc/stub/ClientCalls.java
+++ b/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
diff --git a/stub/src/test/java/io/grpc/stub/ClientCallsTest.java b/stub/src/test/java/io/grpc/stub/ClientCallsTest.java
index f3d101b862a..b711b2a23b5 100644
--- a/stub/src/test/java/io/grpc/stub/ClientCallsTest.java
+++ b/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();
         }
       };
     }