Add a test to show exception in channel does not work

Relates: ES-9537
ywangd · Oct 18, 2024 · f2dd6dc · f2dd6dc
1 parent 5bf446e
commit f2dd6dc
Showing 1 changed file with 33 additions and 0 deletions.
diff --git a/...nalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java b/...nalClusterTest/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;
@@ -211,6 +212,33 @@ 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);
+
+            if (randomBoolean() || true) {
+                handler.stream.channel().eventLoop().submit(() -> { throw new RuntimeException("simulated exception"); });
+            } else {
+                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()) {
@@ -594,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;
@@ -602,6 +631,10 @@ static class ServerRequestHandler implements BaseRestHandler.RequestBodyChunkCon
 
         @Override
         public void handleChunk(RestChannel channel, ReleasableBytesReference chunk, boolean isLast) {
+            Transports.assertTransportThread();
+            if (shouldThrowInsideHandleChunk) {
+                throw new RuntimeException("simulated exception inside handleChunk");
+            }
             recvChunks.add(new Chunk(chunk, isLast));
         }