Move HTTP content aggregation from Netty into RestController #129302

mhl-b · 2025-06-12T03:19:07Z

Netty HTTP content aggregator appears to be not flexible for our needs. In some cases, like bulk API we need to bypass aggregation, and Netty's aggregator does not have any memory safety, like backpressure or circuit breaker. That means we always aggregate requests fully and occasionally run OOM.

This change moves aggregation from the Netty module into Server module, RestController. That means in netty code we no longer deal with fully aggregated requests and everything is a stream of chunks. RestController aggregates chunks when RestHandler indicates that stream is not supported via new flag. Opens up opportunity to apply backpressure/circuit-breaker per chunk basis for all REST handlers. Also simplifies implementation of new streaming handlers.

Once RestHandler is resolved in RestController we check if it supports stream content. If stream is supported then request proceeds to the Handler. If stream is not supported a new RestContentAggregator aggregates chunks, then replaces content of the RestRequest with aggregated content.

PR touches many files but essentially changes are:

Removal of Netty4HttpAggregator and Bulk API predicates from Netty pipeline.
Addition of RestContentAggregator and support for HTTP content replacement.
Adjusting integ tests in netty and security modules to consume stream (AggregatingDispatcher) and not leak buffers.

closes #120746

mhl-b · 2025-06-12T04:16:36Z

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

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


Changing mutability for the HTTP request body, I need to swap stream with full content after aggregation. References to original HTTP request spread roots too deep.

nicktindall

Nice tidy up, I will leave to the other reviewers to approve because they have more context, but LGTM

nicktindall · 2025-06-12T04:25:39Z

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

