Use RefCountingRunnable when creating indices in TransportBulkAction #103786
Conversation
gmarouli
added
:Data Management/Indices APIs
elasticsearchmachine
added
Team:Data Management
|
Pinging @elastic/es-data-management (Team:Data Management) |
| forkExecuteBulk(ActionListener.wrap(listener::onResponse, inner -> { | ||
| inner.addSuppressed(e); | ||
| listener.onFailure(inner); | ||
| })); |
There was a problem hiding this comment.
I am aware that this wrapping of the listener is not happening in the new code. I will explain why:
If my understanding is correct, this wrapping of the listener only happens if the last processed create index response is a failure. For example, the following two scenarios that are semantically equivalent will call or not call this code.
Scenario 1:
- Creating index "good" succeeds
- Creating index "invalid" fails
- The following
forkExecuteBulkwill be executed:
forkExecuteBulk(ActionListener.wrap(listener::onResponse, inner -> {
inner.addSuppressed(e);
listener.onFailure(inner);
}))
Scenario 2:
- Creating index "invalid" fails
- Creating index "good" succeeds
- The following
forkExecuteBulkwill be executed:
forkExecuteBulk(listener)
For this reason I believe the simplicity of the new code is better.
| public void onResponse(CreateIndexResponse result) { | ||
| if (counter.decrementAndGet() == 0) { | ||
| forkExecuteBulk(listener); | ||
| createIndex(index, bulkRequest.timeout(), refs.acquireListener().delegateResponse((l, e) -> { |
There was a problem hiding this comment.
Chaining delegateResponse and map in one expression is kind of suspicious, you're effectively overriding both onResponse and onFailure at the same time. Maybe just create a new ActionListener directly, with something like ActionListener.releaseAfter(new ActionListener<>{...}, refs.acquire()).
There was a problem hiding this comment.
Thank you @DaveCTurner ! Good point, let me try it.
|
I see some failures locally during manual testing. I am investigating further. |
|
@elasticmachine update branch |
After extensive testing I wasn't able to reproduce it. What I saw initially, was Kibana reporting that the index already exists. |
|
@DaveCTurner thank you for the feedback and the quick reactions! |
When an index can be auto-created, the
TransportBulkActionwill first try to create all the missing indices and then it's going to index the documents. This means that it's going to perform a number of async create index requests. This number is equal to the number of distinct indices that can be auto-created in the request.In this refactoring we use
RefCountingRunnableto facilitate these async request. TheRefCountingRunnableis already handling gracefully the different calls and can execute the bulk request when they all complete.This will also makes the code easier to extend when adding more pre-requisite actions, such as #103309.