This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 4.1.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 2fe34ccb7c0adc2dc580d64e06a01e42ef84ee03 Author: Andriy Redko <[email protected]> AuthorDate: Tue Nov 10 07:30:52 2026 -0500 CXF-9271: SharedInputBuffer.consumeContent() can park the I/O reactor thread forever on close()/shutdown() never signal suspendInput (#3697) (cherry picked from commit e346751d5e160b777840e051311133c4acbb821d) --- rt/transports/http-hc5/pom.xml | 5 +++ .../http/asyncclient/hc5/SharedInputBuffer.java | 6 ++-- .../asyncclient/hc5/SharedInputBufferTest.java | 36 ++++++++++++++++++++++ 3 files changed, 45 insertions(+), 2 deletions(-) diff --git a/rt/transports/http-hc5/pom.xml b/rt/transports/http-hc5/pom.xml index e2911c1e7c3..8569e2f1df9 100644 --- a/rt/transports/http-hc5/pom.xml +++ b/rt/transports/http-hc5/pom.xml @@ -79,5 +79,10 @@ <artifactId>cxf-testutils</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java index dc882dce34a..89f0a091022 100644 --- a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java +++ b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBuffer.java @@ -72,7 +72,7 @@ public class SharedInputBuffer extends ExpandableBuffer { } public int consumeContent(final ByteBuffer buffer, boolean last) throws IOException { - if (this.shutdown) { + if (this.shutdown || this.endOfStream) { return -1; } @@ -100,7 +100,7 @@ public class SharedInputBuffer extends ExpandableBuffer { // the reader signals suspendInput, once it has drained the buffer this.condition.signalAll(); - if (!buffer().hasRemaining() && !this.endOfStream) { + if (!buffer().hasRemaining() && !this.endOfStream && !this.shutdown) { try { suspendInput.await(); } catch (InterruptedException ex) { @@ -186,6 +186,7 @@ public class SharedInputBuffer extends ExpandableBuffer { this.endOfStream = true; this.lock.lock(); try { + this.suspendInput.signalAll(); this.condition.signalAll(); } finally { this.lock.unlock(); @@ -199,6 +200,7 @@ public class SharedInputBuffer extends ExpandableBuffer { this.shutdown = true; this.lock.lock(); try { + this.suspendInput.signalAll(); this.condition.signalAll(); } finally { this.lock.unlock(); diff --git a/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java b/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java index 1e66acc0f90..50a72d995fb 100644 --- a/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java +++ b/rt/transports/http-hc5/src/test/java/org/apache/cxf/transport/http/asyncclient/hc5/SharedInputBufferTest.java @@ -21,6 +21,7 @@ package org.apache.cxf.transport.http.asyncclient.hc5; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -28,8 +29,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import org.awaitility.Awaitility; + import org.junit.Test; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertArrayEquals; public class SharedInputBufferTest { @@ -81,4 +86,35 @@ public class SharedInputBufferTest { executor.shutdownNow(); } } + + @Test + public void testCloseWhenDataBlockFillsBuffer() throws Exception { + final CountDownLatch consumerWaiting = new CountDownLatch(1); + final SharedInputBuffer inbuf = new SharedInputBuffer(1024); + final byte[] data = new byte[4096]; + + final Thread t = new Thread(() -> { + try { + // No reader, the consumer should be blocked waiting for input + inbuf.consumeContent(ByteBuffer.wrap(data), false); + consumerWaiting.countDown(); + } catch (IOException ex) { + throw new UncheckedIOException(ex); + } + }); + + try { + t.start(); + + // Await until the consumer thread enters the WAITING state (waiting for condition) + Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> + assertThat(t.getState(), equalTo(Thread.State.WAITING)) + ); + + inbuf.close(); + assertThat(consumerWaiting.await(10, TimeUnit.SECONDS), equalTo(true)); + } finally { + inbuf.shutdown(); + } + } }