+                try {
+                    dispatchRequest(aggregatedRequest, aggChannel, handler, methodHandlers, threadContext);
+                } catch (Exception e) {
+                    assert false : "exception must be handled in dispatchRequest";


Should we log an error here just in case it somehow happens in prod?

nicktindall · 2025-06-12T04:33:30Z

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

+                    replaceBody(restRequest, chunk);
+                    result.accept(restRequest);
+                } else {
+                    var comp = CompositeBytesReference.of(chunks.toArray(new ReleasableBytesReference[0]));


Is it worth making new ReleasableBytesReference[0] a constant to avoid the allocation? Or will the compiler be clever enough to do that I wonder.

I think it is smart enough to skip extra allocation. It still allocates array of the list size. Would be nice to have CompositeBytesReference.of(List<Chunks>).

nicktindall · 2025-06-12T04:39:19Z

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

+
+    public RestRequestAggregator(CircuitBreakerService circuitBreakerService) {
+        this.circuitBreakerService = circuitBreakerService;
+    }


Nit: do we use the circuitBreakerService in here? it doesn't look like it

My intention was to include CB in this PR. So I've done basic plumbing. But scope of CB appears fairly large. I will leave it as it is. CB will work as before, we fully aggregate content and then try breaker.

mhl-b · 2025-06-15T04:43:01Z

...sport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpServerTransportTests.java

-        final HttpValidator httpValidator = (httpRequest, channel, validationListener) -> {
-            // assert that the validator sees the request unaltered
-            assertThat(httpRequest.uri(), is(uri));
-            if (randomBoolean()) {
-                validationListener.onResponse(null);
-            } else {
-                validationListener.onFailure(new ElasticsearchException("Boom"));
-            }
-        };
        try (


Before, oversized requests that fail auth will be rejected at netty aggregator with size exception. Right now it will go through to the RestController, controller will send a response based on auth exception and close(discard) stream.

I think it's ok to have Auth exception with higher precedence than content size. But I think there is a possibility for an evil request with extremely large content that will be discarded, but we want to close channel sooner, rather than discarding content. If we want to keep previous behaviour I need to adjust Netty4HttpContentSizeHandler.

mhl-b · 2025-06-15T04:52:00Z

@DaveCTurner CI has been happy last 3 runs, with minor test fixes. It's ready for review. I will address audit logging in following PR if this one passes.

elasticsearchmachine · 2025-06-15T04:53:02Z

Pinging @elastic/es-distributed-coordination (Team:Distributed Coordination)

elasticsearchmachine · 2025-06-15T04:53:03Z

Hi @mhl-b, I've created a changelog YAML for you.

DaveCTurner

I like it. I haven't quite finished reviewing the changes in detail but have left some comments from my first pass through.

DaveCTurner · 2025-06-16T12:37:33Z

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

+        return new ReleasableBytesReference(randomBytesReference(length), new AbstractRefCounted() {
+            @Override
+            protected void closeInternal() {
+
+            }
+        });


Can we leak-track these things?

Suggested change

return new ReleasableBytesReference(randomBytesReference(length), new AbstractRefCounted() {

@Override

protected void closeInternal() {

}

});

return new ReleasableBytesReference(randomBytesReference(length), LeakTracker.wrap(AbstractRefCounted.of(() -> {})));

:) return new ReleasableBytesReference(randomBytesReference(length), LeakTracker.wrap(new AbstractRefCounted() { @Override protected void closeInternal() {} }));

DaveCTurner · 2025-06-16T12:39:36Z

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


-    @Override
-    public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
+    public void dispatchAggregatedRequest(RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {}


Can we assert this?

Suggested change

public void dispatchAggregatedRequest(RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {}

public void dispatchAggregatedRequest(RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {

assert restRequest.isStreamedContent() == false;

}

DaveCTurner · 2025-06-16T12:44:21Z

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

-        return httpRequest.body().asFull().bytes().length();
+        return switch (httpRequest.body()) {
+            case HttpBody.Full content -> content.bytes().length();
+            case HttpBody.Stream stream -> Math.toIntExact(httpRequest.contentLengthHeader());


I'm suspicious about this, are you sure we need it? I'd rather we only exposed the length of an aggregated body. My worry is that code which is processing a streaming body shouldn't be dependent on the content length since it's not always available, but with this lenience we might end up not covering the unavailable-length case properly in tests.

Agree. This is only used for the circuit breaker estimation. I can remove whole content-length changes, I don't think it's essential for this PR.

Removed content-length changes c7bb930

Added hasContent to the HTTP request, we need to know if content is present before aggregation happens, so I'm keeping logic for headers check. 34e8881

DaveCTurner · 2025-06-16T12:45:55Z

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

+                try {
+                    dispatchRequest(aggregatedRequest, restChannel, handler, methodHandlers, threadContext);
+                } catch (Exception e) {
+                    // dispatchRequest already handles exceptions, this time we wont be able to send response


I don't understand this comment. If dispatchRequest is expected not to throw anything, should we assert false : e here?

Not well written by me. It's a bit tricky. dispatchRequest catches errors and tries to send error response. But sending error response might fail too, at least we have a test that does that by injecting failure into sendResponse path, not sure that production code can throw there. Failing to send response can only go to the error log.
Cannot re-throw Exception without wrapping into RuntimeException but that mess up error handling. "Cannot" means the whole chain need to have throw Exception signature which is pretty large.

Looking at this again, wrapping into RuntimeException might be a right thing to do. Let it bubble up to the pipelining http error handler. This will properly close channel.

Rethrow exception 43bf06a

DaveCTurner · 2025-06-16T12:46:54Z

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

+        boolean closing;
+        ArrayList<ReleasableBytesReference> chunks;
+
+        private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> result) {


naming nit:

Suggested change

private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> result) {

private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> resultConsumer) {

DaveCTurner · 2025-06-16T12:48:51Z

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

+            if (closing == false) {
+                closing = true;
+                if (chunks != null) {
+                    Releasables.close(chunks);


If closed mid-aggregation, do we need some way to notify the handler? Conversely, if we got to the end of the aggregation, could we release the chunks earlier? AIUI we close the request when starting to send the response, which seems much later than necessary.

If closed mid-aggregation, do we need some way to notify the handler?

I don't think so. Handlers that handle whole content shouldn't bother about incomplete requests. Thats the whole point of aggregation, keep it simple downstream, right?

Conversely, if we got to the end of the aggregation, could we release the chunks earlier?

We release chunks at prepareRequest, if you remember we changed that few months ago 7aa07f1. This close handler is for premature termination.

DaveCTurner · 2025-06-16T12:50:29Z

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

@@ -30,6 +30,10 @@ enum HttpVersion {

    HttpBody body();

+    default void setBody(HttpBody body) {
+        throw new UnsupportedOperationException("not implemented");


assert false here too? This'd be a bug right? Although there's only 4 implementing classes (plus a few subclasses) so I'd prefer not to have a default implementation.

DaveCTurner · 2025-06-16T12:54:00Z

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

+            if (encoding != null && encoding.equals("chunked")) {
+                return -1;
+            } else {
+                return 0;


Is this true? What requests might not have a Content-Length or Transfer-Encoding header exactly? I thought that'd not be allowed in HTTP/1.1, and in HTTP/1.0 a request without a Content-Length header would have a body that is terminated by the connection close.

What requests might not have a Content-Length or Transfer-Encoding header exactly?

For example GET request without body is allowed to not have either.

https://www.rfc-editor.org/rfc/rfc9110.html#section-8.6-5

A user agent SHOULD NOT send a Content-Length header field when the request message does not contain content and the method semantics do not anticipate such data.

This if-condition might not be very precise, but it's practical. When Transfer-Encoding is present it is very likely chunked. And in this case Content-Length should be omitted. But if none of them present it should be zero length content.

Replaced with hasContent 34e8881

DaveCTurner · 2025-06-16T12:58:59Z

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

@@ -119,10 +91,18 @@ public HttpBody body() {
        return content;
    }

+    @Override
+    public void setBody(HttpBody body) {
+        if (body.isFull()) {


Can we assert that we only ever replace a streaming body with a fully-aggregated body?

mhl-b · 2025-06-16T20:49:58Z

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

@@ -119,10 +90,14 @@ public HttpBody body() {
        return content;
    }

+    @Override
+    public void setBody(HttpBody body) {
+        this.content = body.asFull();


body.asFull(); includes assertion

Sure, but can we also check that this.content is not already a fully-aggregated body?

…oval

elasticsearchmachine · 2025-06-17T18:55:32Z

Hi @mhl-b, I've updated the changelog YAML for you.

…oval

…tp-aggregator-removal

DaveCTurner

One further comment on supporting contentLength() on a streaming request, and a few nits, otherwise this looks good to go.

DaveCTurner · 2025-06-18T14:49:09Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandler.java

+            if (msg instanceof LastHttpContent) {
+                currentRequestStream = null;


readability nit: this has to happen before the ctx::read above; of course it does today because we're already on the event loop so the dispatched ctx::read happens after this method returns but still it's confusing to write them in this order.

DaveCTurner · 2025-06-18T14:53:11Z

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

-    }
-
-    private Netty4HttpRequest(
+    public Netty4HttpRequest(


nit: probably a mistake to expose the released flag to the public

Suggested change

public Netty4HttpRequest(

private Netty4HttpRequest(

DaveCTurner · 2025-06-18T14:54:42Z

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

+                });
+            ch.pipeline().addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));


nit: may as well carry on chaining the addLast calls

Suggested change

});

ch.pipeline().addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));

})

.addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));

DaveCTurner · 2025-06-18T14:58:16Z

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

-        return httpRequest.body().asFull().bytes().length();
+        return switch (httpRequest.body()) {
+            case HttpBody.Full content -> content.bytes().length();
+            case HttpBody.Stream stream -> 0;


I still think this is trappy, I'd much prefer it remained an error to call contentLength on a streaming request.

DaveCTurner · 2025-06-18T15:00:14Z

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

+
+    public void testFullBodyPassThrough() {
+        var fullRequest = newRestRequest(between(1, 1024));
+        aggregate(fullRequest, (aggregated) -> assertEquals(fullRequest.content(), aggregated.content()));


We need to assert that the consumer is invoked, otherwise this will pass if aggregate doesn't. Also can we assertSame?

…oval

DaveCTurner

LGTM

…#129302)

elasticsearchmachine added the v9.1.0 label Jun 12, 2025

mhl-b requested review from DaveCTurner, JeremyDahlgren and nicktindall June 12, 2025 03:19

mhl-b commented Jun 12, 2025

View reviewed changes

nicktindall reviewed Jun 12, 2025

View reviewed changes

mhl-b force-pushed the http-aggregator-removal branch from cc408fb to 9ced839 Compare June 13, 2025 04:42

move aggregation from netty to rest controller

0cd4f72

mhl-b force-pushed the http-aggregator-removal branch from 7012336 to 0cd4f72 Compare June 14, 2025 20:11

test fix

d994791

mhl-b commented Jun 15, 2025

View reviewed changes

mhl-b marked this pull request as ready for review June 15, 2025 04:50

elasticsearchmachine added the needs:triage Requires assignment of a team area label label Jun 15, 2025

mhl-b added :Distributed Coordination/Network Http and internode communication implementations Team:Distributed Coordination Meta label for Distributed Coordination team >enhancement and removed needs:triage Requires assignment of a team area label labels Jun 15, 2025

Update docs/changelog/129302.yaml

7460a07

DaveCTurner reviewed Jun 16, 2025

View reviewed changes

mhl-b added 2 commits June 16, 2025 13:18

add assertions

3abd181

remove content-length-header

c7bb930

mhl-b commented Jun 16, 2025

View reviewed changes

mhl-b added 5 commits June 16, 2025 14:04

Merge branch 'main' into http-aggregator-removal

2077ad4

rethrow dispatchRequest exception

43bf06a

Merge remote-tracking branch 'upstream/main' into http-aggregator-rem…

5797b76

…oval

leak wrapper fix

db645ed

leak wrapper fix

c88dbbe

mhl-b added 2 commits June 16, 2025 16:18

Merge branch 'main' into http-aggregator-removal

59c0e23

Merge branch 'main' into http-aggregator-removal

7fcf9bb

mhl-b requested a review from DaveCTurner June 17, 2025 04:26

mhl-b added 2 commits June 17, 2025 09:33

Merge remote-tracking branch 'upstream/main' into http-aggregator-rem…

737fb67

…oval

Update docs/changelog/129302.yaml

d2e1622

mhl-b added 3 commits June 17, 2025 17:39

Merge remote-tracking branch 'upstream/main' into http-aggregator-rem…

86b8e28

…oval

add hasContent for stream

34e8881

Merge remote-tracking branch 'origin/http-aggregator-removal' into ht…

cdc8560

…tp-aggregator-removal

DaveCTurner reviewed Jun 18, 2025

View reviewed changes

mhl-b added 2 commits June 18, 2025 21:46

Merge remote-tracking branch 'upstream/main' into http-aggregator-rem…

5dfbd9f

…oval

feedback

43232ee

DaveCTurner approved these changes Jun 19, 2025

View reviewed changes

mhl-b merged commit eeca493 into elastic:main Jun 19, 2025
27 checks passed

mhl-b mentioned this pull request Jun 19, 2025

Add REST content aggregator #117787

Closed

kderusso pushed a commit to kderusso/elasticsearch that referenced this pull request Jun 23, 2025

Move HTTP content aggregation from Netty into RestController (elastic…

606c08f

…#129302)

mridula-s109 pushed a commit to mridula-s109/elasticsearch that referenced this pull request Jun 25, 2025

Move HTTP content aggregation from Netty into RestController (elastic…

0e3eb36

…#129302)

mhl-b mentioned this pull request Jul 4, 2025

Add audit logging for stream content #130594

Merged

	private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> result) {
	private AggregationChunkHandler(RestRequest restRequest, Consumer<RestRequest> resultConsumer) {

		if (msg instanceof LastHttpContent) {
		currentRequestStream = null;

		});
		ch.pipeline().addLast(new Netty4HttpContentSizeHandler(decoder, handlingSettings.maxContentLength()));

Move HTTP content aggregation from Netty into RestController #129302

Move HTTP content aggregation from Netty into RestController #129302

Uh oh!

Conversation

mhl-b commented Jun 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mhl-b Jun 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nicktindall left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b commented Jun 15, 2025

Uh oh!

elasticsearchmachine commented Jun 15, 2025

Uh oh!

elasticsearchmachine commented Jun 15, 2025

Uh oh!

DaveCTurner left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mhl-b Jun 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

mhl-b commented Jun 12, 2025 •

edited

Loading

mhl-b Jun 12, 2025 •

edited

Loading

mhl-b Jun 15, 2025 •

edited

Loading

mhl-b Jun 18, 2025 •

edited

Loading

mhl-b Jun 16, 2025 •

edited

Loading

mhl-b Jun 16, 2025 •

edited

Loading

mhl-b Jun 16, 2025 •

edited

Loading

mhl-b Jun 16, 2025 •

edited

Loading