Assert engine not got by trasnport worker threads

Relates ES-12215

Author: kingherc

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

@@ -1463,11 +1463,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) {
@@ -2669,7 +2670,33 @@ boolean shouldRollTranslogGeneration() {
     public void onSettingsChanged() {
         engineResetLock.readLock().lock();
         try {
-            var engine = getCurrentEngine(true);
+            // TODO this may be called by a cluster state update thread and we need to consider whether it is an issue
+            // java.lang.AssertionError: Expected current thread
+            // [Thread[#97,elasticsearch[node_t3][clusterApplierService#updateTask][T#1],5,TGRP-StatelessIT]] to not be the cluster state
+            // update thread. Reason: [method IndexShard#getCurrentEngine (or one of its variant) can block]
+            // at __randomizedtesting.SeedInfo.seed([6244501B70969C37]:0)
+            // at org.elasticsearch.cluster.service.ClusterApplierService.assertNotClusterStateUpdateThread(ClusterApplierService.java:386)
+            // at org.elasticsearch.index.shard.IndexShard.assertCurrentThreadWithEngine(IndexShard.java:3550)
+            // at org.elasticsearch.index.shard.IndexShard.getCurrentEngine(IndexShard.java:3430)
+            // at org.elasticsearch.index.shard.IndexShard.onSettingsChanged(IndexShard.java:2673)
+            // at org.elasticsearch.index.IndexService.updateMetadata(IndexService.java:1013)
+            // at org.elasticsearch.indices.cluster.IndicesClusterStateService.updateIndices(IndicesClusterStateService.java:662)
+            // at org.elasticsearch.indices.cluster.IndicesClusterStateService.doApplyClusterState(IndicesClusterStateService.java:322)
+            // at org.elasticsearch.indices.cluster.IndicesClusterStateService.applyClusterState(IndicesClusterStateService.java:278)
+            // at org.elasticsearch.cluster.service.ClusterApplierService.callClusterStateAppliers(ClusterApplierService.java:572)
+            // at org.elasticsearch.cluster.service.ClusterApplierService.callClusterStateAppliers(ClusterApplierService.java:558)
+            // at org.elasticsearch.cluster.service.ClusterApplierService.applyChanges(ClusterApplierService.java:531)
+            // at org.elasticsearch.cluster.service.ClusterApplierService.runTask(ClusterApplierService.java:460)
+            // at org.elasticsearch.cluster.service.ClusterApplierService$UpdateTask.run(ClusterApplierService.java:159)
+            // at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:1000)
+            // at
+            // org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:218)
+            // at
+            // org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:184)
+            // at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1095)
+            // at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:619)
+            // at java.base/java.lang.Thread.run(Thread.java:1447)
+            var engine = getCurrentEngine(true, true);
             if (engine != null) {
                 engine.onSettingsChanged();
             }
@@ -3053,7 +3080,11 @@ public void markAllocationIdAsInSync(final String allocationId, final long local
      * @return the local checkpoint
      */
     public long getLocalCheckpoint() {
-        return getEngine().getPersistedLocalCheckpoint();
+        return getLocalCheckpoint(false);
+    }
+
+    public long getLocalCheckpoint(boolean skipAssertions) {
+        return getEngine(skipAssertions).getPersistedLocalCheckpoint();
     }
 
     /**
@@ -3158,7 +3189,7 @@ public PendingReplicationActions getPendingReplicationActions() {
      */
     public void updateGlobalCheckpointOnReplica(final long globalCheckpoint, final String reason) {
         assert assertReplicationTarget();
-        final long localCheckpoint = getLocalCheckpoint();
+        final long localCheckpoint = getLocalCheckpoint(true);
         if (globalCheckpoint > localCheckpoint) {
             /*
              * This can happen during recovery when the shard has started its engine but recovery is not finalized and is receiving global
@@ -3398,9 +3429,14 @@ private void doCheckIndex() throws IOException {
      */
     @Deprecated
     Engine getEngine() {
+        return getEngine(false);
+    }
+
+    @Deprecated
+    Engine getEngine(boolean skipAssertions) {
         engineResetLock.readLock().lock();
         try {
-            return getCurrentEngine(false);
+            return getCurrentEngine(false, skipAssertions);
         } finally {
             engineResetLock.readLock().unlock();
         }
@@ -3416,13 +3452,19 @@ 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) {
+        // We only reset a shard when it's relocating (primary relocation) or started (unhollowing)
+        boolean shardRoutingEngineResettable = shardRouting.started() || shardRouting.relocating();
+        assert skipAssertions
+            || shardRoutingEngineResettable == false
+            || state() != IndexShardState.STARTED // exclude getting the engine when not started. We only reset started shards.
+            || 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) {
@@ -3489,12 +3531,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();
@@ -3526,12 +3568,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();
@@ -3541,8 +3583,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);
@@ -4605,7 +4646,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(