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

[merlimat]

Summary

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

This affects any code path where writeMessage() is called on NoopClientStream or its subclass FailingClientStream:

• Context already cancelled at start() time (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().

Reproducer

We observed this with Oxia-java using LightProto-generated gRPC marshallers. When a standalone server is restarted, Netty's ResourceLeakDetector reports leaked ByteBuf instances:

ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected.
Created at:
    io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:410)
    ...
    io.grpc.MethodDescriptor.streamRequest(MethodDescriptor.java:296)
    io.grpc.internal.ClientCallImpl.sendMessageInternal(ClientCallImpl.java:525)

Test plan

• Added NoopClientStreamTest.writeMessageShouldCloseInputStream() that verifies close() is called on the InputStream passed to writeMessage()

[ejona86]

Yeah, we wouldn't notice this with grpc-java's marshallers, but it should be closed.

[ejona86]

    io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:410)
    ...
    io.grpc.MethodDescriptor.streamRequest(MethodDescriptor.java:296)

I see a direct buffer was allocated as part of streamRequest(). gRPC will just copy out from that buffer again, so I suspect direct buffers don't actually help there. In ProtoInputStream we delay serializing so we can coordinate with the transport's buffering approach (e.g., pooled ByteBuf). However, it seems we've not optimized this in MessageFramer to avoid copying twice. This is related to 5a7f350 . It feels like we should be able to do something better here, although it'd take some thought and may not be clear what is "best."

[merlimat]

Yes, I've noticed that it was all copied into a byte[] and then written into the OutputStream.
I've also tried to use a Netty heap pooled buffer to avoid the first copy and allocation, though it still had the issue of the leak with failures.

It would be great if there could be some sort of shortcut when the InputStream is backed by a ByteBuf that could avoid all the extra copies :)

[ejona86]

Thank you!