feedback

Author: mhl-b

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

@@ -146,12 +146,12 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
                 currentRequestStream.handleNettyContent((HttpContent) msg);
             }
         } finally {
-            if (shouldRead) {
-                ctx.channel().eventLoop().execute(ctx::read);
-            }
             if (msg instanceof LastHttpContent) {
                 currentRequestStream = null;
             }
+            if (shouldRead) {
+                ctx.channel().eventLoop().execute(ctx::read);
+            }
             activityTracker.stopActivity();
         }
     }

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

@@ -55,7 +55,7 @@ public Netty4HttpRequest(int sequence, io.netty.handler.codec.http.HttpRequest n
         this(sequence, nettyRequest, content, new AtomicBoolean(false), null);
     }
 
-    public Netty4HttpRequest(
+    private Netty4HttpRequest(
         int sequence,
         io.netty.handler.codec.http.HttpRequest nettyRequest,
         HttpBody content,
@@ -104,7 +104,9 @@ public HttpBody body() {
 
     @Override
     public void setBody(HttpBody body) {
-        this.content = body.asFull();
+        assert this.content.isStream() : "only stream content can be replaced";
+        assert body.isFull() : "only full content can replace stream";
+        this.content = body;
         this.hasContent = body.isEmpty() == false;
     }
 

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

@@ -396,8 +396,9 @@ protected boolean isContentAlwaysEmpty(HttpResponse msg) {
                         }
                         return super.isContentAlwaysEmpty(msg);
                     }
-                });
-            ch.pipeline().addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));
+                })
+                .addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));
+
             if (handlingSettings.compression()) {
                 ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(handlingSettings.compressionLevel()) {
                     @Override

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

@@ -25,7 +25,7 @@ private static void replaceBody(RestRequest restRequest, ReleasableBytesReferenc
 
     /**
      * Aggregates content of the RestRequest and notifies consumer with updated, in-place, RestRequest.
-     * If content is already aggregated then does nothing.
+     * If content is already aggregated then passes through same request.
      */
     public static void aggregate(RestRequest restRequest, Consumer<RestRequest> resultConsumer) {
         final var httpRequest = restRequest.getHttpRequest();

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

@@ -444,7 +444,7 @@ private void dispatchRequest(
                 return;
             }
         }
-        final int contentLength = request.contentLength();
+        final int contentLength = request.isFullContent() ? request.contentLength() : 0;
         try {
             if (handler.canTripCircuitBreaker()) {
                 inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");

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

@@ -295,10 +295,7 @@ public boolean hasContent() {
     }
 
     public int contentLength() {
-        return switch (httpRequest.body()) {
-            case HttpBody.Full content -> content.bytes().length();
-            case HttpBody.Stream stream -> 0;
-        };
+        return httpRequest.body().asFull().bytes().length();
     }
 
     public boolean isFullContent() {

server/src/test/java/org/elasticsearch/rest/RestContentAggregatorTests.java

@@ -41,7 +41,11 @@ RestRequest newRestRequest(int size) {
 
     public void testFullBodyPassThrough() {
         var fullRequest = newRestRequest(between(1, 1024));
-        aggregate(fullRequest, (aggregated) -> assertEquals(fullRequest.content(), aggregated.content()));
+        var aggRef = new AtomicReference<RestRequest>();
+        aggregate(fullRequest, aggRef::set);
+        var aggRequest = aggRef.get();
+        assertSame(fullRequest, aggRequest);
+        assertSame(fullRequest.content(), aggRequest.content());
     }
 
     public void testZeroLengthStream() {
@@ -61,7 +65,7 @@ public void testAggregateRandomSize() {
         var streamChunks = range(0, nChunks).mapToObj(i -> randomReleasableBytesReference(chunkSize)).toList();
         var request = newRestRequest(chunkSize * nChunks);
         request.getHttpRequest().setBody(stream);
-        AtomicReference<RestRequest> aggregatedRef = new AtomicReference<>();
+        var aggregatedRef = new AtomicReference<RestRequest>();
 
         aggregate(request, aggregatedRef::set);