From 93beb6bef7d52f1b0d3a6d4063539d71129ea084 Mon Sep 17 00:00:00 2001 From: Hyunsik Kang Date: Thu, 6 Aug 2026 07:38:17 +0900 Subject: [PATCH 1/2] Release queued body token buffers on multipart cancel When a multipart subscriber cancels while MultipartParser has already emitted body tokens beyond the downstream demand, those tokens are held in the Flux.create sink queue (and in downstream operator queues such as windowUntil). On cancellation, Reactor discards the queued tokens, but BodyToken is not a DataBuffer, so the buffers inside the discarded tokens are never released and Netty reports "LEAK: ByteBuf.release() was not called before it's garbage-collected". Register a doOnDiscard hook for BodyToken in MultipartParser.parse() so that a discarded body token releases its buffer, both in the sink queue and in any downstream operator queue that supports discarding. Closes gh-37115 Signed-off-by: Hyunsik Kang --- .../http/codec/multipart/MultipartParser.java | 4 +- .../codec/multipart/MultipartParserTests.java | 72 +++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) create mode 100644 spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java index b3c8d1a10dba..a286ba7283ea 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java @@ -96,12 +96,12 @@ private MultipartParser(FluxSink sink, byte[] boundary, int maxHeadersSiz */ public static Flux parse(Flux buffers, byte[] boundary, int maxHeadersSize, Charset headersCharset) { - return Flux.create(sink -> { + return Flux.create(sink -> { MultipartParser parser = new MultipartParser(sink, boundary, maxHeadersSize, headersCharset); sink.onCancel(parser::onSinkCancel); sink.onRequest(l -> parser.requestBuffer()); buffers.subscribe(parser); - }); + }).doOnDiscard(BodyToken.class, body -> DataBufferUtils.release(body.buffer())); } @Override diff --git a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java new file mode 100644 index 000000000000..1d9a5e5d17dd --- /dev/null +++ b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java @@ -0,0 +1,72 @@ +/* + * Copyright 2002-present the original author or 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 + * + * https://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 org.springframework.http.codec.multipart; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; + +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.testfixture.io.buffer.AbstractLeakCheckingTests; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link MultipartParser}. + * + * @author Hyunsik Kang + */ +class MultipartParserTests extends AbstractLeakCheckingTests { + + @Test // gh-37115 + void cancelWithQueuedBodyTokensReleasesBuffers() { + byte[] boundary = "simple-boundary".getBytes(UTF_8); + String content = "--simple-boundary\r\nContent-Type: text/plain\r\n\r\n" + + "a".repeat(1024) + "\r\n--simple-boundary--\r\n"; + byte[] bytes = content.getBytes(UTF_8); + DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length); + buffer.write(bytes); + + Flux tokens = MultipartParser.parse(Flux.just(buffer), boundary, 8192, UTF_8); + + List received = new ArrayList<>(); + BaseSubscriber subscriber = new BaseSubscriber<>() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + request(1); + } + @Override + protected void hookOnNext(MultipartParser.Token token) { + received.add(token); + } + }; + tokens.subscribe(subscriber); + // Flux.just delivers synchronously, so by now the parser has emitted the headers + // token and all body tokens; the body tokens beyond the requested demand of 1 are + // held in the Flux.create sink queue. Cancelling discards that queue, and the + // buffers inside the discarded body tokens must be released. + subscriber.cancel(); + + assertThat(received).singleElement().isInstanceOf(MultipartParser.HeadersToken.class); + } + +} From bb441b86cf3497ecf002387a0cfd12ee19ffb8fc Mon Sep 17 00:00:00 2001 From: Hyunsik Kang Date: Thu, 13 Aug 2026 07:53:10 +0900 Subject: [PATCH 2/2] Do not release body buffers already handed to the sink BodyState.flush() emits every queued buffer and only clears the queue afterwards, so a cancellation arriving while it emits makes dispose() release buffers whose ownership has already been transferred to the sink. Such a buffer is then released twice: once by the parser, and once by the downstream consumer or the discard hook. With Netty, body buffers are slices of the inbound buffer, so the second release frees the inbound buffer prematurely, which surfaces as IllegalReferenceCountException: refCnt: 0, decrement: 1 io.netty.handler.codec.http.DefaultHttpContent.release reactor.netty.channel.FluxReceive.drainReceiver when reactor-netty releases its own share right after onNext. Remove each buffer from the queue before emitting it, mirroring what enqueue() already does, so that dispose() only ever releases buffers the parser still owns. --- .../http/codec/multipart/MultipartParser.java | 14 ++++--- .../codec/multipart/MultipartParserTests.java | 41 +++++++++++++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java index a286ba7283ea..24df791d90f9 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java @@ -602,13 +602,17 @@ private void enqueue(DataBuffer buf) { emit.forEach(buffer -> MultipartParser.this.emitBody(buffer, false)); } + /** + * Emit all queued buffers, removing each from the queue before emitting it so + * that {@link #dispose()} cannot release a buffer that was already handed over + * to the sink. A cancellation arriving while this method emits would otherwise + * release such a buffer a second time. + */ private void flush() { - for (Iterator iterator = this.queue.iterator(); iterator.hasNext(); ) { - DataBuffer buffer = iterator.next(); - boolean last = !iterator.hasNext(); - MultipartParser.this.emitBody(buffer, last); + DataBuffer buffer; + while ((buffer = this.queue.poll()) != null) { + MultipartParser.this.emitBody(buffer, this.queue.isEmpty()); } - this.queue.clear(); } @Override diff --git a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java index 1d9a5e5d17dd..6cd79714b3a0 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java @@ -25,6 +25,8 @@ import reactor.core.publisher.Flux; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferUtils; +import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.core.testfixture.io.buffer.AbstractLeakCheckingTests; import static java.nio.charset.StandardCharsets.UTF_8; @@ -69,4 +71,43 @@ protected void hookOnNext(MultipartParser.Token token) { assertThat(received).singleElement().isInstanceOf(MultipartParser.HeadersToken.class); } + @Test // gh-37115 + void cancelWhileEmittingBodyTokensKeepsEmittedBuffersAllocated() { + byte[] boundary = "simple-boundary".getBytes(UTF_8); + String content = "--simple-boundary\r\nContent-Type: text/plain\r\n\r\n" + + "a".repeat(1024) + "\r\n" + + "--simple-boundary\r\nContent-Type: text/plain\r\n\r\n" + + "b".repeat(1024) + "\r\n--simple-boundary--\r\n"; + byte[] bytes = content.getBytes(UTF_8); + DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length); + buffer.write(bytes); + + Flux tokens = MultipartParser.parse(Flux.just(buffer), boundary, 8192, UTF_8); + + List receivedBuffers = new ArrayList<>(); + BaseSubscriber subscriber = new BaseSubscriber<>() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + request(Long.MAX_VALUE); + } + @Override + protected void hookOnNext(MultipartParser.Token token) { + if (token instanceof MultipartParser.BodyToken bodyToken) { + receivedBuffers.add(bodyToken.buffer()); + cancel(); + } + } + }; + tokens.subscribe(subscriber); + + // The cancellation above arrives while the parser emits its queued body buffers. + // Ownership of an emitted buffer belongs to the sink, so the parser must not + // release it on disposal: with Netty, body buffers are slices of the inbound + // buffer, and releasing one twice releases the inbound buffer prematurely. + assertThat(receivedBuffers).isNotEmpty(); + assertThat(receivedBuffers).allSatisfy(received -> + assertThat(((PooledDataBuffer) received).isAllocated()).isTrue()); + receivedBuffers.forEach(DataBufferUtils::release); + } + }