stub: drain pending reads on cancel to release their buffers

stub/src/main/java/io/grpc/stub/BlockingClientCall.java

@@ -241,14 +241,19 @@ void sendSingleRequest(ReqT request) {
 
   /**
    * Cancel stream and stop any further writes.  Note that some reads that are in flight may still
-   * happen after the cancel.
+   * happen after the cancel.  Reads that have been received but not yet delivered are processed
+   * when cancelling, releasing their resources.
    *
    * @param message if not {@code null}, will appear as the description of the CANCELLED status
    * @param cause if not {@code null}, will appear as the cause of the CANCELLED status
    */
   public void cancel(String message, Throwable cause) {
     writeClosed = true;
     call.cancel(message, cause);
+    // Reads queued by the transport hold buffers that are only released when their delivery task
+    // runs and observes the cancelled stream.  Nothing else will drain the executor once the user
+    // abandons the call, so the queued tasks must be processed here.
+    executor.drain();
   }
 
   /**
@@ -278,6 +283,11 @@ Status getClosedStatus() {
     return (state == null) ? null : state.status;
   }
 
+  @VisibleForTesting
+  ThreadSafeThreadlessExecutor getExecutor() {
+    return executor;
+  }
+
   /**
    * Check for whether some action is ready.
    *

stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java

@@ -221,6 +221,23 @@ public void testCancel() throws Exception {
 
   }
 
+  @Test
+  public void testCancelDrainsPendingReads() throws Exception {
+    biDiStream = ClientCalls.blockingBidiStreamingCall(channel, BIDI_STREAMING_METHOD,
+        CallOptions.DEFAULT);
+
+    // The server delivering a message queues a read-delivery task in the call executor, which
+    // only runs when the user reads/writes or the call is cancelled
+    testMethod.sendValueToClient(60);
+    assertThat(biDiStream.getExecutor()).isNotEmpty();
+
+    biDiStream.cancel("done reading", null);
+
+    // The queued read task must be processed by cancel() itself; when orphaned it holds the
+    // message's transport buffers, which leak (see #12355)
+    assertThat(biDiStream.getExecutor()).isEmpty();
+  }
+
   @Test
   public void testIsActivityReady() throws Exception {
     biDiStream = ClientCalls.blockingBidiStreamingCall(channel,  BIDI_STREAMING_METHOD,