Address review comments.

- Use GrpcUtil.closeQuietly to close the buffered message in halfClosed() to prevent
  unnecessary exception propagation if close fails after successful message delivery.
- Use stream.cancel instead of call.close when detecting too many requests for unary
  calls. This ensures the transport is notified to abort the stream (sending RST_STREAM)
  and immediately releases resources, preventing leaks from clients that withhold END_STREAM.

Author: kannanjgithub

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

@@ -352,9 +352,7 @@ private void messagesAvailableInternal(final MessageProducer producer) {
           if (call.method.getType().clientSendsOneMessage()) {
             if (delayedMessage != null) {
               GrpcUtil.closeQuietly(message);
-              call.close(
-                  Status.INTERNAL.withDescription("Too many requests"),
-                  new Metadata());
+              call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
               GrpcUtil.closeQuietly(delayedMessage);
               delayedMessage = null;
               return;
@@ -401,11 +399,7 @@ public void halfClosed() {
             Throwables.throwIfUnchecked(t);
             throw new RuntimeException(t);
           }
-          try {
-            message.close();
-          } catch (IOException e) {
-            throw new RuntimeException(e);
-          }
+          GrpcUtil.closeQuietly(message);
         }
 
         listener.onHalfClose();

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

@@ -571,7 +571,7 @@ public void streamListener_messageRead_unary_tooManyRequests() {
     // Sending second message should fail
     streamListener.messagesAvailable(new SingleMessageProducer(UNARY_METHOD.streamRequest(5678L)));
 
-    verify(stream).close(any(Status.class), any(Metadata.class));
+    verify(stream).cancel(any(Status.class));
     verify(callListener, never()).onMessage(any(Long.class));
   }
 
@@ -628,14 +628,12 @@ public void streamListener_halfClosed_closeException() {
     // Message should not be delivered yet
     verify(callListener, never()).onMessage(any(Long.class));
 
-    // halfClosed should throw RuntimeException wrapping IOException
-    RuntimeException e = assertThrows(RuntimeException.class,
-        () -> streamListener.halfClosed());
-    assertThat(e).hasCauseThat().isInstanceOf(IOException.class);
-    assertThat(e.getCause()).hasMessageThat().isEqualTo("close failed");
+    // halfClosed should not throw because we use closeQuietly
+    streamListener.halfClosed();
 
-    // The message was delivered before close failed
+    // The message was delivered and halfClosed completed
     verify(callListener).onMessage(1234L);
+    verify(callListener).onHalfClose();
     assertTrue(detachableStream.detachedStream.closed);
   }