remove content-length-header

Author: mhl-b

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

@@ -13,7 +13,6 @@
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
-import io.netty.handler.codec.http.HttpUtil;
 import io.netty.handler.codec.http.QueryStringDecoder;
 import io.netty.handler.codec.http.cookie.Cookie;
 import io.netty.handler.codec.http.cookie.ServerCookieDecoder;
@@ -93,9 +92,7 @@ public HttpBody body() {
 
     @Override
     public void setBody(HttpBody body) {
-        var fullBody = body.asFull();
-        var contentLength = fullBody.bytes().length();
-        HttpUtil.setContentLength(nettyRequest, contentLength);
+        this.content = body.asFull();
     }
 
     @Override

server/src/main/java/org/elasticsearch/http/HttpPreRequest.java

@@ -65,22 +65,4 @@ default String header(String name) {
         return null;
     }
 
-    /**
-     * Returns value of content length header. If content is chunked-encoded and length is not present
-     * then returns -1, unknown length. Requests that has no content are free to drop content length header
-     * and can be assumed as zero length.
-     */
-    default long contentLengthHeader() {
-        var lenstr = header("Content-Length");
-        if (lenstr == null) {
-            var encoding = header("Transfer-Encoding");
-            if (encoding != null && encoding.equals("chunked")) {
-                return -1;
-            } else {
-                return 0;
-            }
-        } else {
-            return Long.parseLong(lenstr);
-        }
-    }
 }

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

@@ -32,15 +32,9 @@ public static void aggregate(RestRequest restRequest, Consumer<RestRequest> resu
         switch (httpRequest.body()) {
             case HttpBody.Full full -> resultConsumer.accept(restRequest);
             case HttpBody.Stream stream -> {
-                if (httpRequest.contentLengthHeader() == 0) {
-                    stream.close();
-                    replaceBody(restRequest, ReleasableBytesReference.empty());
-                    resultConsumer.accept(restRequest);
-                } else {
-                    final var aggregationHandler = new AggregationChunkHandler(restRequest, resultConsumer);
-                    stream.setHandler(aggregationHandler);
-                    stream.next();
-                }
+                final var aggregationHandler = new AggregationChunkHandler(restRequest, resultConsumer);
+                stream.setHandler(aggregationHandler);
+                stream.next();
             }
         }
     }
@@ -73,14 +67,15 @@ public void onNext(ReleasableBytesReference chunk, boolean isLast) {
             } else {
                 if (chunks == null) {
                     replaceBody(restRequest, chunk);
-                    resultConsumer.accept(restRequest);
                 } else {
                     chunks.add(chunk);
                     var comp = CompositeBytesReference.of(chunks.toArray(new ReleasableBytesReference[0]));
                     var relComp = new ReleasableBytesReference(comp, Releasables.wrap(chunks));
                     replaceBody(restRequest, relComp);
-                    resultConsumer.accept(restRequest);
                 }
+                chunks = null;
+                closing = true;
+                resultConsumer.accept(restRequest);
             }
         }
 

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

@@ -445,9 +445,7 @@ private void dispatchRequest(
                 return;
             }
         }
-        // TODO: estimate streamed content size for circuit breaker,
-        // something like http_max_chunk_size * avg_compression_ratio(for compressed content)
-        final int contentLength = request.isFullContent() ? request.contentLength() : 0;
+        final int contentLength = request.contentLength();
         try {
             if (handler.canTripCircuitBreaker()) {
                 inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");

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

@@ -291,7 +291,7 @@ public final String path() {
     }
 
     public boolean hasContent() {
-        return contentLength() != 0;
+        return isStreamedContent() || contentLength() > 0;
     }
 
     /**
@@ -301,7 +301,7 @@ public boolean hasContent() {
     public int contentLength() {
         return switch (httpRequest.body()) {
             case HttpBody.Full content -> content.bytes().length();
-            case HttpBody.Stream stream -> Math.toIntExact(httpRequest.contentLengthHeader());
+            case HttpBody.Stream stream -> 0;
         };
     }
 

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

@@ -48,8 +48,10 @@ public void testZeroLengthStream() {
         var stream = new FakeHttpBodyStream();
         var request = newRestRequest(0);
         request.getHttpRequest().setBody(stream);
-        aggregate(request, (aggregated) -> assertEquals(ReleasableBytesReference.empty(), aggregated.content()));
-        assertTrue(stream.isClosed());
+        var aggregatedRef = new AtomicReference<RestRequest>();
+        aggregate(request, aggregatedRef::set);
+        stream.sendNext(ReleasableBytesReference.empty(), true);
+        assertEquals(0, aggregatedRef.get().contentLength());
     }
 
     public void testAggregateRandomSize() {