Skip to content

Support real-time gets on hollow shards #122012

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 27 commits into from
Feb 11, 2025
Merged

Support real-time gets on hollow shards #122012

Show file tree
Hide file tree
Changes from 25 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
db73b39
Support returning latest hollow generation real-time GET requests
arteam Feb 7, 2025
0bced20
Also support multi-get requests
arteam Feb 8, 2025
44fed83
Merge remote-tracking branch 'origin/main' into hollow-real-time-gets
arteam Feb 8, 2025
3237dfa
Make TransportGetFromTranslogAction and TransportShardMultiGetFomTran…
arteam Feb 9, 2025
47b5083
Revert "Support returning latest hollow generation real-time GET requ…
arteam Feb 9, 2025
f207d15
Bring back original TransportGetFromTranslogAction and TransportShard…
arteam Feb 9, 2025
75431f7
Merge remote-tracking branch 'origin/main' into hollow-real-time-gets
arteam Feb 9, 2025
40301f2
Retry real-time gets for AlreadyClosedException
arteam Feb 10, 2025
ff0c191
Correctly check index service in case index is not a different node
arteam Feb 10, 2025
634cbf4
Merge branch 'main' into hollow-real-time-gets
arteam Feb 10, 2025
e8b47d2
Remove access to multi get request items
arteam Feb 10, 2025
195f5fb
Support returning latest hollow generation real-time GET requests
arteam Feb 7, 2025
0ca0c78
Throw UnsupportedOperationExceptions for `getLastUnsafeSegmentGenerat…
arteam Feb 10, 2025
87240c5
Also support multi-get requests
arteam Feb 8, 2025
25e5e9d
Call getLastUnsafeSegmentGenerationForGets directly on Engine
arteam Feb 10, 2025
ed8c902
Revert "Make TransportGetFromTranslogAction and TransportShardMultiGe…
arteam Feb 10, 2025
07ab4b7
Merge branch 'main' into hollow-real-time-gets
arteam Feb 10, 2025
ecf8aba
Merge branch 'main' into hollow-real-time-gets
arteam Feb 11, 2025
4806d11
Retry getTranslog calls if we encountered `AlreadyClosedException` an…
arteam Feb 11, 2025
1c03540
Merge branch 'main' into hollow-real-time-gets
arteam Feb 11, 2025
b965926
Revert "Correctly check index service in case index is not a differen…
arteam Feb 11, 2025
5d85f39
Merge branch 'main' into hollow-real-time-gets
arteam Feb 11, 2025
847bf5c
Correctly check index service in case index is not a different node
arteam Feb 10, 2025
4bdcafb
Use try versions of get*Translog calls so we don't retry forever
arteam Feb 11, 2025
01b81bb
Add TODOs for ES-10826
arteam Feb 11, 2025
8d4ab8e
Revert "Correctly check index service in case index is not a differen…
arteam Feb 11, 2025
d29028c
Merge branch 'main' into hollow-real-time-gets
arteam Feb 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
18 changes: 15 additions & 3 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,6 +11,7 @@

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 @@ -165,7 +166,9 @@ protected Executor getExecutor(GetRequest request, ShardId shardId) {
final ClusterState clusterState = clusterService.state();
if (clusterState.metadata().getIndexSafe(shardId.getIndex()).isSystem()) {
return threadPool.executor(executorSelector.executorForGet(shardId.getIndexName()));
} else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) {
}
var indexService = indicesService.indexService(shardId.getIndex());
if (indexService != null && indexService.getIndexSettings().isSearchThrottled()) {
return threadPool.executor(ThreadPool.Names.SEARCH_THROTTLED);
} else {
return super.getExecutor(request, shardId);
Expand Down Expand Up @@ -230,7 +233,10 @@ 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) {
if (cause instanceof ShardNotFoundException
|| cause instanceof IndexNotFoundException
|| cause instanceof AlreadyClosedException) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
logger.debug("retrying get_from_translog");
observer.waitForNextChange(new ClusterStateObserver.Listener() {
@Override
Expand All @@ -245,7 +251,13 @@ public void onClusterServiceClose() {

@Override
public void onTimeout(TimeValue timeout) {
l.onFailure(new ElasticsearchException("Timed out retrying get_from_translog", cause));
// 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));
}
}
});
} else {
Expand Down
2 changes: 1 addition & 1 deletion 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 @@ -81,7 +81,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
if (engine == null) {
throw new AlreadyClosedException("engine closed");
}
segmentGeneration = ((InternalEngine) engine).getLastUnsafeSegmentGenerationForGets();
segmentGeneration = engine.getLastUnsafeSegmentGenerationForGets();
}
return new Response(result, indexShard.getOperationPrimaryTerm(), segmentGeneration);
});
Expand Down
18 changes: 15 additions & 3 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,6 +11,7 @@

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 @@ -154,7 +155,9 @@ protected Executor getExecutor(MultiGetShardRequest request, ShardId shardId) {
final ClusterState clusterState = clusterService.state();
if (clusterState.metadata().index(shardId.getIndex()).isSystem()) {
return threadPool.executor(executorSelector.executorForGet(shardId.getIndexName()));
} else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) {
}
var indexService = indicesService.indexService(shardId.getIndex());
Copy link
Contributor

@fcofdez fcofdez Feb 11, 2025

Choose a reason for hiding this comment

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

why is this change needed?

Copy link
Contributor Author

@arteam arteam Feb 11, 2025

Choose a reason for hiding this comment

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

https://github.com/elastic/elasticsearch/pull/122012/files#r1950399319

I've seen an IndexNotFoundException being thrown here during the stress testing of real-time gets which failed the request, because we don't expect it here. Let me double check if it's still the case.

Copy link
Contributor Author

@arteam arteam Feb 11, 2025

Choose a reason for hiding this comment

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

It used to be failing with something like

Caused by: java.lang.IllegalStateException: unexpected exception processing cluster state version [112] in context [ClusterStateObserver[null]]
	at org.elasticsearch.cluster.ClusterStateObserver$ObserverClusterStateListener.logUnexpectedException(ClusterStateObserver.java:317)
	... 12 more
Caused by: [rjsadmjle/ScBnoey6QYqfReJ_JhmkhA] org.elasticsearch.index.IndexNotFoundException: no such index [rjsadmjle]
	at org.elasticsearch.indices.IndicesService.indexServiceSafe(IndicesService.java:586)
	at org.elasticsearch.action.get.TransportGetAction.getExecutor(TransportGetAction.java:169)
	at org.elasticsearch.action.get.TransportGetAction.tryGetFromTranslog(TransportGetAction.java:297)
	at org.elasticsearch.action.get.TransportGetAction.getFromTranslog(TransportGetAction.java:263)
	at org.elasticsearch.action.get.TransportGetAction$1.onNewClusterState(TransportGetAction.java:241)

but it doesn't anymore, so I've reverted in in b965926

Copy link
Contributor

@fcofdez fcofdez Feb 11, 2025

Choose a reason for hiding this comment

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

I still see the change, maybe you forgot to change this branch?

Copy link
Contributor

@kingherc kingherc Feb 11, 2025

Choose a reason for hiding this comment

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

@arteam can we revert this change here (and in GET) if it's not needed? Before I approve.

Copy link
Contributor Author

@arteam arteam Feb 11, 2025

Choose a reason for hiding this comment

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

@kingherc My bad, it came back with another commit. Reverted in 8d4ab8e

if (indexService != null && indexService.getIndexSettings().isSearchThrottled()) {
return threadPool.executor(ThreadPool.Names.SEARCH_THROTTLED);
} else {
return super.getExecutor(request, shardId);
Expand Down Expand Up @@ -211,7 +214,10 @@ 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) {
if (cause instanceof ShardNotFoundException
|| cause instanceof IndexNotFoundException
|| cause instanceof AlreadyClosedException) {
// TODO AlreadyClosedException the engine reset should be fixed by ES-10826
logger.debug("retrying mget_from_translog[shard]");
observer.waitForNextChange(new ClusterStateObserver.Listener() {
@Override
Expand All @@ -226,7 +232,13 @@ public void onClusterServiceClose() {

@Override
public void onTimeout(TimeValue timeout) {
l.onFailure(new ElasticsearchException("Timed out retrying mget_from_translog[shard]", cause));
// 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));
}
}
});
} else {
Expand Down
3 changes: 1 addition & 2 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 @@ -23,7 +23,6 @@
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.engine.InternalEngine;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.indices.IndicesService;
Expand Down Expand Up @@ -102,7 +101,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
if (engine == null) {
throw new AlreadyClosedException("engine closed");
}
segmentGeneration = ((InternalEngine) engine).getLastUnsafeSegmentGenerationForGets();
segmentGeneration = engine.getLastUnsafeSegmentGenerationForGets();
}
return new Response(multiGetShardResponse, indexShard.getOperationPrimaryTerm(), segmentGeneration);
});
Expand Down
4 changes: 4 additions & 0 deletions server/src/main/java/org/elasticsearch/index/engine/Engine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2346,4 +2346,8 @@ public record FlushResult(boolean flushPerformed, long generation) {
public void prepareForEngineReset() throws IOException {
throw new UnsupportedOperationException("does not support engine reset");
}

public long getLastUnsafeSegmentGenerationForGets() {
throw new UnsupportedOperationException("Doesn't support getting the latest segment generation");
}
}