Guard Get operations against Engine resets

Relates elastic#124635
Closes ES-11324

Author: fcofdez

server/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -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;
@@ -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
@@ -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 {

server/src/main/java/org/elasticsearch/action/get/TransportGetFromTranslogAction.java

@@ -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;
@@ -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);
         });

server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java

@@ -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;
@@ -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
@@ -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 {

server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetFomTranslogAction.java

@@ -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;
@@ -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);
         });

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -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));
     }
 
     /**
@@ -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)));
     }
 }