Fix buffer leak in Jackson2 decoders

Prior to this commit, the Jackson2 decoders (JSON, Smile, CBOR) could
leak buffers in case the decoding operation times out or is cancelled
and some buffers are still in flight.

This commit ensures that buffers are released on cancel signals.

Fixes gh-33731

Author: bclozel

Diff for: spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -43,6 +43,7 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.PooledDataBuffer;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.http.codec.HttpMessageDecoder;
 import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -157,7 +158,8 @@ public Flux<Object> decode(Publisher<DataBuffer> input, ResolvableType elementTy
 				catch (IOException ex) {
 					sink.error(processException(ex));
 				}
-			});
+			})
+			.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 		});
 	}
 

Diff for: spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonDecoderTests.java

@@ -116,7 +116,7 @@ void canDecodeWithObjectMapperRegistrationForType() {
 	}
 
 	@Test  // SPR-15866
-	public void canDecodeWithProvidedMimeType() {
+	void canDecodeWithProvidedMimeType() {
 		MimeType textJavascript = new MimeType("text", "javascript", StandardCharsets.UTF_8);
 		Jackson2JsonDecoder decoder = new Jackson2JsonDecoder(new ObjectMapper(), textJavascript);
 
@@ -222,15 +222,15 @@ void invalidData() {
 	}
 
 	@Test // gh-22042
-	public void decodeWithNullLiteral() {
+	void decodeWithNullLiteral() {
 		Flux<Object> result = this.decoder.decode(Flux.concat(stringBuffer("null")),
 				ResolvableType.forType(Pojo.class), MediaType.APPLICATION_JSON, Collections.emptyMap());
 
 		StepVerifier.create(result).expectComplete().verify();
 	}
 
 	@Test // gh-27511
-	public void noDefaultConstructor() {
+	void noDefaultConstructor() {
 		Flux<DataBuffer> input = Flux.from(stringBuffer("{\"property1\":\"foo\",\"property2\":\"bar\"}"));
 
 		testDecode(input, BeanWithNoDefaultConstructor.class, step -> step
@@ -251,7 +251,7 @@ void codecException() {
 	}
 
 	@Test  // SPR-15975
-	public void customDeserializer() {
+	void customDeserializer() {
 		Mono<DataBuffer> input = stringBuffer("{\"test\": 1}");
 
 		testDecode(input, TestObject.class, step -> step
@@ -272,7 +272,7 @@ void bigDecimalFlux() {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void decodeNonUtf8Encoding() {
+	void decodeNonUtf8Encoding() {
 		Mono<DataBuffer> input = stringBuffer("{\"foo\":\"bar\"}", StandardCharsets.UTF_16);
 		ResolvableType type = ResolvableType.forType(new ParameterizedTypeReference<Map<String, String>>() {});
 
@@ -285,7 +285,7 @@ public void decodeNonUtf8Encoding() {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void decodeNonUnicode() {
+	void decodeNonUnicode() {
 		Flux<DataBuffer> input = Flux.concat(stringBuffer("{\"føø\":\"bår\"}", StandardCharsets.ISO_8859_1));
 		ResolvableType type = ResolvableType.forType(new ParameterizedTypeReference<Map<String, String>>() {});
 
@@ -298,7 +298,7 @@ public void decodeNonUnicode() {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void decodeMonoNonUtf8Encoding() {
+	void decodeMonoNonUtf8Encoding() {
 		Mono<DataBuffer> input = stringBuffer("{\"foo\":\"bar\"}", StandardCharsets.UTF_16);
 		ResolvableType type = ResolvableType.forType(new ParameterizedTypeReference<Map<String, String>>() {});
 
@@ -311,7 +311,7 @@ public void decodeMonoNonUtf8Encoding() {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void decodeAscii() {
+	void decodeAscii() {
 		Flux<DataBuffer> input = Flux.concat(stringBuffer("{\"foo\":\"bar\"}", StandardCharsets.US_ASCII));
 		ResolvableType type = ResolvableType.forType(new ParameterizedTypeReference<Map<String, String>>() {});
 
@@ -322,6 +322,15 @@ public void decodeAscii() {
 				null);
 	}
 
+	@Test
+	void cancelWhileDecoding() {
+		Flux<DataBuffer> input = Flux.just(
+				stringBuffer("[{\"bar\":\"b1\",\"foo\":\"f1\"},").block(),
+				stringBuffer("{\"bar\":\"b2\",\"foo\":\"f2\"}]").block());
+
+		testDecodeCancel(input, ResolvableType.forClass(Pojo.class), null, null);
+	}
+
 
 	private Mono<DataBuffer> stringBuffer(String value) {
 		return stringBuffer(value, StandardCharsets.UTF_8);