Noop peer recoveries on closed index

server/src/main/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseAction.java

@@ -20,8 +20,8 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
@@ -109,8 +109,7 @@ private void executeShardOperation(final ShardRequest request, final IndexShard
         if (clusterBlocks.hasIndexBlock(shardId.getIndexName(), request.clusterBlock()) == false) {
             throw new IllegalStateException("Index shard " + shardId + " must be blocked by " + request.clusterBlock() + " before closing");
         }
-        indexShard.verifyShardBeforeIndexClosing();
-        indexShard.flush(new FlushRequest().force(true).waitIfOngoing(true));
+        indexShard.prepareShardBeforeIndexClosing(request.syncId);
         logger.trace("{} shard is ready for closing", shardId);
     }
 
@@ -135,21 +134,28 @@ public void markShardCopyAsStaleIfNeeded(final ShardId shardId, final String all
     public static class ShardRequest extends ReplicationRequest<ShardRequest> {
 
         private final ClusterBlock clusterBlock;
+        private final String syncId;
 
         ShardRequest(StreamInput in) throws IOException {
             super(in);
             clusterBlock = new ClusterBlock(in);
+            if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+                syncId = in.readString();
+            } else {
+                syncId = "";
+            }
         }
 
-        public ShardRequest(final ShardId shardId, final ClusterBlock clusterBlock, final TaskId parentTaskId) {
+        public ShardRequest(final ShardId shardId, final ClusterBlock clusterBlock, final String syncId, final TaskId parentTaskId) {
             super(shardId);
             this.clusterBlock = Objects.requireNonNull(clusterBlock);
+            this.syncId = Objects.requireNonNull(syncId);
             setParentTask(parentTaskId);
         }
 
         @Override
         public String toString() {
-            return "verify shard " + shardId + " before close with block " + clusterBlock;
+            return "verify shard " + shardId + " before close with block " + clusterBlock + " sync_id " + syncId;
         }
 
         @Override
@@ -161,6 +167,9 @@ public void readFrom(final StreamInput in) {
         public void writeTo(final StreamOutput out) throws IOException {
             super.writeTo(out);
             clusterBlock.writeTo(out);
+            if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+                out.writeString(syncId);
+            }
         }
 
         public ClusterBlock clusterBlock() {

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java

@@ -384,8 +384,9 @@ private void sendVerifyShardBeforeCloseRequest(final IndexShardRoutingTable shar
                 return;
             }
             final TaskId parentTaskId = new TaskId(clusterService.localNode().getId(), request.taskId());
+            final String syncId = UUIDs.randomBase64UUID();
             final TransportVerifyShardBeforeCloseAction.ShardRequest shardRequest =
-                new TransportVerifyShardBeforeCloseAction.ShardRequest(shardId, closingBlock, parentTaskId);
+                new TransportVerifyShardBeforeCloseAction.ShardRequest(shardId, closingBlock, syncId, parentTaskId);
             if (request.ackTimeout() != null) {
                 shardRequest.timeout(request.ackTimeout());
             }

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

@@ -265,18 +265,11 @@ protected final DocsStats docsStats(IndexReader indexReader) {
     }
 
     /**
-     * Performs the pre-closing checks on the {@link Engine}.
+     * Performs the pre-closing action on the {@link Engine}.
      *
      * @throws IllegalStateException if the sanity checks failed
      */
-    public void verifyEngineBeforeIndexClosing() throws IllegalStateException {
-        final long globalCheckpoint = engineConfig.getGlobalCheckpointSupplier().getAsLong();
-        final long maxSeqNo = getSeqNoStats(globalCheckpoint).getMaxSeqNo();
-        if (globalCheckpoint != maxSeqNo) {
-            throw new IllegalStateException("Global checkpoint [" + globalCheckpoint
-                + "] mismatches maximum sequence number [" + maxSeqNo + "] on index shard " + shardId);
-        }
-    }
+    public abstract void prepareEngineBeforeIndexClosing(String syncId) throws IllegalStateException;
 
     /**
      * A throttling class that can be activated, causing the

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

@@ -2756,4 +2756,25 @@ private static void trimUnsafeCommits(EngineConfig engineConfig) throws IOExcept
         final long minRetainedTranslogGen = Translog.readMinTranslogGeneration(translogPath, translogUUID);
         store.trimUnsafeCommits(globalCheckpoint, minRetainedTranslogGen, engineConfig.getIndexSettings().getIndexVersionCreated());
     }
+
+    protected void verifyEngineBeforeIndexClosing() {
+        final long globalCheckpoint = engineConfig.getGlobalCheckpointSupplier().getAsLong();
+        final long maxSeqNo = localCheckpointTracker.getMaxSeqNo();
+        if (globalCheckpoint != maxSeqNo) {
+            throw new IllegalStateException("Global checkpoint [" + globalCheckpoint
+                + "] mismatches maximum sequence number [" + maxSeqNo + "] on index shard " + shardId);
+        }
+    }
+
+    @Override
+    public void prepareEngineBeforeIndexClosing(String syncId) throws IllegalStateException {
+        try (ReleasableLock ignored = writeLock.acquire()) {
+            ensureOpen();
+            verifyEngineBeforeIndexClosing();
+            final CommitId commitId = flush(true, true);
+            if (syncId != null) {
+                syncFlush(syncId, commitId);
+            }
+        }
+    }
 }

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

@@ -146,7 +146,7 @@ protected boolean assertMaxSeqNoEqualsToGlobalCheckpoint(final long maxSeqNo, fi
     }
 
     @Override
-    public void verifyEngineBeforeIndexClosing() throws IllegalStateException {
+    public void prepareEngineBeforeIndexClosing(String syncId) throws IllegalStateException {
         // the value of the global checkpoint is verified when the read-only engine is opened,
         // and it is not expected to change during the lifecycle of the engine. We could also
         // check this value before closing the read-only engine but if something went wrong
@@ -156,7 +156,7 @@ public void verifyEngineBeforeIndexClosing() throws IllegalStateException {
     }
 
     protected final DirectoryReader wrapReader(DirectoryReader reader,
-                                                    Function<DirectoryReader, DirectoryReader> readerWrapperFunction) throws IOException {
+                                               Function<DirectoryReader, DirectoryReader> readerWrapperFunction) throws IOException {
         reader = ElasticsearchDirectoryReader.wrap(reader, engineConfig.getShardId());
         if (engineConfig.getIndexSettings().isSoftDeleteEnabled()) {
             reader = new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);

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

@@ -3159,12 +3159,14 @@ public void advanceMaxSeqNoOfUpdatesOrDeletes(long seqNo) {
     }
 
     /**
-     * Performs the pre-closing checks on the {@link IndexShard}.
+     * Performs the pre-closing action on the {@link IndexShard}.
      *
      * @throws IllegalStateException if the sanity checks failed
      */
-    public void verifyShardBeforeIndexClosing() throws IllegalStateException {
-        getEngine().verifyEngineBeforeIndexClosing();
+    public void prepareShardBeforeIndexClosing(String syncId) throws IllegalStateException {
+        // don't issue synced-flush for recovering shards
+        final boolean canSyncedFlush = state == IndexShardState.STARTED || state == IndexShardState.POST_RECOVERY;
+        getEngine().prepareEngineBeforeIndexClosing(canSyncedFlush ? syncId : null);
     }
 
     RetentionLeaseSyncer getRetentionLeaseSyncer() {

server/src/test/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseActionTests.java

@@ -20,7 +20,6 @@
 
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.replication.ReplicationOperation;
@@ -39,8 +38,8 @@
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ReplicationGroup;
 import org.elasticsearch.index.shard.ShardId;
@@ -56,7 +55,6 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.mockito.ArgumentCaptor;
 
 import java.util.Collections;
 import java.util.List;
@@ -71,8 +69,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.is;
-import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -138,7 +135,7 @@ public static void afterClass() {
     private void executeOnPrimaryOrReplica() throws Throwable {
         final TaskId taskId = new TaskId("_node_id", randomNonNegativeLong());
         final TransportVerifyShardBeforeCloseAction.ShardRequest request =
-            new TransportVerifyShardBeforeCloseAction.ShardRequest(indexShard.shardId(), clusterBlock, taskId);
+            new TransportVerifyShardBeforeCloseAction.ShardRequest(indexShard.shardId(), clusterBlock, UUIDs.randomBase64UUID(), taskId);
         final PlainActionFuture<Void> res = PlainActionFuture.newFuture();
         action.shardOperationOnPrimary(request, indexShard, ActionListener.wrap(
             r -> {
@@ -156,22 +153,12 @@ private void executeOnPrimaryOrReplica() throws Throwable {
         }
     }
 
-    public void testShardIsFlushed() throws Throwable {
-        final ArgumentCaptor<FlushRequest> flushRequest = ArgumentCaptor.forClass(FlushRequest.class);
-        when(indexShard.flush(flushRequest.capture())).thenReturn(new Engine.CommitId(new byte[0]));
-
-        executeOnPrimaryOrReplica();
-        verify(indexShard, times(1)).flush(any(FlushRequest.class));
-        assertThat(flushRequest.getValue().force(), is(true));
-    }
-
     public void testOperationFailsWhenNotBlocked() {
         when(indexShard.getActiveOperationsCount()).thenReturn(randomIntBetween(0, 10));
 
         IllegalStateException exception = expectThrows(IllegalStateException.class, this::executeOnPrimaryOrReplica);
         assertThat(exception.getMessage(),
             equalTo("Index shard " + indexShard.shardId() + " is not blocking all operations during closing"));
-        verify(indexShard, times(0)).flush(any(FlushRequest.class));
     }
 
     public void testOperationFailsWithNoBlock() {
@@ -180,20 +167,17 @@ public void testOperationFailsWithNoBlock() {
         IllegalStateException exception = expectThrows(IllegalStateException.class, this::executeOnPrimaryOrReplica);
         assertThat(exception.getMessage(),
             equalTo("Index shard " + indexShard.shardId() + " must be blocked by " + clusterBlock + " before closing"));
-        verify(indexShard, times(0)).flush(any(FlushRequest.class));
     }
 
     public void testVerifyShardBeforeIndexClosing() throws Throwable {
         executeOnPrimaryOrReplica();
-        verify(indexShard, times(1)).verifyShardBeforeIndexClosing();
-        verify(indexShard, times(1)).flush(any(FlushRequest.class));
+        verify(indexShard, times(1)).prepareShardBeforeIndexClosing(anyString());
     }
 
     public void testVerifyShardBeforeIndexClosingFailed() {
-        doThrow(new IllegalStateException("test")).when(indexShard).verifyShardBeforeIndexClosing();
+        doThrow(new IllegalStateException("test")).when(indexShard).prepareShardBeforeIndexClosing(anyString());
         expectThrows(IllegalStateException.class, this::executeOnPrimaryOrReplica);
-        verify(indexShard, times(1)).verifyShardBeforeIndexClosing();
-        verify(indexShard, times(0)).flush(any(FlushRequest.class));
+        verify(indexShard, times(1)).prepareShardBeforeIndexClosing(anyString());
     }
 
     public void testUnavailableShardsMarkedAsStale() throws Exception {
@@ -227,7 +211,7 @@ public void testUnavailableShardsMarkedAsStale() throws Exception {
         final PlainActionFuture<PrimaryResult> listener = new PlainActionFuture<>();
         TaskId taskId = new TaskId(clusterService.localNode().getId(), 0L);
         TransportVerifyShardBeforeCloseAction.ShardRequest request =
-            new TransportVerifyShardBeforeCloseAction.ShardRequest(shardId, clusterBlock, taskId);
+            new TransportVerifyShardBeforeCloseAction.ShardRequest(shardId, clusterBlock, UUIDs.randomBase64UUID(), taskId);
         ReplicationOperation.Replicas<TransportVerifyShardBeforeCloseAction.ShardRequest> proxy = action.newReplicasProxy();
         ReplicationOperation<TransportVerifyShardBeforeCloseAction.ShardRequest,
             TransportVerifyShardBeforeCloseAction.ShardRequest, PrimaryResult> operation = new ReplicationOperation<>(

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

@@ -20,6 +20,7 @@
 
 import org.apache.lucene.util.LuceneTestCase;
 import org.elasticsearch.Version;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.index.mapper.ParsedDocument;
@@ -183,10 +184,10 @@ public void testReadOnly() throws IOException {
     }
 
     /**
-     * Test that {@link ReadOnlyEngine#verifyEngineBeforeIndexClosing()} never fails
+     * Test that {@link ReadOnlyEngine#prepareEngineBeforeIndexClosing(String)} never fails
      * whatever the value of the global checkpoint to check is.
      */
-    public void testVerifyShardBeforeIndexClosingIsNoOp() throws IOException {
+    public void testPrepareShardBeforeIndexClosingIsNoOp() throws IOException {
         IOUtils.close(engine, store);
         final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
         try (Store store = createStore()) {
@@ -195,7 +196,7 @@ public void testVerifyShardBeforeIndexClosingIsNoOp() throws IOException {
             try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , null, true, Function.identity())) {
                 globalCheckpoint.set(randomNonNegativeLong());
                 try {
-                    readOnlyEngine.verifyEngineBeforeIndexClosing();
+                    readOnlyEngine.prepareEngineBeforeIndexClosing(UUIDs.randomBase64UUID());
                 } catch (final IllegalStateException e) {
                     fail("Read-only engine pre-closing verifications failed");
                 }

server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.indices.IndexClosedException;
+import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.test.BackgroundIndexer;
 import org.elasticsearch.test.ESIntegTestCase;
 
@@ -50,6 +51,7 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
@@ -98,17 +100,34 @@ public void testCloseNullIndex() {
     }
 
     public void testCloseIndex() throws Exception {
+        final int numberOfReplicas = randomIntBetween(0, 2);
+        internalCluster().ensureAtLeastNumDataNodes(numberOfReplicas + 1);
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        createIndex(indexName);
-
+        createIndex(indexName, Settings.builder()
+            .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), randomIntBetween(1, 5))
+            .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), numberOfReplicas).build());
+        ensureGreen(indexName);
         final int nbDocs = randomIntBetween(0, 50);
         indexRandom(randomBoolean(), false, randomBoolean(), IntStream.range(0, nbDocs)
             .mapToObj(i -> client().prepareIndex(indexName, "_doc", String.valueOf(i)).setSource("num", i)).collect(toList()));
 
         assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
         assertIndexIsClosed(indexName);
-
+        ensureGreen(indexName);
+        for (RecoveryState recoveryState :
+            client().admin().indices().prepareRecoveries(indexName).get().shardRecoveryStates().get(indexName)) {
+            if (recoveryState.getPrimary() == false) {
+                assertThat(recoveryState.getIndex().fileDetails(), empty());
+            }
+        }
         assertAcked(client().admin().indices().prepareOpen(indexName));
+        ensureGreen(indexName);
+        for (RecoveryState recoveryState :
+            client().admin().indices().prepareRecoveries(indexName).get().shardRecoveryStates().get(indexName)) {
+            if (recoveryState.getPrimary() == false) {
+                assertThat(recoveryState.getIndex().fileDetails(), empty());
+            }
+        }
         assertHitCount(client().prepareSearch(indexName).setSize(0).get(), nbDocs);
     }
 

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java

@@ -210,7 +210,7 @@ public long getNumberOfOptimizedIndexing() {
     }
 
     @Override
-    public void verifyEngineBeforeIndexClosing() throws IllegalStateException {
+    protected void verifyEngineBeforeIndexClosing() {
         // the value of the global checkpoint is not verified when the following engine is closed,
         // allowing it to be closed even in the case where all operations have not been fetched and
         // processed from the leader and the operations history has gaps. This way the following

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.CheckedBiFunction;
 import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
@@ -643,18 +644,20 @@ public void testProcessOnceOnPrimary() throws Exception {
     }
 
     /**
-     * Test that {@link FollowingEngine#verifyEngineBeforeIndexClosing()} never fails
+     * Test that {@link FollowingEngine#prepareEngineBeforeIndexClosing(String)} never fails
      * whatever the value of the global checkpoint to check is.
      */
-    public void testVerifyShardBeforeIndexClosingIsNoOp() throws IOException {
+    public void testPrepareShardBeforeIndexClosingIsNoOp() throws IOException {
         final long seqNo = randomIntBetween(0, Integer.MAX_VALUE);
         runIndexTest(
             seqNo,
             Engine.Operation.Origin.PRIMARY,
             (followingEngine, index) -> {
                 globalCheckpoint.set(randomNonNegativeLong());
                 try {
-                    followingEngine.verifyEngineBeforeIndexClosing();
+                    String syncId = UUIDs.randomBase64UUID();
+                    followingEngine.prepareEngineBeforeIndexClosing(syncId);
+                    assertThat(followingEngine.commitStats().syncId(), equalTo(syncId));
                 } catch (final IllegalStateException e) {
                     fail("Following engine pre-closing verifications failed");
                 }