add assertions

Author: mhl-b

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

@@ -93,11 +93,9 @@ public HttpBody body() {
 
     @Override
     public void setBody(HttpBody body) {
-        if (body.isFull()) {
-            var contentLength = body.asFull().bytes().length();
-            HttpUtil.setContentLength(nettyRequest, contentLength);
-        }
-        this.content = body;
+        var fullBody = body.asFull();
+        var contentLength = fullBody.bytes().length();
+        HttpUtil.setContentLength(nettyRequest, contentLength);
     }
 
     @Override

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

@@ -30,9 +30,7 @@ enum HttpVersion {
 
     HttpBody body();
 
-    default void setBody(HttpBody body) {
-        throw new UnsupportedOperationException("not implemented");
-    }
+    void setBody(HttpBody body);
 
     List<String> strictCookies();
 

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

@@ -27,17 +27,17 @@ 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.
      */
-    public static void aggregate(RestRequest restRequest, Consumer<RestRequest> result) {
+    public static void aggregate(RestRequest restRequest, Consumer<RestRequest> resultConsumer) {
         final var httpRequest = restRequest.getHttpRequest();
         switch (httpRequest.body()) {
-            case HttpBody.Full full -> result.accept(restRequest);
+            case HttpBody.Full full -> resultConsumer.accept(restRequest);
             case HttpBody.Stream stream -> {
                 if (httpRequest.contentLengthHeader() == 0) {
                     stream.close();
                     replaceBody(restRequest, ReleasableBytesReference.empty());
-                    result.accept(restRequest);
+                    resultConsumer.accept(restRequest);
                 } else {
-                    final var aggregationHandler = new AggregationChunkHandler(restRequest, result);
+                    final var aggregationHandler = new AggregationChunkHandler(restRequest, resultConsumer);
                     stream.setHandler(aggregationHandler);
                     stream.next();
                 }
@@ -47,14 +47,14 @@ public static void aggregate(RestRequest restRequest, Consumer<RestRequest> resu
 
     private static class AggregationChunkHandler implements HttpBody.ChunkHandler {
         final RestRequest restRequest;
-        final Consumer<RestRequest> result;
+        final Consumer<RestRequest> resultConsumer;
         final HttpBody.Stream stream;
         boolean closing;
         ArrayList<ReleasableBytesReference> chunks;
 
-        private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> result) {
+        private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> resultConsumer) {
             this.restRequest = restRequest;
-            this.result = result;
+            this.resultConsumer = resultConsumer;
             this.stream = restRequest.getHttpRequest().body().asStream();
         }
 
@@ -73,13 +73,13 @@ public void onNext(ReleasableBytesReference chunk, boolean isLast) {
             } else {
                 if (chunks == null) {
                     replaceBody(restRequest, chunk);
-                    result.accept(restRequest);
+                    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);
-                    result.accept(restRequest);
+                    resultConsumer.accept(restRequest);
                 }
             }
         }

server/src/test/java/org/elasticsearch/http/TestHttpRequest.java

@@ -52,6 +52,11 @@ public HttpBody body() {
         return HttpBody.empty();
     }
 
+    @Override
+    public void setBody(HttpBody body) {
+        throw new IllegalStateException("not allowed");
+    }
+
     @Override
     public Map<String, List<String>> getHeaders() {
         return headers;

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

@@ -879,6 +879,11 @@ public HttpBody body() {
                 return HttpBody.empty();
             }
 
+            @Override
+            public void setBody(HttpBody body) {
+                throw new IllegalStateException("not allowed");
+            }
+
             @Override
             public Map<String, List<String>> getHeaders() {
                 Map<String, List<String>> headers = new HashMap<>();

test/framework/src/main/java/org/elasticsearch/http/AggregatingDispatcher.java

@@ -16,7 +16,9 @@
 
 public class AggregatingDispatcher implements HttpServerTransport.Dispatcher {
 
-    public void dispatchAggregatedRequest(RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {}
+    public void dispatchAggregatedRequest(RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {
+        assert restRequest.isStreamedContent();
+    }
 
     @Override
     public final void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {

test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java

@@ -1004,7 +1004,7 @@ public ReleasableBytesReference randomReleasableBytesReference(int length) {
         return new ReleasableBytesReference(randomBytesReference(length), new AbstractRefCounted() {
             @Override
             protected void closeInternal() {
-
+                LeakTracker.wrap(AbstractRefCounted.of(() -> {}));
             }
         });
     }