Request chunks emitted after response is completed are not released

[romain-grecourt]

Request chunks emitted in the content publisher after the response is completed are not released, this triggers LEAK warning messages.

Any endpoint that returns a response instantly would reproduce this behavior.
Note that this is a regression introduced by #1830, and was originally fixed in #912

Reproduce

Any endpoint that returns a response without consuming its payload.
The example below invokes System.gc() to expose the LEAK warning message.

WebServer.builder(
        Routing.builder()
               .get((req, res) -> {
                   System.gc();
                   res.send("OK");
               })
               .build())
         .port(8080)
         .build()
         .start();

The chunks are emitted after routing, so this requires two invocation to trigger the LEAK warning message.

Invoke the endpoint twice:

curl -sSf http://localhost:8080/health -X GET -H "Content-Type: application/json" -d {}
curl -sSf http://localhost:8080/health -X GET -H "Content-Type: application/json" -d {}

Any HTTP method can be used, GET is only used as an example, but PUT or POST would work as well.

In the logs you should see:

WARNING: LEAK: RequestChunk.release() was not called before it was garbage collected. While the Reactive WebServer is designed to automatically release all the RequestChunks, it still comes with a considerable performance penalty and a demand for a large memory space (depending on expected throughput, it might require even more than 2GB). As such the users are strongly advised to release all the RequestChunk instances explicitly when they're not needed.

Fix

The following changes fix the issue:

diff --git a/webserver/webserver/src/main/java/io/helidon/webserver/ForwardingHandler.java b/webserver/webserver/src/main/java/io/helidon/webserver/ForwardingHandler.java
index e72bf391a..6e967f745 100644
--- a/webserver/webserver/src/main/java/io/helidon/webserver/ForwardingHandler.java
+++ b/webserver/webserver/src/main/java/io/helidon/webserver/ForwardingHandler.java
@@ -21,6 +21,7 @@ import java.util.Iterator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
+import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Logger;
@@ -196,7 +197,25 @@ public class ForwardingHandler extends SimpleChannelInboundHandler<Object> {
                             if (queue.release()) {
                                 queues.remove(queue);
                             }
-                            publisherRef.clearBuffer(DataChunk::release);
+                            publisherRef.subscribe(new Flow.Subscriber<DataChunk>() {
+                                @Override
+                                public void onSubscribe(Flow.Subscription subscription) {
+                                    subscription.request(Long.MAX_VALUE);
+                                }
+
+                                @Override
+                                public void onNext(DataChunk item) {
+                                    item.release();
+                                }
+
+                                @Override
+                                public void onError(Throwable throwable) {
+                                }
+
+                                @Override
+                                public void onComplete() {
+                                }
+                            });
                             // Enable auto-read only after response has been completed
                             // to avoid a race condition with the next response

[romain-grecourt]

@danielkec I've assigned this to you as there may be some subtle changes that are required.
Let me know if you need more details / or assistance.