Repository: micronaut-projects/micronaut-core (language: java)

Issue:
A NPE occurred when I use ProxyHttpClient to forward a request with body
### Expected Behavior

Forwards requests with body normally

### Actual Behaviour

This is an occasional bug that requires several attempts. You will see the above error.


`11:02:54.221 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 1320ms. Server Running: http://localhost:8080
Micronaut Version 4.7.5
11:03:00.916 [default-nioEventLoopGroup-1-15] WARN  i.n.u.c.AbstractEventExecutor - A task raised an exception. Task: io.micronaut.http.netty.body.StreamingNettyByteBody$SharedBuffer$$Lambda/0x0000000131635f58@6d48294e
java.lang.NullPointerException: Cannot invoke "io.micronaut.http.netty.EventLoopFlow.executeNow(java.lang.Runnable)" because "this.flow" is null
at io.micronaut.http.client.netty.StreamWriter.add(StreamWriter.java:90)
at io.micronaut.http.netty.body.StreamingNettyByteBody$SharedBuffer.subscribe0(StreamingNettyByteBody.java:348)
at io.micronaut.http.netty.body.StreamingNettyByteBody$SharedBuffer.lambda$subscribe$1(StreamingNettyByteBody.java:328)
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)`

### Steps To Reproduce

Clone https://github.com/mwz1994/hello-proxy.git (main branch)

Run the app

Call the API under the ProxyFilter path (demo/proxy/**) through POST or PUT protocol with body information.
The application code forwards the request to another remote service.

Such as 

`curl --location --request PUT 'https://mydomain.com/demo/proxy/freeapi/posts/1' \
--header 'Content-Type: application/json' \
--data '{
"name": "peter",
"want2say":"Hello there! I'\''m Peter"
}'`

This is an occasional bug that requires several attempts. You will see the above error.

I get an error after 30-50 calls with a tool like jmeter.

The error message for the HTTPS protocol is as follows; the exception message for the HTTP protocol is somewhat different

### Environment Information

micronaut 4.7.5
jdk 21
gradle 8.10.2

### Example Application

https://github.com/mwz1994/hello-proxy.git

### Version

4.7.5


Discussion / hints:

Fix patch:
diff --git a/http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java b/http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java
index 1ea457aa490..c0f6559f5f0 100644
--- a/http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java
+++ b/http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java
@@ -37,10 +37,11 @@
  */
 @Internal
 final class StreamWriter extends ChannelInboundHandlerAdapter implements BufferConsumer {
+    private final StreamingNettyByteBody body;
     private final Consumer<Throwable> errorHandler;
     private ChannelHandlerContext ctx;
     private EventLoopFlow flow;
-    private final Upstream upstream;
+    private Upstream upstream;
     private long unwritten = 0;
     private boolean completed = false;
 
@@ -49,8 +50,8 @@ final class StreamWriter extends ChannelInboundHandlerAdapter implements BufferC
      * @param errorHandler Handler to call when the streaming body emits an error
      */
     StreamWriter(StreamingNettyByteBody body, Consumer<Throwable> errorHandler) {
+        this.body = body;
         this.errorHandler = errorHandler;
-        this.upstream = body.primary(this);
     }
 
     /**
@@ -60,6 +61,7 @@ void startWriting() {
         if (ctx == null) {
             throw new IllegalStateException("Not added to a channel yet");
         }
+        upstream = body.primary(this);
         try {
             upstream.start();
         } catch (Exception e) {
@@ -71,8 +73,11 @@ void startWriting() {
      * Cancel writing the body (e.g. because a {@code CONTINUE} response was never received).
      */
     void cancel() {
-        upstream.allowDiscard();
-        upstream.disregardBackpressure();
+        if (upstream != null) {
+            upstream.allowDiscard();
+            upstream.disregardBackpressure();
+        }
+        body.close();
     }
 
     boolean isCompleted() {


Test patch: