Walkthrough

Added channel tracking and release functionality for Reactor Netty 4 HTTP transport during node shutdown. Introduced integration tests to validate HTTP channel lifecycle management under concurrent requests, modified channel closure to handle async context coordination, and updated request consumers to register accepted channels with the transport.

Changes

Sequence Diagram(s)

sequenceDiagram
    participant Client
    participant Transport as ReactorNetty4HttpServerTransport
    participant Consumer as ReactorNetty4StreamingRequestConsumer
    participant Channel as HttpChannel
    participant CloseContext

    Client->>Transport: Incoming HTTP request
    Transport->>Consumer: Create consumer (pass transport)
    Consumer->>Transport: serverAcceptedChannel(channel)
    Transport->>Channel: Register accepted channel
    
    Consumer->>Consumer: Process request
    Consumer->>Channel: Close on completion
    
    alt closeContext not completed
        Channel->>CloseContext: addListener(close future)
        CloseContext->>Channel: Complete → triggers close
    else closeContext already completed
        Channel->>Channel: Close immediately
    end
    
    Note over Transport,Channel: On node shutdown
    Transport->>Transport: Track and release all<br/>accepted channels

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

• Files to focus on:
  • ReactorNetty4StreamingRequestConsumer.java and ReactorNetty4NonStreamingRequestConsumer.java — verify correct transport type narrowing and channel registration timing
  • ReactorNetty4StreamingHttpChannel.java and ReactorNetty4NonStreamingHttpChannel.java — carefully review async close coordination logic with closeContext completion state
  • ReactorNetty4HttpServerTransport.java — confirm serverAcceptedChannel override integrates properly with the request pipeline
  • Integration test classes — validate test assertions accurately capture channel lifecycle expectations

Suggested labels

transport, netty, test

Suggested reviewers

• andrross
• reta

Poem

A rabbit hops through channels bright,
Tracking requests, day and night,
Close contexts bind with async grace,
As channels find their resting place. 🐰✨

Pre-merge checks and finishing touches

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Thanks for the quick follow up here, @fdesu! Why does this channel not get closed? Manually closing the client seems to violate the intent of the test framework per this comment:

@andrross oh, good point, wasn’t aware of that!

I’m not entirely sure why the channel doesn’t get closed automatically but it just stays open since the deregistration doesn’t fire up which would have caused the RestCancellableNodeClient housekeeping to kick in. Maybe it’s somehow related to the channel type which is Reactor Netty4 kind of channel but I’m not really sure.

Another thought that I had in mind was that the test(s) don’t fully consume the Response body and hence the receiving side just keeps the channel alive but it doesn’t seem to be the case since we use EntityUtils.toString which tries-with-resources the underlying InputStream.

I’ll look deeper and would really appreciate if anyone can point me in the right direction.

❌ Gradle check result for 25166b1: null

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

Thanks for the quick follow up here, @fdesu! Why does this channel not get closed? Manually closing the client seems to violate the intent of the test framework per this comment:

OpenSearch/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java

Lines 2381 to 2382 in 8f838f1

	* The returned client gets automatically closed when needed, it shouldn't be closed as part of tests otherwise
	* it cannot be reused by other tests anymore.

@coderabbitai do you find @andrross comment above helpful to improve your review?

#!/bin/bash
# Fetch the referenced file to understand the framework's intent
cat test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java | sed -n '2370,2395p'

#!/bin/bash
# Look for how the rest client is created and managed in the test framework
rg -n -A5 -B5 "createRestClient|getRestClient" test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java | head -80

#!/bin/bash
# Check the OpenSearchTestClusterRule class to understand how it manages resources
fd -e java "OpenSearchTestClusterRule" --exec cat {}

#!/bin/bash
# Look at how RestCancellableNodeClient manages channel registrations
rg -n -A10 -B5 "class RestCancellableNodeClient" --type java

@peterzhuamazon Yes, @andrross's comment is very helpful and raises a valid architectural concern that should influence the review.

After examining the framework code, I can confirm that the test framework is explicitly designed to:

1. Reuse the RestClient across tests: The getRestClient() method creates a singleton client that's shared within the test scope
2. Handle cleanup centrally: The clearClusters() method in OpenSearchTestClusterRule is responsible for closing the client and verifying all channels are deregistered from RestCancellableNodeClient

The framework documentation at lines 2381-2382 explicitly states: "The returned client gets automatically closed when needed, it shouldn't be closed as part of tests otherwise it cannot be reused by other tests anymore."

The Core Issue:

The current fix uses try-with-resources to close the client in individual test methods, which:

