Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix concurrent search and index delete #42621

Expand Up @@ -44,6 +44,7 @@
import org.elasticsearch.common.util.concurrent.ConcurrentMapLong;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.engine.Engine;
Expand Down Expand Up @@ -550,12 +551,19 @@ final SearchContext createAndPutContext(ShardSearchRequest request) throws IOExc
SearchContext context = createContext(request);
boolean success = false;
try {
putContext(context);
if (request.scroll() != null) {
openScrollContexts.incrementAndGet();
context.indexShard().getSearchOperationListener().onNewScrollContext(context);
}
context.indexShard().getSearchOperationListener().onNewContext(context);
putContext(context);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't understand why you moved the putContext(context)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By registering the search context in activeContexts after having invoked onNewContext/onNewScrollContext, we guarantee that for a specific SearchContext, the call to onNewXXX happens before the matching call the onFreeXXX.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see I didn't think about SearchService.afterIndexRemoved. my issue here is that we have to call onFreeXXX but if one of the onNewContext /onNewScrollContext fails we don't register and fail? I think we need extra protection for this?

// ensure that if index is deleted concurrently, we free the context immediately, either here or in afterIndexRemoved
try {
indicesService.indexServiceSafe(request.shardId().getIndex());
} catch (IndexNotFoundException e) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think you need the catch clause here - we free below

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How did I miss that, thanks! Fixed in 8edda7b

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but is this necessary or can we just rearrange the registration?

freeContext(context.id());
throw e;
}
success = true;
return context;
} finally {
Expand Down
Expand Up @@ -19,7 +19,6 @@
package org.elasticsearch.search;

import com.carrotsearch.hppc.IntArrayList;

import org.apache.lucene.search.Query;
import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.ElasticsearchException;
Expand Down Expand Up @@ -261,12 +260,16 @@ public void onFailure(Exception e) {
try {
final int rounds = scaledRandomIntBetween(100, 10000);
SearchRequest searchRequest = new SearchRequest().allowPartialSearchResults(true);
SearchRequest scrollSearchRequest = new SearchRequest().allowPartialSearchResults(true)
.scroll(new Scroll(TimeValue.timeValueMinutes(1)));
for (int i = 0; i < rounds; i++) {
try {
try {
PlainActionFuture<SearchPhaseResult> result = new PlainActionFuture<>();
boolean useScroll = randomBoolean();
service.executeQueryPhase(
new ShardSearchLocalRequest(searchRequest, indexShard.shardId(), 1,
new ShardSearchLocalRequest(useScroll ? searchRequest : scrollSearchRequest,
indexShard.shardId(), 1,
new AliasFilter(null, Strings.EMPTY_ARRAY), 1.0f, -1, null, null),
new SearchTask(123L, "", "", "", null, Collections.emptyMap()), result);
SearchPhaseResult searchPhaseResult = result.get();
Expand All @@ -276,6 +279,9 @@ public void onFailure(Exception e) {
PlainActionFuture<FetchSearchResult> listener = new PlainActionFuture<>();
service.executeFetchPhase(req, new SearchTask(123L, "", "", "", null, Collections.emptyMap()), listener);
listener.get();
if (useScroll) {
service.freeContext(searchPhaseResult.requestId);
}
} catch (ExecutionException ex) {
assertThat(ex.getCause(), instanceOf(RuntimeException.class));
throw ((RuntimeException)ex.getCause());
Expand Down