core: logging the error message when onClose() itself fails (#11880)

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

@@ -561,7 +561,11 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT> {
   }
 
   private void closeObserver(Listener<RespT> observer, Status status, Metadata trailers) {
-    observer.onClose(status, trailers);
+    try {
+      observer.onClose(status, trailers);
+    } catch (RuntimeException ex) {
+      log.log(Level.WARNING, "Exception thrown by onClose() in ClientCall", ex);
+    }
   }
 
   @Override

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

@@ -1105,6 +1105,32 @@ public class ClientCallImplTest {
     assertEquals(attrs, call.getAttributes());
   }
 
+  @Test
+  public void onCloseExceptionCaughtAndLogged() {
+    DelayedExecutor executor = new DelayedExecutor();
+    ClientCallImpl<Void, Void> call = new ClientCallImpl<>(
+        method,
+        executor,
+        baseCallOptions,
+        clientStreamProvider,
+        deadlineCancellationExecutor,
+        channelCallTracer, configSelector);
+
+    call.start(callListener, new Metadata());
+    verify(stream).start(listenerArgumentCaptor.capture());
+    final ClientStreamListener streamListener = listenerArgumentCaptor.getValue();
+    streamListener.headersRead(new Metadata());
+
+    doThrow(new RuntimeException("Exception thrown by onClose() in ClientCall")).when(callListener)
+        .onClose(any(Status.class), any(Metadata.class));
+
+    Status status = Status.RESOURCE_EXHAUSTED.withDescription("simulated");
+    streamListener.closed(status, PROCESSED, new Metadata());
+    executor.release();
+
+    verify(callListener).onClose(same(status), any(Metadata.class));
+  }
+
   private static final class DelayedExecutor implements Executor {
     private final BlockingQueue<Runnable> commands = new LinkedBlockingQueue<>();