core: Close InputStream in NoopClientStream.writeMessage() to prevent resource leaks

NoopClientStream.writeMessage() silently discards the InputStream without
closing it. When a Marshaller.stream() returns an InputStream backed by a
ref-counted ByteBuf (e.g. Netty's PooledByteBufAllocator), this causes a
direct memory leak.

This affects any code path where writeMessage() is called on a
NoopClientStream or its subclass FailingClientStream:
- Context cancelled before stream start (ClientCallImpl line 197)
- Compressor not found (ClientCallImpl line 219)
- Deadline already exceeded (ClientCallImpl line 262, via FailingClientStream)
- DelayedStream draining buffered messages after cancellation sets
  realStream to NoopClientStream.INSTANCE

The fix calls GrpcUtil.closeQuietly(message) to ensure the InputStream is
always closed, matching the contract in AbstractStream.writeMessage().

Author: merlimat

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

@@ -45,7 +45,9 @@ public Attributes getAttributes() {
   public void request(int numMessages) {}
 
   @Override
-  public void writeMessage(InputStream message) {}
+  public void writeMessage(InputStream message) {
+    GrpcUtil.closeQuietly(message);
+  }
 
   @Override
   public void flush() {}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.io.InputStream;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Unit tests for {@link NoopClientStream}.
+ */
+@RunWith(JUnit4.class)
+public class NoopClientStreamTest {
+
+  @Test
+  public void writeMessageShouldCloseInputStream() throws Exception {
+    // NoopClientStream.writeMessage() is called when a stream is cancelled or failed
+    // before the real transport stream is established (e.g. via DelayedStream draining
+    // buffered messages to NoopClientStream on cancellation, or FailingClientStream
+    // which extends NoopClientStream). The InputStream must be closed to avoid leaking
+    // resources such as ref-counted ByteBufs.
+    InputStream message = mock(InputStream.class);
+    NoopClientStream.INSTANCE.writeMessage(message);
+    verify(message).close();
+  }
+}