Remove gap skipping when opening engine

Today when opening the engine we skip gaps in the history, advancing the
local checkpoint until it is equal to the maximum sequence number
contained in the commit. This allows history to advance, but it leaves
gaps. A previous change filled these gaps when recovering from store,
but since we were skipping the gaps while opening the engine, this
change had no effect. This commit removes the gap skipping when opening
the engine allowing the gap filling to do its job.

Relates #24535

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

@@ -1421,7 +1421,7 @@ public abstract class Engine implements Closeable {
      * @param primaryTerm the shards primary term this engine was created for
      * @return the number of no-ops added
      */
-    public abstract int fillSequenceNumberHistory(long primaryTerm) throws IOException;
+    public abstract int fillSeqNoGaps(long primaryTerm) throws IOException;
 
     /**
      * Performs recovery from the transaction log.

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

@@ -177,15 +177,6 @@ public class InternalEngine extends Engine {
                 logger.trace("recovered [{}]", seqNoStats);
                 seqNoService = sequenceNumberService(shardId, engineConfig.getIndexSettings(), seqNoStats);
                 updateMaxUnsafeAutoIdTimestampFromWriter(writer);
-                // norelease
-                /*
-                 * We have no guarantees that all operations above the local checkpoint are in the Lucene commit or the translog. This means
-                 * that we there might be operations greater than the local checkpoint that will not be replayed. Here we force the local
-                 * checkpoint to the maximum sequence number in the commit (at the potential expense of correctness).
-                 */
-                while (seqNoService().getLocalCheckpoint() < seqNoService().getMaxSeqNo()) {
-                    seqNoService().markSeqNoAsCompleted(seqNoService().getLocalCheckpoint() + 1);
-                }
                 indexWriter = writer;
                 translog = openTranslog(engineConfig, writer, () -> seqNoService().getGlobalCheckpoint());
                 assert translog.getGeneration() != null;
@@ -226,21 +217,20 @@ public class InternalEngine extends Engine {
     }
 
     @Override
-    public int fillSequenceNumberHistory(long primaryTerm) throws IOException {
+    public int fillSeqNoGaps(long primaryTerm) throws IOException {
-        try (ReleasableLock lock = writeLock.acquire()) {
+        try (ReleasableLock ignored = writeLock.acquire()) {
             ensureOpen();
-            final long localCheckpoint = seqNoService.getLocalCheckpoint();
+            final long localCheckpoint = seqNoService().getLocalCheckpoint();
-            final long maxSeqId = seqNoService.getMaxSeqNo();
+            final long maxSeqNo = seqNoService().getMaxSeqNo();
             int numNoOpsAdded = 0;
-            for (long seqNo = localCheckpoint + 1; seqNo <= maxSeqId;
+            for (
-                 // the local checkpoint might have been advanced so we are leap-frogging
+                    long seqNo = localCheckpoint + 1;
-                 // to the next seq ID we need to process and create a noop for
+                    seqNo <= maxSeqNo;
-                 seqNo = seqNoService.getLocalCheckpoint()+1) {
+                    seqNo = seqNoService().getLocalCheckpoint() + 1 /* the local checkpoint might have advanced so we leap-frog */) {
-                final NoOp noOp = new NoOp(seqNo, primaryTerm, Operation.Origin.PRIMARY, System.nanoTime(), "filling up seqNo history");
+                innerNoOp(new NoOp(seqNo, primaryTerm, Operation.Origin.PRIMARY, System.nanoTime(), "filling gaps"));
-                innerNoOp(noOp);
                 numNoOpsAdded++;
-                assert seqNo <= seqNoService.getLocalCheckpoint() : "localCheckpoint didn't advanced used to be " + seqNo + " now it's on:"
+                assert seqNo <= seqNoService().getLocalCheckpoint()
-                     + seqNoService.getLocalCheckpoint();
+                        : "local checkpoint did not advance; was [" + seqNo + "], now [" + seqNoService().getLocalCheckpoint() + "]";
 
             }
             return numNoOpsAdded;

core/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -365,7 +365,7 @@ final class StoreRecovery {
                 }
                 indexShard.performTranslogRecovery(indexShouldExists);
                 assert indexShard.shardRouting.primary() : "only primary shards can recover from store";
-                indexShard.getEngine().fillSequenceNumberHistory(indexShard.getPrimaryTerm());
+                indexShard.getEngine().fillSeqNoGaps(indexShard.getPrimaryTerm());
             }
             indexShard.finalizeRecovery();
             indexShard.postRecovery("post recovery from shard_store");

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

@@ -114,7 +114,6 @@ import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.mapper.RootObjectMapper;
 import org.elasticsearch.index.mapper.SeqNoFieldMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
-import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.mapper.Uid;
 import org.elasticsearch.index.mapper.UidFieldMapper;
 import org.elasticsearch.index.seqno.SequenceNumbers;
@@ -3580,6 +3579,8 @@ public class InternalEngineTests extends ESTestCase {
 
         try (Engine recoveringEngine =
                  new InternalEngine(copy(initialEngine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG))) {
+            recoveringEngine.recoverFromTranslog();
+            recoveringEngine.fillSeqNoGaps(2);
             assertThat(recoveringEngine.seqNoService().getLocalCheckpoint(), greaterThanOrEqualTo((long) (docs - 1)));
         }
     }
@@ -3618,6 +3619,8 @@ public class InternalEngineTests extends ESTestCase {
 
         try (Engine recoveringEngine =
                  new InternalEngine(copy(initialEngine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG))) {
+            recoveringEngine.recoverFromTranslog();
+            recoveringEngine.fillSeqNoGaps(1);
             assertThat(recoveringEngine.seqNoService().getLocalCheckpoint(), greaterThanOrEqualTo((long) (3 * (docs - 1) + 2 - 1)));
         }
     }
@@ -3719,8 +3722,15 @@ public class InternalEngineTests extends ESTestCase {
                         throw new UnsupportedOperationException();
                     }
                 };
-            noOpEngine = createEngine(defaultSettings, store, primaryTranslogDir, newMergePolicy(), null, () -> seqNoService);
+            noOpEngine = new InternalEngine(copy(engine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG)) {
+                @Override
+                public SequenceNumbersService seqNoService() {
+                    return seqNoService;
+                }
+            };
+            noOpEngine.recoverFromTranslog();
             final long primaryTerm = randomNonNegativeLong();
+            final int gapsFilled = noOpEngine.fillSeqNoGaps(primaryTerm);
             final String reason = randomAlphaOfLength(16);
             noOpEngine.noOp(
                     new Engine.NoOp(
@@ -3730,10 +3740,17 @@ public class InternalEngineTests extends ESTestCase {
                             System.nanoTime(),
                             reason));
             assertThat(noOpEngine.seqNoService().getLocalCheckpoint(), equalTo((long) (maxSeqNo + 1)));
-            assertThat(noOpEngine.getTranslog().totalOperations(), equalTo(1));
+            assertThat(noOpEngine.getTranslog().totalOperations(), equalTo(1 + gapsFilled));
-            final Translog.Operation op = noOpEngine.getTranslog().newSnapshot().next();
+            // skip to the op that we added to the translog
-            assertThat(op, instanceOf(Translog.NoOp.class));
+            Translog.Operation op;
-            final Translog.NoOp noOp = (Translog.NoOp) op;
+            Translog.Operation last = null;
+            final Translog.Snapshot snapshot = noOpEngine.getTranslog().newSnapshot();
+            while ((op = snapshot.next()) != null) {
+                last = op;
+            }
+            assertNotNull(last);
+            assertThat(last, instanceOf(Translog.NoOp.class));
+            final Translog.NoOp noOp = (Translog.NoOp) last;
             assertThat(noOp.seqNo(), equalTo((long) (maxSeqNo + 1)));
             assertThat(noOp.primaryTerm(), equalTo(primaryTerm));
             assertThat(noOp.reason(), equalTo(reason));
@@ -3873,7 +3890,7 @@ public class InternalEngineTests extends ESTestCase {
             recoveringEngine.recoverFromTranslog();
             assertEquals(maxSeqIDOnReplica, recoveringEngine.seqNoService().getMaxSeqNo());
             assertEquals(checkpointOnReplica, recoveringEngine.seqNoService().getLocalCheckpoint());
-            assertEquals((maxSeqIDOnReplica+1) - numDocsOnReplica, recoveringEngine.fillSequenceNumberHistory(2));
+            assertEquals((maxSeqIDOnReplica + 1) - numDocsOnReplica, recoveringEngine.fillSeqNoGaps(2));
 
             // now snapshot the tlog and ensure the primary term is updated
             Translog.Snapshot snapshot = recoveringEngine.getTranslog().newSnapshot();
@@ -3905,7 +3922,7 @@ public class InternalEngineTests extends ESTestCase {
             recoveringEngine.recoverFromTranslog();
             assertEquals(maxSeqIDOnReplica, recoveringEngine.seqNoService().getMaxSeqNo());
             assertEquals(maxSeqIDOnReplica, recoveringEngine.seqNoService().getLocalCheckpoint());
-            assertEquals(0, recoveringEngine.fillSequenceNumberHistory(3));
+            assertEquals(0, recoveringEngine.fillSeqNoGaps(3));
             assertEquals(maxSeqIDOnReplica, recoveringEngine.seqNoService().getMaxSeqNo());
             assertEquals(maxSeqIDOnReplica, recoveringEngine.seqNoService().getLocalCheckpoint());
         } finally {