Repository navigation
opensearch-java: send bulk requests with the OpenSearchAsyncClient - #2225
Merged
Merged
Conversation
AsyncBulkProcessor ran the blocking OpenSearchClient.bulk() call on a ThreadPoolExecutor bounded to concurrentRequests threads, with a SynchronousQueue and CallerRunsPolicy. Since the semaphore permit was released before the listener processed the response, the bolt could submit the next request while the only worker was still in afterBulk; the task was then rejected and the bulk call executed on the bolt's own thread, blocking it for the whole HTTP round trip. Use OpenSearchAsyncClient instead, as the legacy BulkProcessor did with bulkAsync, and release the permit only once the listener has run. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Contributor
Author
|
This brings the performance on par with that of the legacy module. |
dpol1
approved these changes
Oct 7, 2026
dpol1
left a comment
Contributor
There was a problem hiding this comment.
Thanks for tracking this down! The pool with CallerRunsPolicy from #1868 was the culprit: with the permit released before afterBulk, under load the next bulk ended up running on the calling thread. LGTM, good to see it on par with the legacy module.
| */ | ||
| @Test | ||
| @Timeout(10) | ||
| void addDoesNotWaitForBulkResponse() throws Exception { |
Contributor
There was a problem hiding this comment.
One optional follow-up: I have a small regression test that holds afterBulk on a latch and checks that a second add() waits. It catches the permit release moving back before the callback. Happy to contribute it.
rzo1
approved these changes
Oct 7, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The
StatusUpdaterBoltof theopensearch-javamodule is slower than the one of the legacyopensearchmodule. A likely cause is the wayAsyncBulkProcessorexecutes the bulk requests.It ran the blocking
OpenSearchClient.bulk()call on aThreadPoolExecutorbounded toconcurrentRequeststhreads, with aSynchronousQueueandCallerRunsPolicy. The semaphore permit was released before the listener processed the response, so the bolt could acquire it and submit the next request while the only worker (with the defaultconcurrentRequests=1) was still busy inafterBulk. The task was then rejected and the bulk call ran on the bolt's own executor thread, blocking it for the whole HTTP round trip instead of letting it fill the next batch.Changes
AsyncBulkProcessorusesOpenSearchAsyncClient.bulk(), like the legacyBulkProcessorusedbulkAsync; the dedicated executor is removed.add()blocks only whenconcurrentRequestsbulks are in flight.OpenSearchConnectionbuilds the async client on the same transport as the sync one.add()does not wait for the bulk response.Note that
afterBulknow runs on the HTTP client's I/O reactor thread, which is also where the legacy module ran it.🤖 Generated with Claude Code