Replica starts peer recovery with safe commit

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

@@ -1512,6 +1512,11 @@ public interface Warmer {
      */
     public abstract Engine recoverFromTranslog() throws IOException;
 
+    /**
+     * Do not replay translog operations, but make the engine be ready.
+     */
+    public abstract void skipTranslogRecovery();
+
     /**
      * Returns <code>true</code> iff this engine is currently recovering from translog.
      */

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

@@ -401,6 +401,15 @@ public InternalEngine recoverFromTranslog() throws IOException {
         return this;
     }
 
+    @Override
+    public void skipTranslogRecovery() {
+        if (openMode != EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG) {
+            throw new IllegalStateException("Can't skip translog recovery with open mode: " + openMode);
+        }
+        assert pendingTranslogRecovery.get() : "translogRecovery is not pending but should be";
+        pendingTranslogRecovery.set(false); // we are good - now we can commit
+    }
+
     private IndexCommit getStartingCommitPoint() throws IOException {
         if (openMode == EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG) {
             final long lastSyncedGlobalCheckpoint = translog.getLastSyncedGlobalCheckpoint();

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

@@ -1304,9 +1304,20 @@ public void openIndexAndCreateTranslog(boolean forceNewHistoryUUID, long globalC
      * opens the engine on top of the existing lucene engine and translog.
      * Operations from the translog will be replayed to bring lucene up to date.
      **/
-    public void openIndexAndTranslog() throws IOException {
+    public void openIndexAndRecoveryFromTranslog() throws IOException {
         assert recoveryState.getRecoverySource().getType() == RecoverySource.Type.EXISTING_STORE;
         innerOpenEngineAndTranslog(EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG, false);
+        getEngine().recoverFromTranslog();
+    }
+
+    /**
+     * Opens the engine on top of the existing lucene engine and translog.
+     * The translog is kept but its operations won't be replayed.
+     */
+    public void openIndexAndSkipTranslogRecovery() throws IOException {
+        assert recoveryState.getRecoverySource().getType() == RecoverySource.Type.PEER;
+        innerOpenEngineAndTranslog(EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG, false);
+        getEngine().skipTranslogRecovery();
     }
 
     private void innerOpenEngineAndTranslog(final EngineConfig.OpenMode openMode, final boolean forceNewHistoryUUID) throws IOException {
@@ -1339,13 +1350,12 @@ private void innerOpenEngineAndTranslog(final EngineConfig.OpenMode openMode, fi
             globalCheckpointTracker.updateGlobalCheckpointOnReplica(Translog.readGlobalCheckpoint(translogConfig.getTranslogPath()),
                 "read from translog checkpoint");
         }
-        Engine newEngine = createNewEngine(config);
+        createNewEngine(config);
         verifyNotClosed();
         if (openMode == EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG) {
             // We set active because we are now writing operations to the engine; this way, if we go idle after some time and become inactive,
             // we still give sync'd flush a chance to run:
             active.set(true);
-            newEngine.recoverFromTranslog();
         }
         assertSequenceNumbersInCommit();
         assert recoveryState.getStage() == RecoveryState.Stage.TRANSLOG : "TRANSLOG stage expected but was: " + recoveryState.getStage();

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

@@ -401,7 +401,7 @@ private void internalRecoverFromStore(IndexShard indexShard) throws IndexShardRe
                     logger.debug("failed to list file details", e);
                 }
                 if (indexShouldExists) {
-                    indexShard.openIndexAndTranslog();
+                    indexShard.openIndexAndRecoveryFromTranslog();
                     indexShard.getEngine().fillSeqNoGaps(indexShard.getPrimaryTerm());
                 } else {
                     indexShard.createIndexAndTranslog();

core/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java

@@ -21,6 +21,8 @@
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.RateLimiter;
 import org.elasticsearch.ElasticsearchException;
@@ -39,6 +41,7 @@
 import org.elasticsearch.common.util.CancellableThreads;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.engine.CombinedDeletionPolicy;
 import org.elasticsearch.index.engine.RecoveryEngineException;
 import org.elasticsearch.index.mapper.MapperException;
 import org.elasticsearch.index.seqno.SequenceNumbers;
@@ -60,6 +63,7 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -108,8 +112,8 @@ public PeerRecoveryTargetService(Settings settings, ThreadPool threadPool, Trans
                 FileChunkTransportRequestHandler());
         transportService.registerRequestHandler(Actions.CLEAN_FILES, RecoveryCleanFilesRequest::new, ThreadPool.Names.GENERIC, new
                 CleanFilesRequestHandler());
-        transportService.registerRequestHandler(Actions.PREPARE_TRANSLOG, RecoveryPrepareForTranslogOperationsRequest::new, ThreadPool
-                .Names.GENERIC, new PrepareForTranslogOperationsRequestHandler());
+        transportService.registerRequestHandler(Actions.PREPARE_TRANSLOG, ThreadPool.Names.GENERIC,
+                RecoveryPrepareForTranslogOperationsRequest::new, new PrepareForTranslogOperationsRequestHandler());
         transportService.registerRequestHandler(Actions.TRANSLOG_OPS, RecoveryTranslogOperationsRequest::new, ThreadPool.Names.GENERIC,
                 new TranslogOperationsRequestHandler());
         transportService.registerRequestHandler(Actions.FINALIZE, RecoveryFinalizeRecoveryRequest::new, ThreadPool.Names.GENERIC, new
@@ -353,7 +357,9 @@ private StartRecoveryRequest getStartRecoveryRequest(final RecoveryTarget recove
     public static long getStartingSeqNo(final RecoveryTarget recoveryTarget) {
         try {
             final long globalCheckpoint = Translog.readGlobalCheckpoint(recoveryTarget.translogLocation());
-            final SequenceNumbers.CommitInfo seqNoStats = recoveryTarget.store().loadSeqNoInfo(null);
+            final List<IndexCommit> existingCommits = DirectoryReader.listCommits(recoveryTarget.store().directory());
+            final IndexCommit safeCommit = CombinedDeletionPolicy.findSafeCommitPoint(existingCommits, globalCheckpoint);
+            final SequenceNumbers.CommitInfo seqNoStats = recoveryTarget.store().loadSeqNoInfo(safeCommit);
             if (seqNoStats.maxSeqNo <= globalCheckpoint) {
                 assert seqNoStats.localCheckpoint <= globalCheckpoint;
                 /*
@@ -387,7 +393,7 @@ class PrepareForTranslogOperationsRequestHandler implements TransportRequestHand
         public void messageReceived(RecoveryPrepareForTranslogOperationsRequest request, TransportChannel channel) throws Exception {
             try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId()
             )) {
-                recoveryRef.target().prepareForTranslogOperations(request.totalTranslogOps());
+                recoveryRef.target().prepareForTranslogOperations(request.deleteLocalTranslog(), request.totalTranslogOps());
             }
             channel.sendResponse(TransportResponse.Empty.INSTANCE);
         }

core/src/main/java/org/elasticsearch/indices/recovery/RecoveryPrepareForTranslogOperationsRequest.java

@@ -28,19 +28,33 @@
 
 import java.io.IOException;
 
-public class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
+class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
 
-    private long recoveryId;
-    private ShardId shardId;
-    private int totalTranslogOps = RecoveryState.Translog.UNKNOWN;
+    private final long recoveryId;
+    private final ShardId shardId;
+    private final int totalTranslogOps;
+    private final boolean deleteLocalTranslog;
 
-    public RecoveryPrepareForTranslogOperationsRequest() {
-    }
-
-    RecoveryPrepareForTranslogOperationsRequest(long recoveryId, ShardId shardId, int totalTranslogOps) {
+    RecoveryPrepareForTranslogOperationsRequest(long recoveryId, ShardId shardId, int totalTranslogOps, boolean deleteLocalTranslog) {
         this.recoveryId = recoveryId;
         this.shardId = shardId;
         this.totalTranslogOps = totalTranslogOps;
+        this.deleteLocalTranslog = deleteLocalTranslog;
+    }
+
+    RecoveryPrepareForTranslogOperationsRequest(StreamInput in) throws IOException {
+        super.readFrom(in);
+        recoveryId = in.readLong();
+        shardId = ShardId.readShardId(in);
+        totalTranslogOps = in.readVInt();
+        if (in.getVersion().before(Version.V_6_0_0_alpha1)) {
+            in.readLong(); // maxUnsafeAutoIdTimestamp
+        }
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            deleteLocalTranslog = in.readBoolean();
+        } else {
+            deleteLocalTranslog = true;
+        }
     }
 
     public long recoveryId() {
@@ -55,15 +69,11 @@ public int totalTranslogOps() {
         return totalTranslogOps;
     }
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        recoveryId = in.readLong();
-        shardId = ShardId.readShardId(in);
-        totalTranslogOps = in.readVInt();
-        if (in.getVersion().before(Version.V_6_0_0_alpha1)) {
-            in.readLong(); // maxUnsafeAutoIdTimestamp
-        }
+    /**
+     * Whether or not the recover target should delete its local translog
+     */
+    boolean deleteLocalTranslog() {
+        return deleteLocalTranslog;
     }
 
     @Override
@@ -75,5 +85,8 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().before(Version.V_6_0_0_alpha1)) {
             out.writeLong(IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP); // maxUnsafeAutoIdTimestamp
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            out.writeBoolean(deleteLocalTranslog);
+        }
     }
 }

core/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -188,7 +188,9 @@ public RecoveryResponse recoverToTarget() throws IOException {
             runUnderPrimaryPermit(() -> shard.initiateTracking(request.targetAllocationId()));
 
             try {
-                prepareTargetForTranslog(translog.estimateTotalOperationsFromMinSeq(startingSeqNo));
+                // For a sequence based recovery, the target can keep its local translog
+                prepareTargetForTranslog(isSequenceNumberBasedRecoveryPossible == false,
+                    translog.estimateTotalOperationsFromMinSeq(startingSeqNo));
             } catch (final Exception e) {
                 throw new RecoveryEngineException(shard.shardId(), 1, "prepare target for translog failed", e);
             }
@@ -421,13 +423,13 @@ public void phase1(final IndexCommit snapshot, final Supplier<Integer> translogO
         }
     }
 
-    void prepareTargetForTranslog(final int totalTranslogOps) throws IOException {
+    void prepareTargetForTranslog(final boolean deleteLocalTranslog, final int totalTranslogOps) throws IOException {
         StopWatch stopWatch = new StopWatch().start();
         logger.trace("recovery [phase1]: prepare remote engine for translog");
         final long startEngineStart = stopWatch.totalTime().millis();
         // Send a request preparing the new shard's translog to receive operations. This ensures the shard engine is started and disables
         // garbage collection (not the JVM's GC!) of tombstone deletes.
-        cancellableThreads.executeIO(() -> recoveryTarget.prepareForTranslogOperations(totalTranslogOps));
+        cancellableThreads.executeIO(() -> recoveryTarget.prepareForTranslogOperations(deleteLocalTranslog, totalTranslogOps));
         stopWatch.stop();
 
         response.startTime = stopWatch.totalTime().millis() - startEngineStart;

core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java

@@ -362,10 +362,13 @@ private void ensureRefCount() {
     /*** Implementation of {@link RecoveryTargetHandler } */
 
     @Override
-    public void prepareForTranslogOperations(int totalTranslogOps) throws IOException {
+    public void prepareForTranslogOperations(boolean deleteLocalTranslog, int totalTranslogOps) throws IOException {
         state().getTranslog().totalOperations(totalTranslogOps);
-        // TODO: take the local checkpoint from store as global checkpoint, once we know it's safe
-        indexShard().openIndexAndCreateTranslog(false, SequenceNumbers.UNASSIGNED_SEQ_NO);
+        if (deleteLocalTranslog) {
+            indexShard().openIndexAndCreateTranslog(false, SequenceNumbers.UNASSIGNED_SEQ_NO);
+        } else {
+            indexShard().openIndexAndSkipTranslogRecovery();
+        }
     }
 
     @Override

core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTargetHandler.java

@@ -33,9 +33,10 @@ public interface RecoveryTargetHandler {
     /**
      * Prepares the target to receive translog operations, after all file have been copied
      *
-     * @param totalTranslogOps total translog operations expected to be sent
+     * @param deleteLocalTranslog whether or not to delete the local translog on the target
+     * @param totalTranslogOps    total translog operations expected to be sent
      */
-    void prepareForTranslogOperations(int totalTranslogOps) throws IOException;
+    void prepareForTranslogOperations(boolean deleteLocalTranslog, int totalTranslogOps) throws IOException;
 
     /**
      * The finalize request refreshes the engine now that new segments are available, enables garbage collection of tombstone files, and

core/src/main/java/org/elasticsearch/indices/recovery/RemoteRecoveryTargetHandler.java

@@ -76,9 +76,9 @@ public RemoteRecoveryTargetHandler(long recoveryId, ShardId shardId, TransportSe
     }
 
     @Override
-    public void prepareForTranslogOperations(int totalTranslogOps) throws IOException {
+    public void prepareForTranslogOperations(boolean deleteLocalTranslog, int totalTranslogOps) throws IOException {
         transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.PREPARE_TRANSLOG,
-                new RecoveryPrepareForTranslogOperationsRequest(recoveryId, shardId, totalTranslogOps),
+                new RecoveryPrepareForTranslogOperationsRequest(recoveryId, shardId, totalTranslogOps, deleteLocalTranslog),
                 TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
                 EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
     }

core/src/test/java/org/elasticsearch/index/replication/RecoveryDuringReplicationTests.java

@@ -31,7 +31,9 @@
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.lucene.uid.Versions;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.EngineConfig;
@@ -226,7 +228,6 @@ public void testRecoveryAfterPrimaryPromotion() throws Exception {
             final IndexShard oldPrimary = shards.getPrimary();
             final IndexShard newPrimary = shards.getReplicas().get(0);
             final IndexShard replica = shards.getReplicas().get(1);
-            boolean expectSeqNoRecovery = true;
             if (randomBoolean()) {
                 // simulate docs that were inflight when primary failed, these will be rolled back
                 final int rollbackDocs = randomIntBetween(1, 5);
@@ -239,7 +240,6 @@ public void testRecoveryAfterPrimaryPromotion() throws Exception {
                 }
                 if (randomBoolean()) {
                     oldPrimary.flush(new FlushRequest(index.getName()));
-                    expectSeqNoRecovery = false;
                 }
             }
 
@@ -252,9 +252,29 @@ public void testRecoveryAfterPrimaryPromotion() throws Exception {
                 equalTo(totalDocs - 1L));
 
             // index some more
-            totalDocs += shards.indexDocs(randomIntBetween(0, 5));
+            int moreDocs = shards.indexDocs(randomIntBetween(0, 5));
+            totalDocs += moreDocs;
+
+            // As a replica keeps a safe commit, the file-based recovery only happens if the required translog
+            // for the sequence based recovery are not fully retained and extra documents were added to the primary.
+            boolean expectSeqNoRecovery = (moreDocs == 0 || frequently());
+            int uncommittedOpsOnPrimary = 0;
+            if (expectSeqNoRecovery == false) {
+                IndexMetaData.Builder builder = IndexMetaData.builder(newPrimary.indexSettings().getIndexMetaData());
+                builder.settings(Settings.builder().put(newPrimary.indexSettings().getSettings())
+                    .put(IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getKey(), "-1")
+                    .put(IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.getKey(), "-1")
+                );
+                newPrimary.indexSettings().updateIndexMetaData(builder.build());
+                newPrimary.onSettingsChanged();
+                shards.syncGlobalCheckpoint();
+                newPrimary.flush(new FlushRequest());
+                uncommittedOpsOnPrimary = shards.indexDocs(randomIntBetween(0, 10));
+            }
 
             if (randomBoolean()) {
+                uncommittedOpsOnPrimary = 0;
+                shards.syncGlobalCheckpoint();
                 newPrimary.flush(new FlushRequest());
             }
 
@@ -269,7 +289,7 @@ public void testRecoveryAfterPrimaryPromotion() throws Exception {
                 assertThat(newReplica.recoveryState().getTranslog().recoveredOperations(), equalTo(totalDocs - committedDocs));
             } else {
                 assertThat(newReplica.recoveryState().getIndex().fileDetails(), not(empty()));
-                assertThat(newReplica.recoveryState().getTranslog().recoveredOperations(), equalTo(totalDocs));
+                assertThat(newReplica.recoveryState().getTranslog().recoveredOperations(), equalTo(uncommittedOpsOnPrimary));
             }
 
             // roll back the extra ops in the replica

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

@@ -2109,7 +2109,7 @@ public void testShardActiveDuringInternalRecovery() throws IOException {
         shard.prepareForIndexRecovery();
         // Shard is still inactive since we haven't started recovering yet
         assertFalse(shard.isActive());
-        shard.openIndexAndTranslog();
+        shard.openIndexAndRecoveryFromTranslog();
         // Shard should now be active since we did recover:
         assertTrue(shard.isActive());
         closeShards(shard);
@@ -2137,8 +2137,8 @@ public void testShardActiveDuringPeerRecovery() throws IOException {
             new RecoveryTarget(shard, discoveryNode, recoveryListener, aLong -> {
             }) {
                 @Override
-                public void prepareForTranslogOperations(int totalTranslogOps) throws IOException {
-                    super.prepareForTranslogOperations(totalTranslogOps);
+                public void prepareForTranslogOperations(boolean deleteLocalTranslog, int totalTranslogOps) throws IOException {
+                    super.prepareForTranslogOperations(deleteLocalTranslog, totalTranslogOps);
                     // Shard is still inactive since we haven't started recovering yet
                     assertFalse(replica.isActive());
 
@@ -2186,8 +2186,8 @@ public void testRefreshListenersDuringPeerRecovery() throws IOException {
             }) {
             // we're only checking that listeners are called when the engine is open, before there is no point
                 @Override
-                public void prepareForTranslogOperations(int totalTranslogOps) throws IOException {
-                    super.prepareForTranslogOperations(totalTranslogOps);
+                public void prepareForTranslogOperations(boolean deleteLocalTranslog, int totalTranslogOps) throws IOException {
+                    super.prepareForTranslogOperations(deleteLocalTranslog, totalTranslogOps);
                     assertListenerCalled.accept(replica);
                 }