Introduce OpenSearchStreamingTransport

[reta]

Description

Introduce OpenSearchStreamingTransport

Issues Resolved

N/A

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[Copilot]

🟡 Changes recommended

Streaming framing, retries, dependency publication, option handling, and CI syntax contain blocking correctness issues.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Introduces reactive streaming bulk operations across the Java client’s transport implementations.

Changes:

• Adds streaming transport and endpoint APIs.
• Implements streaming bulk requests and responses.
• Adds integration coverage and CI configuration.

File summaries

File	Description
StreamingIT.java (restclient)	Tests REST-client streaming.
StreamingIT.java (httpclient5)	Tests HttpClient 5 streaming.
AbstractStreamingIT.java	Defines shared streaming integration tests.
StreamingEndpoint.java	Defines streaming endpoint behavior.
RestClientTransport.java	Implements streaming through RestClient.
OpenSearchStreamingTransport.java	Adds the streaming transport interface.
ByteBufferInputStream.java	Adapts byte buffers to an input stream.
StreamingResponse.java	Wraps reactive HTTP responses.
StreamingRequest.java	Models reactive HTTP requests.
ResponseWarningsExtractor.java	Extracts warning headers.
Response.java	Reuses warning extraction logic.
ApacheHttpClient5Transport.java	Implements reactive streaming transport.
ApacheHttpClient5Options.java	Adds query-parameter storage.
SimpleStreamingEndpoint.java	Provides a streaming endpoint implementation.
OpenSearchClient.java	Exposes streaming bulk operations.
BulkStreamingResponse.java	Models streaming bulk responses.
BulkStreamingRequest.java	Models streaming bulk requests.
build.gradle.kts	Configures streaming tests and dependencies.
CHANGELOG.md	Records the new transport.
test-integration-unreleased.yml	Configures streaming integration infrastructure.

Review details

Suppressed comments (4)

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5Transport.java:296

• Framing each network ByteBuffer independently is not valid because HTTP buffer boundaries need not align with CRLF-delimited records. A delimiter can span two buffers, and a buffer with no delimiter is scanned to its limit and then returned exhausted by frame(), dropping those bytes entirely. Use a stateful framing operator that carries partial records across publisher emissions before deserializing them.

                        new Message<>(message.getHead(), Flux.from(message.getBody()).flatMapSequential(b -> Flux.fromIterable(frame(b))))

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5Transport.java:482

• The synthetic response omits all headers from the actual response, so warningsHandler always receives an empty warning list and ResponseException also loses response headers. Copy the response headers before constructing the wrapper.

        final ClassicHttpResponse httpResponse = new BasicClassicHttpResponse(
            message.getHead().getCode(),
            message.getHead().getReasonPhrase()
        );
        final Response response = new Response(new RequestLine(request), node.getHost(), httpResponse);

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5Transport.java:501

• For every non-ignored error status, this converts the response into an error signal while the synthetic ResponseException has no streaming entity. getHighLevelStreamingResponse() can recover the status but getBody() only re-emits that exception, so endpoint error deserialization fails for statuses such as 409/413 and after retries are exhausted. Preserve the final message/body for high-level error decoding while retaining retry decisions.

        ResponseException responseException = new ResponseException(response);
        if (isRetryStatus(statusCode)) {
            // mark host dead and retry against next one
            onFailure(node);
            return new ResponseOrResponseException(responseException);
        }
        // mark host alive and don't retry, as the error should be a request problem
        onResponse(node);
        throw responseException;

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5Transport.java:300

• Retrying by resubscribing to the same request-body Publisher is unsafe: publishers such as Flux.fromStream(...) are one-shot, and a streaming bulk request may already have applied operations before a retryable response arrives. A retry can therefore fail on resubscription or duplicate writes. Streaming retries must be disabled unless the API has an explicit replay/resume contract.

                    if (nodeTuple.nodes.hasNext()) {
                        return Mono.from(streamRequest(nodeTuple, options, request, requestBodyPublisher, warningsHandler));

• Files reviewed: 20/20 changed files
• Comments generated: 8
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.