Close channel on stream handler exception (#115505)

In case a stream handler throws uncaught exception, we should close the
channel and release associated resources to avoid the channel entering a
limbo state. This PR does that.

Resolves: ES-9537

Author: ywangd

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java

@@ -70,6 +70,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.transport.Transports;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.util.Collection;
@@ -215,6 +216,29 @@ public void testServerCloseConnectionMidStream() throws Exception {
         }
     }
 
+    public void testServerExceptionMidStream() throws Exception {
+        try (var ctx = setupClientCtx()) {
+            var opaqueId = opaqueId(0);
+
+            // write half of http request
+            ctx.clientChannel.write(httpRequest(opaqueId, 2 * 1024));
+            ctx.clientChannel.writeAndFlush(randomContent(1024, false));
+
+            // await stream handler is ready and request full content
+            var handler = ctx.awaitRestChannelAccepted(opaqueId);
+            assertBusy(() -> assertNotNull(handler.stream.buf()));
+            assertFalse(handler.streamClosed);
+
+            handler.shouldThrowInsideHandleChunk = true;
+            handler.stream.next();
+
+            assertBusy(() -> {
+                assertNull(handler.stream.buf());
+                assertTrue(handler.streamClosed);
+            });
+        }
+    }
+
     // ensure that client's socket buffers data when server is not consuming data
     public void testClientBackpressure() throws Exception {
         try (var ctx = setupClientCtx()) {
@@ -598,6 +622,7 @@ static class ServerRequestHandler implements BaseRestHandler.RequestBodyChunkCon
         RestChannel channel;
         boolean recvLast = false;
         volatile boolean streamClosed = false;
+        volatile boolean shouldThrowInsideHandleChunk = false;
 
         ServerRequestHandler(String opaqueId, Netty4HttpRequestBodyStream stream) {
             this.opaqueId = opaqueId;
@@ -606,6 +631,12 @@ static class ServerRequestHandler implements BaseRestHandler.RequestBodyChunkCon
 
         @Override
         public void handleChunk(RestChannel channel, ReleasableBytesReference chunk, boolean isLast) {
+            Transports.assertTransportThread();
+            if (shouldThrowInsideHandleChunk) {
+                // Must close the chunk. This is the contract of this method.
+                chunk.close();
+                throw new RuntimeException("simulated exception inside handleChunk");
+            }
             recvChunks.add(new Chunk(chunk, isLast));
         }
 

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

@@ -71,7 +71,11 @@ public void next() {
             if (buf == null) {
                 channel.read();
             } else {
-                send();
+                try {
+                    send();
+                } catch (Exception e) {
+                    channel.pipeline().fireExceptionCaught(e);
+                }
             }
         });
     }

server/src/main/java/org/elasticsearch/rest/BaseRestHandler.java

@@ -210,6 +210,15 @@ default void close() {}
     }
 
     public interface RequestBodyChunkConsumer extends RestChannelConsumer {
+
+        /**
+         * Handle one chunk of the request body. The handler <b>must</b> close the chunk once it is no longer
+         * needed to avoid leaking.
+         *
+         * @param channel The rest channel associated to the request
+         * @param chunk The chunk of request body that is ready for processing
+         * @param isLast Whether the chunk is the last one of the request
+         */
         void handleChunk(RestChannel channel, ReleasableBytesReference chunk, boolean isLast);
 
         /**