Fix Brotli decompression hang with large input buffers

httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java

@@ -87,12 +87,9 @@ public void updateCapacity(final CapacityChannel capacityChannel) throws IOExcep
     public void consume(final ByteBuffer src) throws IOException {
         while (src.hasRemaining()) {
             final ByteBuffer in = decoder.getInputBuffer();
+            in.clear();
+
             final int xfer = Math.min(src.remaining(), in.remaining());
-            if (xfer == 0) {
-                decoder.push(0);
-                pump();
-                continue;
-            }
             final int lim = src.limit();
             src.limit(src.position() + xfer);
             in.put(src);

httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java

@@ -26,19 +26,24 @@
  */
 package org.apache.hc.client5.http.async.methods;
 
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -246,4 +251,52 @@ void registerInExec() {
         final ContentCompressionAsyncExec exec = new ContentCompressionAsyncExec(map);
         assertNotNull(exec);
     }
+
+    @Test
+    void inflateBrotliLargerThanInputBuffer() {
+        assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+            final byte[] original = new byte[64 * 1024];
+            new Random(42).nextBytes(original);
+
+            final byte[] compressed = Encoder.compress(
+                    original,
+                    new Encoder.Parameters()
+                            .setQuality(6)
+                            .setWindow(22));
+
+            assertTrue(compressed.length > 8 * 1024);
+
+            final ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+            final AsyncDataConsumer downstream = new AsyncDataConsumer() {
+
+                @Override
+                public void updateCapacity(final CapacityChannel capacityChannel) {
+                }
+
+                @Override
+                public void consume(final ByteBuffer src) {
+                    final byte[] buf = new byte[src.remaining()];
+                    src.get(buf);
+                    output.write(buf, 0, buf.length);
+                }
+
+                @Override
+                public void streamEnd(final List<? extends Header> trailers) {
+                }
+
+                @Override
+                public void releaseResources() {
+                }
+            };
+
+            final InflatingBrotliDataConsumer inflating =
+                    new InflatingBrotliDataConsumer(downstream);
+
+            inflating.consume(ByteBuffer.wrap(compressed));
+            inflating.streamEnd(Collections.emptyList());
+
+            assertArrayEquals(original, output.toByteArray());
+        });
+    }
 }

httpclient5/src/test/java/org/apache/hc/client5/http/examples/AsyncClientServerBrotliRoundTrip.java

@@ -26,7 +26,6 @@
  */
 package org.apache.hc.client5.http.examples;
 
-import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -35,11 +34,9 @@
 import java.util.concurrent.Future;
 
 import com.aayushatharva.brotli4j.Brotli4jLoader;
+import com.aayushatharva.brotli4j.decoder.BrotliInputStream;
 import com.aayushatharva.brotli4j.encoder.BrotliOutputStream;
 
-import org.apache.commons.compress.compressors.CompressorException;
-import org.apache.commons.compress.compressors.CompressorInputStream;
-import org.apache.commons.compress.compressors.CompressorStreamFactory;
 import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
 import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
 import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
@@ -74,7 +71,7 @@
  * <p>
  * Notes:
  * - Encoding uses brotli4j (native JNI); make sure matching native dependency is on the runtime classpath.
- * - Decoding here uses Commons Compress via CompressorStreamFactory("br").
+ * - Decoding here uses brotli4j.
  */
 public final class AsyncClientServerBrotliRoundTrip {
 
@@ -124,8 +121,7 @@ public static void main(final String[] args) throws Exception {
             final boolean isBr = ce != null && BR.equalsIgnoreCase(ce.getValue());
             System.out.println("Response C-E     : " + (isBr ? BR : "(none)"));
 
-            final byte[] respPlain = isBr ? brotliDecompress(respBodyRaw) : respBodyRaw;
-            System.out.println("Response (plain) : " + new String(respPlain, StandardCharsets.UTF_8));
+            System.out.println("Response (plain) : " + new String(respBodyRaw, StandardCharsets.UTF_8));
         } finally {
             server.close(CloseMode.GRACEFUL);
         }
@@ -155,8 +151,7 @@ public void handle(
                 final Header ce = request.getFirstHeader(HttpHeaders.CONTENT_ENCODING);
                 if (ce != null && BR.equalsIgnoreCase(ce.getValue())) {
                     try (final InputStream in = entity.getContent();
-                         final CompressorInputStream bin =
-                                 new CompressorStreamFactory().createCompressorInputStream(BR, in)) {
+                         final BrotliInputStream bin = new BrotliInputStream(in)) {
                         requestPlain = readAll(bin);
                     }
                 } else {
@@ -174,7 +169,7 @@ public void handle(
                 response.addHeader(HttpHeaders.CONTENT_ENCODING, BR);
                 response.setEntity(new ByteArrayEntity(respCompressed, ContentType.APPLICATION_OCTET_STREAM));
 
-            } catch (final CompressorException ex) {
+            } catch (final IOException ex) {
                 response.setCode(HttpStatus.SC_BAD_REQUEST);
                 response.setEntity(new StringEntity("Invalid Brotli payload", StandardCharsets.UTF_8));
             } catch (final Exception ex) {
@@ -207,16 +202,4 @@ private static byte[] brotliCompress(final byte[] plain) throws IOException {
         }
         return baos.toByteArray();
     }
-
-    /**
-     * Decompress a Brotli-compressed byte[] using Commons Compress.
-     */
-    private static byte[] brotliDecompress(final byte[] compressed) throws IOException {
-        try (final InputStream in = new ByteArrayInputStream(compressed);
-             final CompressorInputStream bin = new CompressorStreamFactory().createCompressorInputStream(BR, in)) {
-            return readAll(bin);
-        } catch (final CompressorException e) {
-            throw new IOException("Failed to decompress Brotli data", e);
-        }
-    }
-}
+}