defer engine

Author: tlrx

server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -50,6 +50,7 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexShardClosedException;
@@ -78,6 +79,7 @@
 import java.util.Optional;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 
@@ -507,7 +509,8 @@ public void handleException(TransportException exp) {
 
                         if (syncGlobalCheckpointAfterOperation) {
                             try {
-                                primaryShardReference.indexShard.maybeSyncGlobalCheckpoint("post-operation");
+                                var seqNoStats = primaryShardReference.seqNoStatsSupplier.get();
+                                primaryShardReference.indexShard.maybeSyncGlobalCheckpoint("post-operation", seqNoStats);
                             } catch (final Exception e) {
                                 // only log non-closed exceptions
                                 if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
@@ -1136,10 +1139,16 @@ class PrimaryShardReference
 
         protected final IndexShard indexShard;
         private final Releasable operationLock;
+        private final Supplier<SeqNoStats> seqNoStatsSupplier;
+        private final Supplier<Long> localCheckpointSupplier;
+        private final Supplier<Long> globalCheckpointSupplier;
 
         PrimaryShardReference(IndexShard indexShard, Releasable operationLock) {
             this.indexShard = indexShard;
             this.operationLock = operationLock;
+            this.seqNoStatsSupplier = indexShard.getSeqNoStatsSupplier();
+            this.localCheckpointSupplier = indexShard.getLocalCheckpointSupplier();
+            this.globalCheckpointSupplier = indexShard.getLastSyncedGlobalCheckpointSupplier();
         }
 
         @Override
@@ -1189,12 +1198,12 @@ public void updateGlobalCheckpointForShard(final String allocationId, final long
 
         @Override
         public long localCheckpoint() {
-            return indexShard.getLocalCheckpoint();
+            return localCheckpointSupplier.get();
         }
 
         @Override
         public long globalCheckpoint() {
-            return indexShard.getLastSyncedGlobalCheckpoint();
+            return globalCheckpointSupplier.get();
         }
 
         @Override

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

@@ -1351,6 +1351,18 @@ public SeqNoStats seqNoStats() {
         return getEngine().getSeqNoStats(replicationTracker.getGlobalCheckpoint());
     }
 
+    /**
+     * Returns a supplier that supplies the {@link SeqNoStats} of the engine that is referenced at the time this method is called.
+     * Uses this method in place where the current engine reference cannot be resolved directly.
+     *
+     * @return a supplier of {@link SeqNoStats}
+     * @throws AlreadyClosedException if shard is closed
+     */
+    public Supplier<SeqNoStats> getSeqNoStatsSupplier() {
+        var engine = getEngine();
+        return () -> engine.getSeqNoStats(replicationTracker.getGlobalCheckpoint());
+    }
+
     public IndexingStats indexingStats() {
         Engine engine = getEngineOrNull();
         final boolean throttled;
@@ -2974,6 +2986,18 @@ public long getLocalCheckpoint() {
         return getEngine().getPersistedLocalCheckpoint();
     }
 
+    /**
+     * Returns a supplier that supplies the local checkpoint of the engine that is referenced at the time this method is called.
+     * Uses this method in place where the current engine reference cannot be resolved directly.
+     *
+     * @return a supplier of the local checkpoint
+     * @throws AlreadyClosedException if shard is closed
+     */
+    public Supplier<Long> getLocalCheckpointSupplier() {
+        var engine = getEngine();
+        return () -> engine.getPersistedLocalCheckpoint();
+    }
+
     /**
      * Returns the global checkpoint for the shard.
      *
@@ -2990,6 +3014,18 @@ public long getLastSyncedGlobalCheckpoint() {
         return getEngine().getLastSyncedGlobalCheckpoint();
     }
 
+    /**
+     * Returns a supplier that supplies the latest global checkpoint of the engine that is referenced at the time this method is called.
+     * Uses this method in place where the current engine reference cannot be resolved directly.
+     *
+     * @return a supplier of the latest global checkpoint value that has been persisted in the underlying storage
+     * @throws AlreadyClosedException if shard is closed
+     */
+    public Supplier<Long> getLastSyncedGlobalCheckpointSupplier() {
+        var engine = getEngine();
+        return () -> engine.getLastSyncedGlobalCheckpoint();
+    }
+
     /**
      * Get the local knowledge of the global checkpoints for all in-sync allocation IDs.
      *
@@ -3012,8 +3048,22 @@ public void maybeSyncGlobalCheckpoint(final String reason) {
             return;
         }
         assert assertPrimaryMode();
-        // only sync if there are no operations in flight, or when using async durability
         final SeqNoStats stats = getEngine().getSeqNoStats(replicationTracker.getGlobalCheckpoint());
+        doMaybeSyncGlobalCheckpoint(reason, stats);
+    }
+
+    public void maybeSyncGlobalCheckpoint(final String reason, final SeqNoStats stats) {
+        verifyNotClosed();
+        assert shardRouting.primary() : "only call maybeSyncGlobalCheckpoint on primary shard";
+        if (replicationTracker.isPrimaryMode() == false) {
+            return;
+        }
+        assert assertPrimaryMode();
+        doMaybeSyncGlobalCheckpoint(reason, stats);
+    }
+
+    private void doMaybeSyncGlobalCheckpoint(final String reason, final SeqNoStats stats) {
+        // only sync if there are no operations in flight, or when using async durability
         final boolean asyncDurability = indexSettings().getTranslogDurability() == Translog.Durability.ASYNC;
         if (stats.getMaxSeqNo() == stats.getGlobalCheckpoint() || asyncDurability) {
             final var trackedGlobalCheckpointsNeedSync = replicationTracker.trackedGlobalCheckpointsNeedSync();