[siddiqueirshad]

Fixes #4332

Problem Description

In NettyWriteResponseFilter, the response body stream is created with:

final Flux<DataBuffer> body = connection
        .inbound()
        .receive()
        .retain()
        .map(byteBuf -> wrap(byteBuf, response));
return response.writeWith(body);

receive().retain() increments the reference count of inbound Netty ByteBuf slices. When a client cancels or disconnects mid-response (e.g. mobile/browser connection drop), cleanup() only disposes the connection. Elements in flight between retain() and the outbound response write—including buffers buffered or queued in Reactor operators—are not released, leading to direct memory leak (LEAK: ByteBuf.release() was not called before it's garbage-collected) and pinning PoolChunks.

Solution

Add doOnDiscard handlers to ensure discarded buffers are released upon cancellation or error:

1. doOnDiscard(ByteBuf.class, ReferenceCountUtil::safeRelease) immediately following retain() to handle discarded raw ByteBuf instances.
2. doOnDiscard(DataBuffer.class, ...) following map(...) to safely release discarded wrapped DataBuffer instances (catching Throwable to safely tolerate potential concurrent release with FluxReceive.cleanQueue).

[spencergibb]

Please add a test if possible

[siddiqueirshad]

Hi @spencergibb, added unit test
etainedBuffersAreReleasedOnDiscard\ in \NettyWriteResponseFilterTests\ verifying that retained ByteBuf and DataBuffer instances are safely released when downstream cancels or discards in-flight buffers, and signed off the commit.