I wonder, rather than another layer of indirection here could we add a chunk-consuming method to RestChannelConsumer? Or possibly return something from prepareRequest which implements both RestChannelConsumer and a new RequestBodyChunkConsumer method to indicate we want to process a streaming body?

I added RequestBodyChunkConsumer

    public interface RequestBodyChunkConsumer extends RestChannelConsumer {
        void handleChunk(RestChannel channel, ReleasableBytesReference chunk, boolean isLast);
    }

and wiring for handler into body stream in handleRequest

if (request.isStreamedContent()) {
  assert action instanceof RequestBodyChunkConsumer;
  var chunkConsumer = (RequestBodyChunkConsumer) action;
  request.contentStream().setHandler((chunk, isLast) -> chunkConsumer.handleChunk(channel, chunk, isLast));
}

implementation looks like this

@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
    return new RequestBodyChunkConsumer() {

        int totalBytes = 0;

        @Override
        public void handleChunk(RestChannel channel, ReleasableBytesReference chunk, boolean isLast) {
            try (chunk) {
                totalBytes += chunk.length();
                if (isLast == false) {
                    request.contentStream().requestBytes(1024);
                } else {
                    channel.sendResponse(new RestResponse(RestStatus.OK, Integer.toString(totalBytes)));
                }
            }
        }

        @Override
        public void accept(RestChannel channel) throws Exception {
            request.contentStream().requestBytes(1024); // ask for first chunk
        }
    };
}

Are we just using a FullHttpRequest because this is a POC, and to minimise the change? HttpRequest would seem more appropriate here.

to minimise the change?

Yes, it's one of the non-essential small things that touches many lines of code and bloat PR. Still need to address this, no reason to pass FullHttpRequest here.

Similarly, can we assert currentRequestStream is null here or it can be non-null?

It can be non-null if previous request was stream too. When we see HttpRequest that means we received all parts of previous request - all HttpContent's and single LastHttpContent. At this point previous parts should be either in previous stream queue or processed by rest handler.

It can be non-null if previous request was stream too

We set currentRequestStream back to null on receiving LastHttpContent. Or do you mean that may not necessary happen for every streaming request?

Right, sorry I lost track of my own changes :) In normal circumstances should be null. If request is not properly terminated (no last content) and there is new request then currentRequestStream might be not null, but it will end up with decoding failure and connection shutdown. I will add test.

++ to a test, but could we also assert currentRequestStream == null here? It would be useful documentation if nothing else.

I think the activityTracker needs to include the else branch for stream process?

Still figuring out activity tracker change. When there is enough data in channel and we turned off auto-read then next channel.read() will be executed right a way in same thread and stack. That means we hit activity tracker multiple times due to recursive calls in netty. It trips our assertions that thread is already active.
This recursion should not be an issue, since read-buffer is 64kb, and chunks would be about 8kb, so up to 8 recursive calls might happen.

We need to make ActivityTracker reentrant for tracking write-path activity anyway, may as well do that here too:

public boolean maybeStartActivity() {
    assert trackedThread == Thread.currentThread() : trackedThread.getName() + " vs " + Thread.currentThread().getName();
    if (isIdle(get())) {
        startActivity();
        return true;
    } else {
        return false;
    }
}

After my changes in Netty4HttpRequestBodyStream that incorporated flow control and queue its no longer an issue. Moved start/end activity to it's original place. * scratching my head *

Hmm this now has a problem with audit logs which can be configured to log requests body. It is not really advisable to do that for bulk requests. But we do see that from time to time. The streaming here now makes it impossible to log the request body. Unless we find a workaround, it would be a breaking change.

It's possible to log with stream. You can wrap stream handlers one into another and do metering, logging, sniffing, even filtering. Might be not too friendly, pseudo-code:

currentHandler = httpRequest.body().contentStream().handler();
loggingBuffer = new SomeByteBuffer(); 
httpRequest.body().contentStream().setHandler((chunk, isLast) -> {
  copyChunk = chunk.copy();
  loggingBuffer.add(copyChunk);
  if (isLast) {
    log.info("content={}", loggingBuffer);
  }
  currentHandler.onNext(chunk, isLast);
});

Not sure how viable this would be with how auditing works today. The request payload is logged when the Rest handler is invoked. At this point, we simply don't have the full body.

We can add a bit of logic into HttpAggregator, when logging is enabled do full aggregation. But it's confusing side effect of enabling logging. Yet not breaking.

++ to a test, but could we also assert currentRequestStream == null here? It would be useful documentation if nothing else.

nit: please restore interrupt status flag:

Do we need to assertBusy() here? If so, could you comment explaining why? I would expect that once the client's event loop is idle this assertion should pass straight away. Maybe there's no easy way to wait for the event loop to become idle tho?

Can we assert that this future's isDone() returns false because of the backpressure?

I will add assertion, but I dont think it's a reliable indicator of backpressure. Even without backpressure, flushing 50MB might return false immediately, flush is not a blocking operation and there might be a few flushes to kernel buffer before netty's channel buffer will be drained.

Could we also have some chunked requests which don't specify the content-length up front?

I expect there'll only be one active handler at once, so I think we could use that to simplify things along the lines of org.elasticsearch.rest.ChunkedZipResponseIT.RandomZipResponsePlugin#responseRef.

Suppose to be used by pipelining tests where multiple requests in fly. I havent added test yet.