Release queued body token buffers on multipart cancel

[cj848]

MultipartParser emits multiple tokens per input DataBuffer (a headers token followed by body tokens), regardless of downstream demand. Tokens emitted beyond the demand are buffered in the Flux.create sink queue, and downstream operators used by DefaultPartHttpMessageReader / PartEventHttpMessageReader (windowUntil, concatMap) keep their own token queues as well.

When the subscriber cancels while such tokens are queued — for example when the client aborts the connection mid-upload — Reactor discards the queued tokens. BodyToken is not a DataBuffer, so no existing doOnDiscard(DataBuffer.class, ...) hook can reach the buffer it wraps, and the buffer is dropped without being released. With Netty this surfaces in production as:

LEAK: ByteBuf.release() was not called before it's garbage-collected.

with leak-detector access records ending in MultipartParser$BodyState.onNext / DataBufferUtils$AbstractNestedMatcher.match and no matching release. This is related to, but distinct from, gh-36262: that fix covered PartGenerator, while this leak happens one stage earlier, in the parser token stream itself.

This PR registers a doOnDiscard(BodyToken.class, ...) hook in MultipartParser.parse(). Since the discard handler is propagated upstream through the subscriber context, a single hook at this convergence point covers the sink queue as well as any downstream operator queue that supports discarding, for both Part and PartEvent reading.

The new MultipartParserTests.cancelWithQueuedBodyTokensReleasesBuffers test reproduces the leak deterministically (no race): a whole multipart message in a single input buffer with a downstream demand of 1 leaves the body token in the sink queue at cancellation time. With the fix reverted, the test fails with 1 buffer leaks detected; with the fix applied, it passes.

:spring-web:test --tests "org.springframework.http.codec.multipart.*", :spring-web:checkstyleMain and :spring-web:checkstyleTest pass locally.

[cj848]

I pushed a second commit to this PR, because the discard hook added here exposes a related ownership defect in the same class.

BodyState.flush() emits every queued buffer and clears the queue only afterwards:

private void flush() {
    for (Iterator<DataBuffer> iterator = this.queue.iterator(); iterator.hasNext(); ) {
        DataBuffer buffer = iterator.next();
        boolean last = !iterator.hasNext();
        MultipartParser.this.emitBody(buffer, last);   // ownership handed to the sink
    }
    this.queue.clear();                                // removed from the queue only now
}

@Override
public void dispose() {
    this.queue.forEach(DataBufferUtils::release);
    this.queue.clear();
}

If a cancellation arrives while flush() is emitting, onSinkCancel() disposes the state and releases buffers that were already handed to the sink. Those buffers are released a second time — by the consumer that received them, or by the discard hook this PR adds. enqueue() does not have this problem because it calls iterator.remove() before emitting; only flush() is inconsistent.

With Netty this is not just a log line: body buffers are slices of the inbound buffer, so the extra release frees the inbound buffer prematurely, and reactor-netty then fails when releasing its own share right after onNext:

IllegalReferenceCountException: refCnt: 0, decrement: 1
  io.netty.handler.codec.http.DefaultHttpContent.release
  io.netty.util.ReferenceCountUtil.release
  reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:300)

We observe this in production on a WebFlux service (1–5 occurrences per day under PartEvent uploads with client aborts).

The fix removes each buffer from the queue before emitting it, so dispose() can only release buffers the parser still owns:

private void flush() {
    DataBuffer buffer;
    while ((buffer = this.queue.poll()) != null) {
        MultipartParser.this.emitBody(buffer, this.queue.isEmpty());
    }
}

The added test cancelWhileEmittingBodyTokensKeepsEmittedBuffersAllocated cancels while the parser emits queued body tokens and asserts that the buffers already delivered are still allocated. Without the fix it fails with UnpooledSlicedByteBuf(freed); with the fix MultipartParserTests passes.

Happy to split this into a separate PR if you prefer to keep the two changes independent.