(cxf) branch main updated: CXF-9271: SharedInputBuffer.consumeContent() can park the I/O reactor thread forever on close()/shutdown() never signal suspendInput (#3697)

[reta.apache.org via commits]

This is an automated email from the ASF dual-hosted git repository.

reta pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git


The following commit(s) were added to refs/heads/main by this push:
     new e346751d5e1 CXF-9271: SharedInputBuffer.consumeContent() can park the 
I/O reactor thread forever on close()/shutdown() never signal suspendInput 
(#3697)
e346751d5e1 is described below

commit e346751d5e160b777840e051311133c4acbb821d
Author: Andriy Redko <drr...@gmail.com>
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)
---
 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 2754fa61513..487fdfa005f 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();
+        }
+    }
 }