Reset replica engine before primary-replica resync

server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java

@@ -121,16 +121,14 @@ protected WriteReplicaResult shardOperationOnReplica(ResyncReplicationRequest re
     public static Translog.Location performOnReplica(ResyncReplicationRequest request, IndexShard replica) throws Exception {
         Translog.Location location = null;
         for (Translog.Operation operation : request.getOperations()) {
-            final Engine.Result operationResult = replica.applyTranslogOperation(operation, Engine.Operation.Origin.REPLICA);
+            final Engine.Result operationResult = replica.applyResyncOperation(operation);
             if (operationResult.getResultType() == Engine.Result.Type.MAPPING_UPDATE_REQUIRED) {
                 throw new TransportReplicationAction.RetryOnReplicaException(replica.shardId(),
                     "Mappings are not available on the replica yet, triggered update: " + operationResult.getRequiredMappingUpdate());
             }
             location = syncOperationResultOrThrow(operationResult, location);
         }
-        if (request.getTrimAboveSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
-            replica.trimOperationOfPreviousPrimaryTerms(request.getTrimAboveSeqNo());
-        }
+        replica.afterApplyResyncOperationsBulk(request.getTrimAboveSeqNo());
         return location;
     }
 

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

@@ -1623,16 +1623,21 @@ public interface Warmer {
     public abstract int fillSeqNoGaps(long primaryTerm) throws IOException;
 
     /**
-     * Performs recovery from the transaction log.
+     * Performs recovery from the transaction log up to {@code recoverUpToSeqNo} .
      * This operation will close the engine if the recovery fails.
      */
-    public abstract Engine recoverFromTranslog() throws IOException;
+    public abstract Engine recoverFromTranslog(long recoverUpToSeqNo) throws IOException;
 
     /**
      * Do not replay translog operations, but make the engine be ready.
      */
     public abstract void skipTranslogRecovery();
 
+    /**
+     * Returns a {@link ReadOnlyEngine} that points to the last commit of the current engine.
+     */
+    public abstract Engine lockDownEngine() throws IOException;
+
     /**
      * Returns <code>true</code> iff this engine is currently recovering from translog.
      */

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

@@ -364,15 +364,15 @@ private void bootstrapAppendOnlyInfoFromWriter(IndexWriter writer) {
     }
 
     @Override
-    public InternalEngine recoverFromTranslog() throws IOException {
+    public InternalEngine recoverFromTranslog(long recoverUpToSeqNo) throws IOException {
         flushLock.lock();
         try (ReleasableLock lock = readLock.acquire()) {
             ensureOpen();
             if (pendingTranslogRecovery.get() == false) {
                 throw new IllegalStateException("Engine has already been recovered");
             }
             try {
-                recoverFromTranslogInternal();
+                recoverFromTranslogInternal(recoverUpToSeqNo);
             } catch (Exception e) {
                 try {
                     pendingTranslogRecovery.set(true); // just play safe and never allow commits on this see #ensureCanFlush
@@ -388,17 +388,23 @@ public InternalEngine recoverFromTranslog() throws IOException {
         return this;
     }
 
+    // for testing
+    final Engine recoverFromTranslog() throws IOException {
+        return recoverFromTranslog(Long.MAX_VALUE);
+    }
+
     @Override
     public void skipTranslogRecovery() {
         assert pendingTranslogRecovery.get() : "translogRecovery is not pending but should be";
         pendingTranslogRecovery.set(false); // we are good - now we can commit
     }
 
-    private void recoverFromTranslogInternal() throws IOException {
+    private void recoverFromTranslogInternal(long recoverUpToSeqNo) throws IOException {
         Translog.TranslogGeneration translogGeneration = translog.getGeneration();
         final int opsRecovered;
         final long translogGen = Long.parseLong(lastCommittedSegmentInfos.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
-        try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(translogGen)) {
+        try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(
+            new Translog.TranslogGeneration(translog.getTranslogUUID(), translogGen), recoverUpToSeqNo)) {
             opsRecovered = config().getTranslogRecoveryRunner().run(this, snapshot);
         } catch (Exception e) {
             throw new EngineException(shardId, "failed to recover from translog", e);
@@ -2234,6 +2240,22 @@ public SeqNoStats getSeqNoStats(long globalCheckpoint) {
         return localCheckpointTracker.getStats(globalCheckpoint);
     }
 
+    @Override
+    public Engine lockDownEngine() throws IOException {
+        try (ReleasableLock ignored = writeLock.acquire()) {
+            refresh("lockdown", SearcherScope.INTERNAL);
+            Searcher searcher = acquireSearcher("lockdown", SearcherScope.INTERNAL);
+            try {
+                ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(this.engineConfig, lastCommittedSegmentInfos, searcher,
+                    getSeqNoStats(getLastSyncedGlobalCheckpoint()), getTranslogStats());
+                searcher = null;
+                return readOnlyEngine;
+            } finally {
+                IOUtils.close(searcher);
+            }
+        }
+    }
+
     /**
      * Returns the number of times a version was looked up either from the index.
      * Note this is only available if assertions are enabled