The reported hang grpc#12109 in blockingUnaryCall was caused by an unhandled framing exception during the draining of DelayedStream. When MessageFramer throws an

  exception (e.g., RESOURCE_EXHAUSTED), it bubbles up through DelayedStream.drainPendingCalls and is eventually caught and swallowed by
  ThreadlessExecutor.runQuietly. This leaves the DelayedStream in an inconsistent state (where passThrough is still false), and the responseFuture never
  completes, causing the blocking call to hang forever.

  I have implemented a fix that adds proper exception handling to the draining loops in both DelayedStream and DelayedClientCall. When an exception occurs
  during draining:
   1. The realStream (or realCall) is explicitly cancelled with the error.
   2. The pending calls are cleared.
   3. The stream/call transitions to passThrough = true to prevent getting stuck.
   4. The listener's pending callbacks are drained, ensuring that any closure notifications are delivered to the application.

Author: kannanjgithub

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

@@ -294,6 +294,7 @@ private void drainPendingCalls() {
     assert !passThrough;
     List<Runnable> toRun = new ArrayList<>();
     DelayedListener<RespT> delayedListener ;
+    drainOut:
     while (true) {
       synchronized (this) {
         if (pendingRunnables.isEmpty()) {
@@ -311,8 +312,18 @@ private void drainPendingCalls() {
       }
       for (Runnable runnable : toRun) {
         // Must not call transport while lock is held to prevent deadlocks.
-        // TODO(ejona): exception handling
-        runnable.run();
+        try {
+          runnable.run();
+        } catch (Throwable t) {
+          Status status = Status.fromThrowable(t).withDescription("Failed to drain pending calls");
+          realCall.cancel(status.getDescription(), status.getCause());
+          synchronized (this) {
+            pendingRunnables = null;
+            passThrough = true;
+            delayedListener = this.delayedListener;
+          }
+          break drainOut;
+        }
       }
       toRun.clear();
     }
@@ -519,6 +530,7 @@ public void run() {
     void drainPendingCallbacks() {
       assert !passThrough;
       List<Runnable> toRun = new ArrayList<>();
+      drainOut:
       while (true) {
         synchronized (this) {
           if (pendingCallbacks.isEmpty()) {
@@ -535,8 +547,15 @@ void drainPendingCallbacks() {
         }
         for (Runnable runnable : toRun) {
           // Avoid calling listener while lock is held to prevent deadlocks.
-          // TODO(ejona): exception handling
-          runnable.run();
+          try {
+            runnable.run();
+          } catch (Throwable t) {
+            synchronized (this) {
+              pendingCallbacks = null;
+              passThrough = true;
+            }
+            throw t;
+          }
         }
         toRun.clear();
       }

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

@@ -172,6 +172,7 @@ private void drainPendingCalls() {
     assert !passThrough;
     List<Runnable> toRun = new ArrayList<>();
     DelayedStreamListener delayedListener = null;
+    drainOut:
     while (true) {
       synchronized (this) {
         if (pendingCalls.isEmpty()) {
@@ -189,8 +190,18 @@ private void drainPendingCalls() {
       }
       for (Runnable runnable : toRun) {
         // Must not call transport while lock is held to prevent deadlocks.
-        // TODO(ejona): exception handling
-        runnable.run();
+        try {
+          runnable.run();
+        } catch (Throwable t) {
+          Status status = Status.fromThrowable(t).withDescription("Failed to drain pending calls");
+          realStream.cancel(status);
+          synchronized (this) {
+            pendingCalls = null;
+            passThrough = true;
+            delayedListener = this.delayedListener;
+          }
+          break drainOut;
+        }
       }
       toRun.clear();
     }
@@ -525,6 +536,7 @@ public void run() {
     public void drainPendingCallbacks() {
       assert !passThrough;
       List<Runnable> toRun = new ArrayList<>();
+      drainOut:
       while (true) {
         synchronized (this) {
           if (pendingCallbacks.isEmpty()) {
@@ -541,8 +553,15 @@ public void drainPendingCallbacks() {
         }
         for (Runnable runnable : toRun) {
           // Avoid calling listener while lock is held to prevent deadlocks.
-          // TODO(ejona): exception handling
-          runnable.run();
+          try {
+            runnable.run();
+          } catch (Throwable t) {
+            synchronized (this) {
+              pendingCallbacks = null;
+              passThrough = true;
+            }
+            throw t;
+          }
         }
         toRun.clear();
       }

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

@@ -20,6 +20,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -229,6 +230,32 @@ public void delayedCallsRunUnderContext() throws Exception {
     assertThat(contextKey.get(readyContext.get())).isEqualTo(goldenValue);
   }
 
+  @Test
+  public void drainPendingCallFails() {
+    DelayedClientCall<String, Integer> delayedClientCall =
+        new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null);
+    delayedClientCall.start(listener, new Metadata());
+    delayedClientCall.request(1);
+
+    final RuntimeException error = new RuntimeException("fail");
+    org.mockito.Mockito.doAnswer(new org.mockito.stubbing.Answer<Void>() {
+      @Override
+      public Void answer(org.mockito.invocation.InvocationOnMock invocation) {
+        throw error;
+      }
+    }).when(mockRealCall).request(1);
+
+    Runnable runnable = delayedClientCall.setCall(mockRealCall);
+    assertThat(runnable).isNotNull();
+    try {
+      runnable.run();
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+
+    verify(mockRealCall).cancel(eq("Failed to drain pending calls"), same(error));
+  }
+
   private void callMeMaybe(Runnable r) {
     if (r != null) {
       r.run();

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

@@ -472,6 +472,36 @@ public Void answer(InvocationOnMock in) {
         .matches("\\[test_op_delay=[0-9]+ns, remote_addr=127\\.0\\.0\\.1:443\\]");
   }
 
+  @Test
+  public void drainPendingCallFails() {
+    stream.start(listener);
+    stream.request(1);
+    final RuntimeException error = new RuntimeException("fail");
+    doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        throw error;
+      }
+    }).when(realStream).request(1);
+
+    Runnable runnable = stream.setStream(realStream);
+    assertNotNull(runnable);
+    try {
+      runnable.run();
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+
+    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
+    verify(realStream).cancel(statusCaptor.capture());
+    assertThat(statusCaptor.getValue().getCode()).isEqualTo(Status.Code.UNKNOWN);
+    assertThat(statusCaptor.getValue().getCause()).isSameInstanceAs(error);
+
+    verify(realStream).start(listenerCaptor.capture());
+    listenerCaptor.getValue().closed(statusCaptor.getValue(), RpcProgress.PROCESSED, new Metadata());
+    verify(listener).closed(same(statusCaptor.getValue()), any(RpcProgress.class), any(Metadata.class));
+  }
+
   private void callMeMaybe(Runnable r) {
     if (r != null) {
       r.run();