Assert engine not got by trasnport worker threads

Relates ES-12215

Author: kingherc

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

@@ -1458,11 +1458,12 @@ public StoreStats storeStats() {
     }
 
     public MergeStats mergeStats() {
-        final Engine engine = getEngineOrNull();
-        if (engine == null) {
-            return new MergeStats();
-        }
-        return engine.getMergeStats();
+        return tryWithEngineOrNull(engine -> {
+            if (engine == null) {
+                return new MergeStats();
+            }
+            return engine.getMergeStats();
+        });
     }
 
     public SegmentsStats segmentStats(boolean includeSegmentFileSizes, boolean includeUnloadedSegments) {
@@ -2664,7 +2665,7 @@ boolean shouldRollTranslogGeneration() {
     public void onSettingsChanged() {
         engineResetLock.readLock().lock();
         try {
-            var engine = getCurrentEngine(true);
+            var engine = getCurrentEngine(true, false);
             if (engine != null) {
                 engine.onSettingsChanged();
             }
@@ -3379,7 +3380,7 @@ private void doCheckIndex() throws IOException {
     Engine getEngine() {
         engineResetLock.readLock().lock();
         try {
-            return getCurrentEngine(false);
+            return getCurrentEngine(false, false);
         } finally {
             engineResetLock.readLock().unlock();
         }
@@ -3395,13 +3396,17 @@ Engine getEngine() {
     public Engine getEngineOrNull() {
         engineResetLock.readLock().lock();
         try {
-            return getCurrentEngine(true);
+            return getCurrentEngine(true, false);
         } finally {
             engineResetLock.readLock().unlock();
         }
     }
 
-    private Engine getCurrentEngine(boolean allowNoEngine) {
+    private Engine getCurrentEngine(boolean allowNoEngine, boolean skipAssertions) {
+        assert skipAssertions
+            || shardRouting.started() == false // exclude getting the engine when not started. We only reset started engines.
+            || state() != IndexShardState.STARTED // exclude getting the engine when not started. We only reset started engines.
+            || assertCurrentThreadWithEngine("method IndexShard#getCurrentEngine (or one of its variant) can block");
         assert engineResetLock.isReadLockedByCurrentThread() || engineResetLock.isWriteLockedByCurrentThread() /* for resets */;
         var engine = currentEngine.get();
         if (engine == null && allowNoEngine == false) {
@@ -3468,12 +3473,12 @@ public <R> R withEngine(Function<Engine, R> operation) {
      * @throws              AlreadyClosedException if the current engine instance is {@code null}.
      */
     public <R, E extends Exception> R withEngineException(CheckedFunction<Engine, R, E> operation) throws E {
-        assert assertCurrentThreadWithEngine();
+        assert assertCurrentThreadWithEngine("method IndexShard#withEngineException (or one of its variant) can block");
         assert operation != null;
 
         engineResetLock.readLock().lock();
         try {
-            var engine = getCurrentEngine(false);
+            var engine = getCurrentEngine(false, false);
             return operation.apply(engine);
         } finally {
             engineResetLock.readLock().unlock();
@@ -3505,12 +3510,12 @@ private <R> R withEngine(Function<Engine, R> operation, boolean allowNoEngine, b
         if (noEngineDuringReset) {
             locked = engineResetLock.readLock().tryLock();
         } else {
-            assert assertCurrentThreadWithEngine();
+            assert assertCurrentThreadWithEngine("method IndexShard#withEngine (or one of its variant) can block");
             engineResetLock.readLock().lock();
         }
         if (locked) {
             try {
-                var engine = getCurrentEngine(allowNoEngine);
+                var engine = getCurrentEngine(allowNoEngine, noEngineDuringReset);
                 return operation.apply(engine);
             } finally {
                 engineResetLock.readLock().unlock();
@@ -3520,8 +3525,7 @@ private <R> R withEngine(Function<Engine, R> operation, boolean allowNoEngine, b
         }
     }
 
-    private static boolean assertCurrentThreadWithEngine() {
-        var message = "method IndexShard#withEngine (or one of its variant) can block";
+    private static boolean assertCurrentThreadWithEngine(String message) {
         assert ClusterApplierService.assertNotClusterStateUpdateThread(message);
         assert MasterService.assertNotMasterUpdateThread(message);
         assert Transports.assertNotTransportThread(message);
@@ -4551,7 +4555,7 @@ public void resetEngine(Consumer<Engine> postResetNewEngineConsumer) {
                 try {
                     engineResetLock.writeLock().lock();
                     try {
-                        var engine = getCurrentEngine(false);
+                        var engine = getCurrentEngine(false, true);
                         engine.prepareForEngineReset();
                         var newEngine = createEngine(newEngineConfig(replicationTracker));
                         getAndSetCurrentEngine(newEngine);

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardChangesAction.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.ccr.action;
 
+import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -403,7 +404,12 @@ protected void asyncShardOperation(final Request request, final ShardId shardId,
             throws IOException {
             final IndexService indexService = indicesService.indexServiceSafe(request.getShard().getIndex());
             final IndexShard indexShard = indexService.getShard(request.getShard().id());
-            final SeqNoStats seqNoStats = indexShard.seqNoStats();
+            final SetOnce<SeqNoStats> seqNoStatsSetOnce = new SetOnce<>();
+            threadPool.generic().execute(() -> { seqNoStatsSetOnce.set(indexShard.seqNoStats()); });
+            while (seqNoStatsSetOnce.get() == null) {
+                ;
+            }
+            final SeqNoStats seqNoStats = seqNoStatsSetOnce.get();
 
             if (request.getFromSeqNo() > seqNoStats.getGlobalCheckpoint()) {
                 logger.trace(