Synchronously mark the call as cancelled by calling closedInternal to avoid races with more messages or halfClose after the error scenario was observed.

Author: kannanjgithub

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

@@ -355,6 +355,7 @@ private void messagesAvailableInternal(final MessageProducer producer) {
               call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
               GrpcUtil.closeQuietly(delayedMessage);
               delayedMessage = null;
+              closedInternal(Status.INTERNAL.withDescription("Too many requests"));
               return;
             }
             try {

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

@@ -573,6 +573,8 @@ public void streamListener_messageRead_unary_tooManyRequests() {
 
     verify(stream).cancel(any(Status.class));
     verify(callListener, never()).onMessage(any(Long.class));
+    verify(callListener).onCancel();
+    assertTrue(context.isCancelled());
   }
 
   @Test