Track max seq_no of updates or deletes on primary

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -88,6 +88,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -126,6 +127,9 @@ public abstract class Engine implements Closeable {
      *  inactive shards.
      */
     protected volatile long lastWriteNanos = System.nanoTime();
+    // The maximum sequence number of either update or delete operations have been processed by this engine.
+    // This value is started with an unassigned status (-2) and will be initialized from outside.
+    private final AtomicLong maxSeqNoOfUpdatesOrDeletes = new AtomicLong(SequenceNumbers.UNASSIGNED_SEQ_NO);
 
     protected Engine(EngineConfig engineConfig) {
         Objects.requireNonNull(engineConfig.getStore(), "Store must be provided to the engine");
@@ -1694,4 +1698,25 @@ public boolean isRecovering() {
     public interface TranslogRecoveryRunner {
         int run(Engine engine, Translog.Snapshot snapshot) throws IOException;
     }
+
+    /**
+     * Returns the maximum sequence number of either update or delete operations have been processed
+     * in this engine or the sequence number from {@link #advanceMaxSeqNoOfUpdatesOrDeletes(long)}.
+     * <p>
+     * For a primary engine, this value is initialized once, then advanced internally when it processes
+     * an update or a delete operation. Whereas a replica engine never updates this value by itself but
+     * only inherits the latest value from its primary. In both cases, this value never goes backwards.
+     */
+    public final long getMaxSeqNoOfUpdatesOrDeletes() {
+        return maxSeqNoOfUpdatesOrDeletes.get();
+    }
+
+    /**
+     * Advances the max_seq_no_of_updates marker of this engine to at least the given sequence number.
+     * @see #getMaxSeqNoOfUpdatesOrDeletes()
+     */
+    public final void advanceMaxSeqNoOfUpdatesOrDeletes(long seqNo) {
+        maxSeqNoOfUpdatesOrDeletes.updateAndGet(curr -> Math.max(curr, seqNo));
+        assert maxSeqNoOfUpdatesOrDeletes.get() >= seqNo;
+    }
 }

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -922,11 +922,13 @@ protected IndexingStrategy indexingStrategyForOperation(final Index index) throw
 
     protected final IndexingStrategy planIndexingAsPrimary(Index index) throws IOException {
         assert index.origin() == Operation.Origin.PRIMARY : "planing as primary but origin isn't. got " + index.origin();
+        assert getMaxSeqNoOfUpdatesOrDeletes() != SequenceNumbers.UNASSIGNED_SEQ_NO : "max_seq_no_of_updates is not initialized";
         final IndexingStrategy plan;
         // resolve an external operation into an internal one which is safe to replay
         if (canOptimizeAddDocument(index)) {
             if (mayHaveBeenIndexedBefore(index)) {
                 plan = IndexingStrategy.overrideExistingAsIfNotThere(generateSeqNoForOperation(index), 1L);
+                advanceMaxSeqNoOfUpdatesOrDeletes(plan.seqNoForIndexing);
                 versionMap.enforceSafeAccess();
             } else {
                 plan = IndexingStrategy.optimizedAppendOnly(generateSeqNoForOperation(index));
@@ -954,6 +956,10 @@ protected final IndexingStrategy planIndexingAsPrimary(Index index) throws IOExc
                     generateSeqNoForOperation(index),
                     index.versionType().updateVersion(currentVersion, index.version())
                 );
+                final boolean toAppend = plan.indexIntoLucene && plan.useLuceneUpdateDocument == false;
+                if (toAppend == false) {
+                    advanceMaxSeqNoOfUpdatesOrDeletes(plan.seqNoForIndexing);
+                }
             }
         }
         return plan;
@@ -1245,6 +1251,7 @@ protected boolean assertNonPrimaryOrigin(final Operation operation) {
 
     protected final DeletionStrategy planDeletionAsPrimary(Delete delete) throws IOException {
         assert delete.origin() == Operation.Origin.PRIMARY : "planing as primary but got " + delete.origin();
+        assert getMaxSeqNoOfUpdatesOrDeletes() != SequenceNumbers.UNASSIGNED_SEQ_NO : "max_seq_no_of_updates is not initialized";
         // resolve operation from external to internal
         final VersionValue versionValue = resolveDocVersion(delete);
         assert incrementVersionLookup();
@@ -1266,6 +1273,7 @@ protected final DeletionStrategy planDeletionAsPrimary(Delete delete) throws IOE
                     currentlyDeleted,
                     generateSeqNoForOperation(delete),
                     delete.versionType().updateVersion(currentVersion, delete.version()));
+            advanceMaxSeqNoOfUpdatesOrDeletes(plan.seqNoOfDeletion);
         }
         return plan;
     }

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

@@ -516,6 +516,7 @@ public void updateShardState(final ShardRouting newRouting,
                                  */
                                 engine.rollTranslogGeneration();
                                 engine.fillSeqNoGaps(newPrimaryTerm);
+                                engine.advanceMaxSeqNoOfUpdatesOrDeletes(seqNoStats().getMaxSeqNo());
                                 replicationTracker.updateLocalCheckpoint(currentRouting.allocationId().getId(), getLocalCheckpoint());
                                 primaryReplicaSyncer.accept(this, new ActionListener<ResyncTask>() {
                                     @Override
@@ -1324,6 +1325,7 @@ public void openEngineAndRecoverFromTranslog() throws IOException {
         };
         innerOpenEngineAndTranslog();
         getEngine().recoverFromTranslog(translogRecoveryRunner, Long.MAX_VALUE);
+        advanceMaxSeqNoOfUpdatesOrDeletes(seqNoStats().getMaxSeqNo());
     }
 
     /**
@@ -1947,6 +1949,7 @@ public void activateWithPrimaryContext(final ReplicationTracker.PrimaryContext p
             getLocalCheckpoint() == primaryContext.getCheckpointStates().get(routingEntry().allocationId().getId()).getLocalCheckpoint();
         synchronized (mutex) {
             replicationTracker.activateWithPrimaryContext(primaryContext); // make changes to primaryMode flag only under mutex
+            advanceMaxSeqNoOfUpdatesOrDeletes(seqNoStats().getMaxSeqNo());
         }
     }
 
@@ -2720,4 +2723,27 @@ void resetEngineToGlobalCheckpoint() throws IOException {
             });
         newEngine.recoverFromTranslog(translogRunner, globalCheckpoint);
     }
+
+    /**
+     * Returns the maximum sequence number of either update operations (overwrite existing documents) or delete operations
+     * have been processed in this shard or the sequence number from {@link #advanceMaxSeqNoOfUpdatesOrDeletes(long)}.
+     * <p>
+     * The primary captures this value after executes a replication request, then transfers it to a replica before executing
+     * that replication request on a replica.
+     */
+    public long getMaxSeqNoOfUpdatesOrDeletes() {
+        return getEngine().getMaxSeqNoOfUpdatesOrDeletes();
+    }
+
+    /**
+     * Advances the max_seq_no_of_updates marker of the engine of this shard to at least the given sequence number.
+     * <p>
+     * A primary calls this method only once to initialize this maker after being promoted or when it finishes its
+     * recovery or relocation. Whereas a replica calls this method before executing a replication request or before
+     * applying translog operations in peer-recovery.
+     */
+    public void advanceMaxSeqNoOfUpdatesOrDeletes(long seqNo) {
+        getEngine().advanceMaxSeqNoOfUpdatesOrDeletes(seqNo);
+        assert seqNo <= getMaxSeqNoOfUpdatesOrDeletes();
+    }
 }

server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -679,6 +679,7 @@ public void testFlushIsDisabledDuringTranslogRecovery() throws IOException {
         expectThrows(IllegalStateException.class, () -> engine.flush(true, true));
         assertTrue(engine.isRecovering());
         engine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+        engine.advanceMaxSeqNoOfUpdatesOrDeletes(engine.getLocalCheckpointTracker().getMaxSeqNo());
         assertFalse(engine.isRecovering());
         doc = testParsedDocument("2", null, testDocumentWithTextField(), SOURCE, null);
         engine.index(indexForDoc(doc));
@@ -2678,6 +2679,7 @@ protected void commitIndexWriter(IndexWriter writer, Translog translog, String s
                 }
             }) {
                 engine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+                engine.advanceMaxSeqNoOfUpdatesOrDeletes(engine.getLocalCheckpointTracker().getMaxSeqNo());
                 final ParsedDocument doc1 = testParsedDocument("1", null, testDocumentWithTextField(), SOURCE, null);
                 engine.index(indexForDoc(doc1));
                 globalCheckpoint.set(engine.getLocalCheckpoint());
@@ -3461,9 +3463,11 @@ public void testEngineMaxTimestampIsInitialized() throws IOException {
             engine.index(appendOnlyPrimary(doc, true, timestamp1));
             assertEquals(timestamp1, engine.segmentsStats(false).getMaxUnsafeAutoIdTimestamp());
         }
-        try (Store store = createStore(newFSDirectory(storeDir)); Engine engine = new InternalEngine(configSupplier.apply(store))) {
+        try (Store store = createStore(newFSDirectory(storeDir));
+             InternalEngine engine = new InternalEngine(configSupplier.apply(store))) {
             assertEquals(IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP, engine.segmentsStats(false).getMaxUnsafeAutoIdTimestamp());
             engine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+            engine.advanceMaxSeqNoOfUpdatesOrDeletes(engine.getLocalCheckpointTracker().getMaxSeqNo());
             assertEquals(timestamp1, engine.segmentsStats(false).getMaxUnsafeAutoIdTimestamp());
             final ParsedDocument doc = testParsedDocument("1", null, testDocumentWithTextField(),
                 new BytesArray("{}".getBytes(Charset.defaultCharset())), null);
@@ -4353,7 +4357,7 @@ public void testKeepTranslogAfterGlobalCheckpoint() throws Exception {
 
         final EngineConfig engineConfig = config(indexSettings, store, translogPath, NoMergePolicy.INSTANCE, null, null,
             () -> globalCheckpoint.get());
-        try (Engine engine = new InternalEngine(engineConfig) {
+        try (InternalEngine engine = new InternalEngine(engineConfig) {
                 @Override
                 protected void commitIndexWriter(IndexWriter writer, Translog translog, String syncId) throws IOException {
                     // Advance the global checkpoint during the flush to create a lag between a persisted global checkpoint in the translog
@@ -4365,6 +4369,7 @@ protected void commitIndexWriter(IndexWriter writer, Translog translog, String s
                 }
             }) {
             engine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+            engine.advanceMaxSeqNoOfUpdatesOrDeletes(engine.getLocalCheckpointTracker().getMaxSeqNo());
             int numDocs = scaledRandomIntBetween(10, 100);
             for (int docId = 0; docId < numDocs; docId++) {
                 ParseContext.Document document = testDocumentWithTextField();
@@ -5033,6 +5038,47 @@ public void testAcquireSearcherOnClosingEngine() throws Exception {
         expectThrows(AlreadyClosedException.class, () -> engine.acquireSearcher("test"));
     }
 
+    public void testTrackMaxSeqNoOfUpdatesOrDeletes() throws Exception {
+        engine.close();
+        Set<String> liveDocIds = new HashSet<>();
+        engine = new InternalEngine(engine.config());
+        assertThat(engine.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(-2L));
+        engine.advanceMaxSeqNoOfUpdatesOrDeletes(randomLongBetween(-1L, 50L));
+        int numOps = between(1, 500);
+        for (int i = 0; i < numOps; i++) {
+            long currentMaxSeqNoOfUpdates = engine.getMaxSeqNoOfUpdatesOrDeletes();
+            ParsedDocument doc = createParsedDoc(Integer.toString(between(1, 100)), null);
+            if (randomBoolean()) {
+                if (randomBoolean()) {
+                    Engine.IndexResult result = engine.index(indexForDoc(doc));
+                    if (liveDocIds.add(doc.id()) == false) {
+                        assertThat("update operations on primary must advance max_seq_no_of_updates",
+                            engine.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(Math.max(currentMaxSeqNoOfUpdates, result.getSeqNo())));
+                    } else {
+                        assertThat(engine.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(currentMaxSeqNoOfUpdates));
+                    }
+                } else {
+                    Engine.DeleteResult result = engine.delete(new Engine.Delete(doc.type(), doc.id(), newUid(doc.id()), primaryTerm.get()));
+                    liveDocIds.remove(doc.id());
+                    assertThat("delete operations on primary must advance max_seq_no_of_updates",
+                        engine.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(Math.max(currentMaxSeqNoOfUpdates, result.getSeqNo())));
+                }
+            } else {
+                long maxSeqNo = engine.getLocalCheckpointTracker().getMaxSeqNo();
+                long seqNo = randomLongBetween(maxSeqNo + 1, maxSeqNo + 10);
+                if (randomBoolean()) {
+                    engine.index(replicaIndexForDoc(doc, 1, seqNo, randomBoolean()));
+                    liveDocIds.add(doc.id());
+                } else {
+                    engine.delete(replicaDeleteForDoc(doc.id(), 1, seqNo, threadPool.relativeTimeInMillis()));
+                    liveDocIds.remove(doc.id());
+                }
+                assertThat("non-primary operations should not advance max_seq_no_of_updates",
+                    engine.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(currentMaxSeqNoOfUpdates));
+            }
+        }
+    }
+
     static void trimUnsafeCommits(EngineConfig config) throws IOException {
         final Store store = config.getStore();
         final TranslogConfig translogConfig = config.getTranslogConfig();

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -947,6 +947,7 @@ public void onFailure(Exception e) {
         resyncLatch.await();
         assertThat(indexShard.getLocalCheckpoint(), equalTo(maxSeqNo));
         assertThat(indexShard.seqNoStats().getMaxSeqNo(), equalTo(maxSeqNo));
+        assertThat(indexShard.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(maxSeqNo));
         closeShard(indexShard, false);
     }
 

server/src/test/java/org/elasticsearch/index/shard/RefreshListenersTests.java

@@ -128,6 +128,7 @@ indexSettings, null, store, newMergePolicy(), iwc.getAnalyzer(), iwc.getSimilari
             EngineTestCase.tombstoneDocSupplier());
         engine = new InternalEngine(config);
         engine.recoverFromTranslog((e, s) -> 0, Long.MAX_VALUE);
+        engine.advanceMaxSeqNoOfUpdatesOrDeletes(engine.getSeqNoStats(-1).getMaxSeqNo());
         listeners.setCurrentRefreshLocationSupplier(engine::getTranslogLastWriteLocation);
     }
 

server/src/test/java/org/elasticsearch/indices/flush/FlushIT.java

@@ -239,7 +239,8 @@ public void testUnallocatedShardsDoesNotHang() throws InterruptedException {
 
     private void indexDoc(Engine engine, String id) throws IOException {
         final ParsedDocument doc = InternalEngineTests.createParsedDoc(id, null);
-        final Engine.IndexResult indexResult = engine.index(new Engine.Index(new Term("_id", Uid.encodeId(doc.id())), 1L, doc));
+        final Engine.IndexResult indexResult = engine.index(new Engine.Index(new Term("_id", Uid.encodeId(doc.id())), doc,
+            engine.getLocalCheckpoint() + 1, 1L, 1L, null, Engine.Operation.Origin.REPLICA, randomLong(), -1L, false));
         assertThat(indexResult.getFailure(), nullValue());
     }
 

test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java

@@ -478,6 +478,8 @@ private InternalEngine createEngine(@Nullable IndexWriterFactory indexWriterFact
         }
         InternalEngine internalEngine = createInternalEngine(indexWriterFactory, localCheckpointTrackerSupplier, seqNoForOperation, config);
         internalEngine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+        // IndexShard initializes this value after replaying local translog.
+        internalEngine.advanceMaxSeqNoOfUpdatesOrDeletes(internalEngine.getLocalCheckpointTracker().getMaxSeqNo());
         return internalEngine;
     }
 

test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java

@@ -94,6 +94,7 @@
 
 import static org.elasticsearch.cluster.routing.TestShardRouting.newShardRouting;
 import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
 /**
@@ -444,6 +445,7 @@ protected IndexShard newStartedShard(CheckedFunction<Boolean, IndexShard, IOExce
         IndexShard shard = shardFunction.apply(primary);
         if (primary) {
             recoverShardFromStore(shard);
+            assertThat(shard.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(shard.seqNoStats().getMaxSeqNo()));
         } else {
             recoveryEmptyReplica(shard, true);
         }