Skip to content

Guard Get operations against Engine resets #125646

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.

Sign up for GitHub

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

Merged
merged 6 commits into from
Apr 11, 2025
Merged

Guard Get operations against Engine resets #125646

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3aa270a
Guard Get operations against Engine resets
fcofdez Mar 26, 2025
a72dd02
Merge branch 'main' into ES-11324
fcofdez Mar 27, 2025
ce9b0c3
Merge branch 'main' into ES-11324
tlrx Apr 10, 2025
6e1255f
update + few changes
tlrx Apr 10, 2025
b0086d0
Merge branch 'main' into ES-11324
tlrx Apr 11, 2025
2e84ffb
Merge branch 'main' into ES-11324
tlrx Apr 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 2 additions & 12 deletions server/src/main/java/org/elasticsearch/action/get/TransportGetAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
Expand Down Expand Up @@ -235,10 +234,7 @@ private void getFromTranslog(
final var retryingListener = listener.delegateResponse((l, e) -> {
final var cause = ExceptionsHelper.unwrapCause(e);
logger.debug("get_from_translog failed", cause);
if (cause instanceof ShardNotFoundException
|| cause instanceof IndexNotFoundException
|| cause instanceof AlreadyClosedException) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
if (cause instanceof ShardNotFoundException || cause instanceof IndexNotFoundException) {
logger.debug("retrying get_from_translog");
observer.waitForNextChange(new ClusterStateObserver.Listener() {
@Override
Expand All @@ -253,13 +249,7 @@ public void onClusterServiceClose() {

@Override
public void onTimeout(TimeValue timeout) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
if (cause instanceof AlreadyClosedException) {
// Do an additional retry just in case AlreadyClosedException didn't generate a cluster update
tryGetFromTranslog(request, indexShard, node, l);
} else {
l.onFailure(new ElasticsearchException("Timed out retrying get_from_translog", cause));
}
l.onFailure(new ElasticsearchException("Timed out retrying get_from_translog", cause));
}
});
} else {
Expand Down
7 changes: 1 addition & 6 deletions server/src/main/java/org/elasticsearch/action/get/TransportGetFromTranslogAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@

package org.elasticsearch.action.get;

import org.apache.lucene.store.AlreadyClosedException;
import org.apache.lucene.util.BytesRef;
import org.elasticsearch.TransportVersions;
import org.elasticsearch.action.ActionListener;
Expand Down Expand Up @@ -77,11 +76,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
);
long segmentGeneration = -1;
if (result == null) {
Engine engine = indexShard.getEngineOrNull();
if (engine == null) {
throw new AlreadyClosedException("engine closed");
}
segmentGeneration = engine.getLastUnsafeSegmentGenerationForGets();
segmentGeneration = indexShard.withEngine(Engine::getLastUnsafeSegmentGenerationForGets);
}
return new Response(result, indexShard.getOperationPrimaryTerm(), segmentGeneration);
});
Expand Down
14 changes: 2 additions & 12 deletions server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
Expand Down Expand Up @@ -216,10 +215,7 @@ private void shardMultiGetFromTranslog(
final var retryingListener = listener.delegateResponse((l, e) -> {
final var cause = ExceptionsHelper.unwrapCause(e);
logger.debug("mget_from_translog[shard] failed", cause);
if (cause instanceof ShardNotFoundException
|| cause instanceof IndexNotFoundException
|| cause instanceof AlreadyClosedException) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
if (cause instanceof ShardNotFoundException || cause instanceof IndexNotFoundException) {
logger.debug("retrying mget_from_translog[shard]");
observer.waitForNextChange(new ClusterStateObserver.Listener() {
@Override
Expand All @@ -234,13 +230,7 @@ public void onClusterServiceClose() {

@Override
public void onTimeout(TimeValue timeout) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
if (cause instanceof AlreadyClosedException) {
// Do an additional retry just in case AlreadyClosedException didn't generate a cluster update
tryShardMultiGetFromTranslog(request, indexShard, node, l);
} else {
l.onFailure(new ElasticsearchException("Timed out retrying mget_from_translog[shard]", cause));
}
l.onFailure(new ElasticsearchException("Timed out retrying mget_from_translog[shard]", cause));
}
});
} else {
Expand Down
7 changes: 1 addition & 6 deletions ...r/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetFomTranslogAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@

package org.elasticsearch.action.get;

import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.TransportVersions;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequest;
Expand Down Expand Up @@ -97,11 +96,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
}
long segmentGeneration = -1;
if (someItemsNotFoundInTranslog) {
Engine engine = indexShard.getEngineOrNull();
if (engine == null) {
throw new AlreadyClosedException("engine closed");
}
segmentGeneration = engine.getLastUnsafeSegmentGenerationForGets();
segmentGeneration = indexShard.withEngine(Engine::getLastUnsafeSegmentGenerationForGets);
}
return new Response(multiGetShardResponse, indexShard.getOperationPrimaryTerm(), segmentGeneration);
});
Expand Down
9 changes: 3 additions & 6 deletions server/src/main/java/org/elasticsearch/index/shard/IndexShard.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1296,9 +1296,9 @@ private Engine.GetResult innerGet(Engine.Get get, boolean translogOnly, Function
throw new IllegalStateException("get operations not allowed on a legacy index");
}
if (translogOnly) {
return getEngine().getFromTranslog(get, mappingLookup, mapperService.documentParser(), searcherWrapper);
return withEngine(engine -> engine.getFromTranslog(get, mappingLookup, mapperService.documentParser(), searcherWrapper));
}
return getEngine().get(get, mappingLookup, mapperService.documentParser(), searcherWrapper);
return withEngine(engine -> engine.get(get, mappingLookup, mapperService.documentParser(), searcherWrapper));
}

/**
Expand Down Expand Up @@ -4698,9 +4698,6 @@ public void waitForPrimaryTermAndGeneration(long primaryTerm, long segmentGenera
* @param listener the listener to be notified when the shard is mutable
*/
public void ensureMutable(ActionListener<Void> listener) {
indexEventListener.beforeIndexShardMutableOperation(this, listener.delegateFailure((l, unused) -> {
// TODO ES-10826: Acquire ref to engine and retry if it's immutable again?
l.onResponse(null);
}));
indexEventListener.beforeIndexShardMutableOperation(this, listener.delegateFailure((l, unused) -> l.onResponse(null)));
Copy link
Contributor Author

Choose a reason for hiding this comment

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

I think that we shouldn't provide the Engine through this listener. It kind of breaks the withEngine API that expects that the Engine won't escape the lambda.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The unhollow/hollow timing should protect the engine from going hollow again.

Copy link
Member

Choose a reason for hiding this comment

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

Yes, I think we can remove this TODO.

}
}