From 03221ef63ba20254702e2f6825ac2fb13ce41547 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 7 Nov 2026 10:19:09 +0100 Subject: [PATCH] Release a chunk when the request body read fails MultipartUploadTest failed on windows-latest JDK 11 (run 42294836025 on PR #2378), and a leaked ByteBuf then failed every later test class. Two causes. The test servlet answered once commons-fileupload reached the closing boundary, which for a chunked body can come before the terminating chunk. Jetty 11 then aborts the connection for unconsumed input (HttpConnection.onCompleted), and on Windows the reset discards the response before the client reads it. The client replayed the request on a new connection, and the replay read the InputStreamPart the first attempt had already closed: "Stream closed". The servlet now reads the request to the end before it answers. Locally, on Windows, the chunked upload failed once in 492 repetitions without that and passed 1,400 repetitions with it. BodyChunkedInput.readChunk allocated a chunk and did not release it when Body.transferTo threw, so that IOException leaked the buffer. ChunkedWriteHandler releases only chunks it is handed. The HTTP/2 body pump already releases its buffer on failure. Local mvnw.cmd clean verify ran on JDK 17; CI covers JDK 11. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../netty/request/body/BodyChunkedInput.java | 9 ++- .../request/body/BodyChunkedInputTest.java | 71 +++++++++++++++++++ .../body/multipart/MultipartUploadTest.java | 4 ++ 3 files changed, 83 insertions(+), 1 deletion(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/body/BodyChunkedInputTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java b/client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java index 772baca43a..334c62ad3a 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java @@ -59,7 +59,14 @@ public ByteBuf readChunk(ByteBufAllocator alloc) throws Exception { } ByteBuf buffer = alloc.buffer(chunkSize); - Body.BodyState state = body.transferTo(buffer); + Body.BodyState state; + try { + state = body.transferTo(buffer); + } catch (Throwable t) { + // ChunkedWriteHandler only releases chunks it is handed, so a failed read would leak this one. + buffer.release(); + throw t; + } progress += buffer.writerIndex(); switch (state) { case STOP: diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/BodyChunkedInputTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/BodyChunkedInputTest.java new file mode 100644 index 0000000000..407e7d9ac3 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/BodyChunkedInputTest.java @@ -0,0 +1,71 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * 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 + * + * http://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.asynchttpclient.netty.request.body; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import org.asynchttpclient.request.body.Body; +import org.junit.jupiter.api.Test; + +import java.io.IOException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class BodyChunkedInputTest { + + @Test + public void readChunkReleasesTheChunkWhenTheBodyFails() { + IOException failure = new IOException("Stream closed"); + FailingBody body = new FailingBody(failure); + BodyChunkedInput input = new BodyChunkedInput(body); + + IOException thrown = assertThrows(IOException.class, () -> input.readChunk(ByteBufAllocator.DEFAULT)); + + assertSame(failure, thrown); + assertNotNull(body.target); + assertEquals(0, body.target.refCnt()); + } + + private static final class FailingBody implements Body { + + private final IOException failure; + private ByteBuf target; + + FailingBody(IOException failure) { + this.failure = failure; + } + + @Override + public long getContentLength() { + return -1; + } + + @Override + public BodyState transferTo(ByteBuf target) throws IOException { + this.target = target; + // Part of a chunk is written before the source fails, as a multipart body does. + target.writeByte('x'); + throw failure; + } + + @Override + public void close() { + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java index 6ffe979599..8717d2f849 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/multipart/MultipartUploadTest.java @@ -416,6 +416,10 @@ public void service(HttpServletRequest request, HttpServletResponse response) th } catch (FileUploadException e) { // } + // The parser stops at the closing boundary, before a chunked body's terminating chunk if that has + // not arrived yet. Jetty then aborts the connection for unconsumed input once the response is + // done, and on Windows the reset discards the response before the client has read it. + request.getInputStream().transferTo(OutputStream.nullOutputStream()); try (Writer w = response.getWriter()) { w.write(Integer.toString(getFilesProcessed())); resetFilesProcessed();