Assert engine not got by trasnport worker threads

Relates ES-12215

Author: kingherc

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

@@ -3389,6 +3389,8 @@ public Engine getEngineOrNull() {
     }
 
     private Engine getCurrentEngine(boolean allowNoEngine) {
+        assert engineResetLock.isWriteLockedByCurrentThread()
+            || 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) {
@@ -3441,7 +3443,7 @@ 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();
@@ -3467,7 +3469,7 @@ public <R, E extends Exception> R withEngineException(CheckedFunction<Engine, R,
      * @param <R>           the type of the result
      */
     private <R> R withEngine(Function<Engine, R> operation, boolean allowNoEngine) {
-        assert assertCurrentThreadWithEngine();
+        assert assertCurrentThreadWithEngine("method IndexShard#withEngine (or one of its variant) can block");
         assert operation != null;
 
         engineResetLock.readLock().lock();
@@ -3479,8 +3481,7 @@ private <R> R withEngine(Function<Engine, R> operation, boolean allowNoEngine) {
         }
     }
 
-    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);

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,15 @@ 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> seqNoStats = new SetOnce<>();
+            threadPool.generic().execute(() -> { seqNoStats.set(indexShard.seqNoStats()); });
+            while (seqNoStats.get() == null) {
+                try {
+                    Thread.sleep(10); // wait for seqNoStats to be set
+                } catch (InterruptedException e) {
+                    ;
+                }
+            }
 
             if (request.getFromSeqNo() > seqNoStats.getGlobalCheckpoint()) {
                 logger.trace(