[merlimat]

Problem

Since 0.9.0, closing the InputStream returned by a LightProto gRPC Marshaller.stream() destroys its content even if nobody has read it. DrainableByteArrayInputStream.close() empties the stream and returns its array to the per-thread pool (#27). GatheringInputStream.close() empties it and releases its ByteBufs (#49).

This breaks proxies that forward the stream they receive unchanged, like Apache BookKeeper's stream-storage proxy. gRPC's ClientCallImpl closes each message as soon as onMessage() returns. Over the in-process transport, the next hop gets that same stream object and reads it later on its own executor, so it sees 0 bytes. BookKeeper's TestStorageContainerStoreImpl gets empty responses, and ProxyPingPongServiceTest hangs. Up to 0.7.x, close() was a no-op.

Example

ServerCall.Listener<InputStream> forward(ServerCall<InputStream, InputStream> serverCall, ...) {
    ...
    public void onMessage(InputStream message) {   // ClientCall.Listener on the backend call
        serverCall.sendMessage(message);           // in-process: the peer reads it later
    }                                              // gRPC then calls message.close()
}

Current generated code:

public void close() {                 // DrainableByteArrayInputStream
    byte[] a = this.buf;
    if (a.length == 0) return;
    this.buf = CLOSED;                // the peer now reads 0 bytes
    this.pos = this.count = this.mark = 0;
    if (a.length <= SPARE_ARRAY_MAX) SPARE_ARRAY.set(a);  // ...and the next message overwrites the array
}

Change

A stream only gives up its array or buffers once it has been read to the end. Closing it any earlier keeps the content readable, and a later reader's close does the recycling. Further closes after that stay no-ops.

DrainableByteArrayInputStream: an early close does nothing, and the array is left to the garbage collector.

if (a.length == 0 || this.pos < this.count) {
    return;
}

GatheringInputStream: an early close copies what is left into an array the stream owns. It then releases the retained ByteBufs and gives the thread's array back, so nothing leaks if the stream is never read again. Later reads come from the copy. The methods are synchronized, because the proxy's close can run on one thread while the peer reads on another. Without that, the proxy test below corrupts every message with a value of 16 KiB or more.

public synchronized void close() {
    byte[] a = this.buf;
    if (a.length == 0 || (remaining > 0 && count == 0)) return;  // handed back, or already copied
    if (remaining == 0) {
        this.buf = CLOSED;
    } else {
        byte[] rest = new byte[remaining];
        read(rest, 0, rest.length);
        this.buf = rest;
        this.length = this.remaining = rest.length;
        this.piece = this.offset = 0;
    }
    // release bufs[k], clear arrays[k]; count = 0; hand back a
}

The steady-state path is unchanged: when Netty drains a stream and then closes it, the array is recycled with no extra allocation or copy. One behaviour change: a send that fails before gRPC reads the stream (it exceeds the size limit, or a queued call is cancelled) no longer returns its small-message array, because that close looks the same as a proxy's.

Testing

• New GrpcProxyTest: an in-process backend, an identity-marshaller proxy modelled on BookKeeper's ProxyCall, and an in-process client, with no directExecutor() anywhere. It covers unary and server-streaming calls with 100 B, 16 KiB and 2 MiB payloads, and checks both what the backend received and what the client got. All 6 cases fail on 0.9.2 and pass with this change.
• New GrpcMarshallerEarlyCloseTest, on both the small-message and gathering paths:
  • close then parse() the same stream;
  • drain then close, and close then drain then close, with ByteBuf refcounts checked back at their starting values;
  • close after a partial read;
  • the caller's buffer may change after an early close;
  • close while another thread reads.
• Updated three existing tests that asserted the old empty-on-close behaviour.
• mvn verify from the root passes, including all gRPC and leak tests.
• Checked against BookKeeper with this build installed as a 0.9.3-SNAPSHOT (JDK 17): ProxyPingPongServiceTest 5/5, DirectPingPongServiceTest 5/5, TestStorageContainerStoreImpl 20/20.