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 <cj848@hanmail.net>

spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java

@@ -96,12 +96,12 @@ final class MultipartParser extends BaseSubscriber<DataBuffer> {
 	 */
 	public static Flux<Token> parse(Flux<DataBuffer> buffers, byte[] boundary, int maxHeadersSize,
 			Charset headersCharset) {
-		return Flux.create(sink -> {
+		return Flux.<Token>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

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<MultipartParser.Token> tokens = MultipartParser.parse(Flux.just(buffer), boundary, 8192, UTF_8);
+
+		List<MultipartParser.Token> received = new ArrayList<>();
+		BaseSubscriber<MultipartParser.Token> 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);
+	}
+
+}