• ❌ Violates the framework's design intent
• ❌ Breaks client reusability across tests in the same scope
• ❌ Treats the symptom (channels not closing) rather than the root cause

What Needs Investigation:

The real question is: Why doesn't the HttpChannel close automatically through the normal framework lifecycle? As @fdesu noted, it could be related to:

• Reactor Netty4 channel behavior
• Channel lifecycle management differences
• Incomplete response consumption (though EntityUtils.toString should handle this)

Recommendation:

Rather than working around the framework's design, I'd suggest:

1. Investigating why the channel remains open after clearClusters() is called
2. Determining if the framework's cleanup logic needs enhancement for Reactor Netty4 channels
3. Fixing the root cause so the framework's intended lifecycle works correctly

This would be a more robust solution that maintains the framework's architecture and benefits all tests using the framework.

I mean, I don't think it's the test harness's design that causes the issue, it's something else. My expectation is, after the restClient.performRequest(MultisearchRequest) is called and the response is fully consumed, the channel should get closed which doesn't happen. Am I missing anything?

Note that the:

1. org.opensearch.client.SearchIT#testMultiSearch works okay with the high level client
2. and the SearchRestCancellationIT#testAutomaticCancellationMultiSearchDuringQueryPhase works okay with getRestClient().performRequestAsync() which gets cancelled.

the channel should get closed which doesn't happen. Am I missing anything?

@fdesu I don't think you're missing anything. I think there might be a latent bug here with the Reactor Netty implementation. The channel gets added to a collection in AbstractHttpServerTransport#serverAcceptedChannel, which will guarantee that on node shutdown those channels get closed here. However, I don't see that AbstractHttpServerTransport#serverAcceptedChannel is ever called by ReactorNetty4HttpServerTransport, so when using the reactor netty implementation that httpChannels collection is always empty. I don't understand why the failure is intermittent though. @reta Do you have any idea what might be happening?

❌ Gradle check result for 4abed82: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

However, I don't see that AbstractHttpServerTransport#serverAcceptedChannel is ever called by ReactorNetty4HttpServerTransport, so when using the reactor netty implementation that httpChannels collection is always empty. I don't understand why the failure is intermittent though. @reta Do you have any idea what might be happening?

Looking @andrross , thanks for taking it @fdesu

@andrross @reta great point, lots of thanks for this hint!

I can confirm the source of failures was randomness in what transport gets picked for the smoke tests:

when ReactorNetty4Plugin was picked, some request channels might not get closed on the node shutdown which would cause issues in the OpenSearchTestClusterRule#clearClusters rule.

I've added a channel init lifecycle hook for the ReactorNetty4HttpServerTransport which is similar to Netty4HttpServerTransport. The added channel wrapper gets registered with the AbstractHttpServerTransport#httpChannels when a new channel gets accepted. That would cause all channels to get closed on the node shutdown which is expected behaviour.

@fdesu I just figured out that I think we do have much simpler solution for the problem, by modifying ReactorNetty4NonStreamingRequestConsumer::process method and adding transport.serverAcceptedChannel(channel):

void process(HttpContent in, FluxSink<HttpContent> emitter) {
        // Consume request body in full before dispatching it
        content.addComponent(true, in.content().retain());

        if (in instanceof LastHttpContent) {
            final ReactorNetty4NonStreamingHttpChannel channel = new ReactorNetty4NonStreamingHttpChannel(request, response, emitter);
            final HttpRequest r = createRequest(request, content);

            try {
                transport.serverAcceptedChannel(channel);
                transport.incomingRequest(r, channel);
            } catch (Exception ex) {
                emitter.error(ex);
                transport.onException(channel, ex);
            } finally {
                r.release();
                if (disposed.compareAndSet(false, true)) {
                    this.content.release();
                }
            }
        }
    }

The serverAcceptedChannel is not visible (protected) but we could add an override to ReactorNetty4HttpServerTransport and use ReactorNetty4HttpServerTransport instead of AbstractHttpServerTransport.

    @Override
    public void serverAcceptedChannel(HttpChannel httpChannel) {
        super.serverAcceptedChannel(httpChannel);
    }

We don't need wrappers and initializers in this case

@reta interesting, I assume the streaming request consumer would need to do the same for the LastHttpContent branch, correct?

I'm not entirely sure but the ReactorNetty4HttpChannel solution also looks pretty simple and being applied on the ReactorNetty4HttpServerTransport level. That makes it pretty independent of the consumers that are, I guess, only responsible for consuming the inbound request. What do you think?

✅ Gradle check result for 93a3457: SUCCESS