test: add unit tests for listener drain exception handling in DelayedStream and DelayedClientCall

kannanjgithub · kannanjgithub · commit 1d5b58be9148 · 2026-03-30T10:51:59.000Z
diff --git a/core/src/test/java/io/grpc/internal/DelayedClientCallTest.java b/core/src/test/java/io/grpc/internal/DelayedClientCallTest.java
@@ -23,6 +23,7 @@
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import com.google.common.util.concurrent.MoreExecutors;
@@ -256,6 +257,55 @@ public Void answer(org.mockito.invocation.InvocationOnMock invocation) {
     verify(mockRealCall).cancel(eq("Failed to drain pending calls"), same(error));
   }
 
+  @Test
+  public void drainPendingCallbacksFails() {
+    DelayedClientCall<String, Integer> delayedClientCall =
+        new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null);
+    delayedClientCall.start(listener, new Metadata());
+
+    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(listener).onReady();
+
+    final AtomicReference<ClientCall.Listener<Integer>> listenerCaptor = new AtomicReference<>();
+    org.mockito.Mockito.doAnswer(new org.mockito.stubbing.Answer<Void>() {
+      @Override
+      public Void answer(org.mockito.invocation.InvocationOnMock invocation) {
+        ClientCall.Listener<Integer> delayedListener = invocation.getArgument(0);
+        listenerCaptor.set(delayedListener);
+        delayedListener.onReady();
+        return null;
+      }
+    }).when(mockRealCall).start(any(ClientCall.Listener.class), any(Metadata.class));
+
+    Runnable runnable = delayedClientCall.setCall(mockRealCall);
+    assertThat(runnable).isNotNull();
+
+    try {
+      runnable.run();
+      org.junit.Assert.fail("Should have thrown");
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+
+    ClientCall.Listener<Integer> delayedListener = listenerCaptor.get();
+    assertThat(delayedListener).isNotNull();
+
+    // Verify it transitioned to passThrough by showing it forwards.
+    try {
+      delayedListener.onReady();
+      org.junit.Assert.fail("Should have thrown");
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+    // Verify it was called twice (once during drain, once just now)
+    verify(listener, times(2)).onReady();
+  }
+
   private void callMeMaybe(Runnable r) {
     if (r != null) {
       r.run();
diff --git a/core/src/test/java/io/grpc/internal/DelayedStreamTest.java b/core/src/test/java/io/grpc/internal/DelayedStreamTest.java
@@ -502,6 +502,51 @@ public Void answer(InvocationOnMock invocation) {
     verify(listener).closed(same(statusCaptor.getValue()), any(RpcProgress.class), any(Metadata.class));
   }
 
+  @Test
+  public void drainPendingCallbacksFails() {
+    stream.start(listener);
+    final RuntimeException error = new RuntimeException("fail");
+    doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        throw error;
+      }
+    }).when(listener).onReady();
+
+    doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        ClientStreamListener delayedListener = invocation.getArgument(0);
+        delayedListener.onReady();
+        return null;
+      }
+    }).when(realStream).start(any(ClientStreamListener.class));
+
+    Runnable runnable = stream.setStream(realStream);
+    assertNotNull(runnable);
+
+    try {
+      runnable.run();
+      org.junit.Assert.fail("Should have thrown");
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+
+    verify(realStream).start(listenerCaptor.capture());
+    ClientStreamListener delayedListener = listenerCaptor.getValue();
+
+    // Verify it transitioned to passThrough. If it didn't, this might NPE or buffer.
+    // If it is passThrough, it will forward to the listener, which we know throws.
+    try {
+      delayedListener.onReady();
+      org.junit.Assert.fail("Should have thrown");
+    } catch (RuntimeException e) {
+      assertThat(e).isSameInstanceAs(error);
+    }
+    // Verify it was called twice (once during drain, once just now)
+    verify(listener, times(2)).onReady();
+  }
+
   private void callMeMaybe(Runnable r) {
     if (r != null) {
       r.run();
