Use retention lease in peer recovery of closed indices

server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -664,7 +664,7 @@ protected void doRun() {
                     retry(new IndexNotFoundException(concreteIndex));
                     return;
                 }
-                if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
+                if (request.indicesOptions().forbidClosedIndices() && indexMetaData.getState() == IndexMetaData.State.CLOSE) {
                     throw new IndexClosedException(indexMetaData.getIndex());
                 }
 

server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.GroupedActionListener;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
-import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.AllocationId;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -827,10 +826,7 @@ private boolean invariant() {
             assert checkpoints.get(aId) != null : "aId [" + aId + "] is pending in sync but isn't tracked";
         }
 
-        if (primaryMode
-            && indexSettings.isSoftDeleteEnabled()
-            && indexSettings.getIndexMetaData().getState() == IndexMetaData.State.OPEN
-            && hasAllPeerRecoveryRetentionLeases) {
+        if (primaryMode && indexSettings.isSoftDeleteEnabled() && hasAllPeerRecoveryRetentionLeases) {
             // all tracked shard copies have a corresponding peer-recovery retention lease
             for (final ShardRouting shardRouting : routingTable.assignedShards()) {
                 if (checkpoints.get(shardRouting.allocationId().getId()).tracked) {
@@ -1011,34 +1007,32 @@ private void addPeerRecoveryRetentionLeaseForSolePrimary() {
         assert primaryMode;
         assert Thread.holdsLock(this);
 
-        if (indexSettings().getIndexMetaData().getState() == IndexMetaData.State.OPEN) {
-            final ShardRouting primaryShard = routingTable.primaryShard();
-            final String leaseId = getPeerRecoveryRetentionLeaseId(primaryShard);
-            if (retentionLeases.get(leaseId) == null) {
-                if (replicationGroup.getReplicationTargets().equals(Collections.singletonList(primaryShard))) {
-                    assert primaryShard.allocationId().getId().equals(shardAllocationId)
-                        : routingTable.assignedShards() + " vs " + shardAllocationId;
-                    // Safe to call innerAddRetentionLease() without a subsequent sync since there are no other members of this replication
-                    // group.
-                    logger.trace("addPeerRecoveryRetentionLeaseForSolePrimary: adding lease [{}]", leaseId);
-                    innerAddRetentionLease(leaseId, Math.max(0L, checkpoints.get(shardAllocationId).globalCheckpoint + 1),
-                        PEER_RECOVERY_RETENTION_LEASE_SOURCE);
-                    hasAllPeerRecoveryRetentionLeases = true;
-                } else {
-                    /*
-                     * We got here here via a rolling upgrade from an older version that doesn't create peer recovery retention
-                     * leases for every shard copy, but in this case we do not expect any leases to exist.
-                     */
-                    assert hasAllPeerRecoveryRetentionLeases == false : routingTable + " vs " + retentionLeases;
-                    logger.debug("{} becoming primary of {} with missing lease: {}", primaryShard, routingTable, retentionLeases);
-                }
-            } else if (hasAllPeerRecoveryRetentionLeases == false && routingTable.assignedShards().stream().allMatch(shardRouting ->
-                retentionLeases.contains(getPeerRecoveryRetentionLeaseId(shardRouting))
-                    || checkpoints.get(shardRouting.allocationId().getId()).tracked == false)) {
-                // Although this index is old enough not to have all the expected peer recovery retention leases, in fact it does, so we
-                // don't need to do any more work.
+        final ShardRouting primaryShard = routingTable.primaryShard();
+        final String leaseId = getPeerRecoveryRetentionLeaseId(primaryShard);
+        if (retentionLeases.get(leaseId) == null) {
+            if (replicationGroup.getReplicationTargets().equals(Collections.singletonList(primaryShard))) {
+                assert primaryShard.allocationId().getId().equals(shardAllocationId)
+                    : routingTable.assignedShards() + " vs " + shardAllocationId;
+                // Safe to call innerAddRetentionLease() without a subsequent sync since there are no other members of this replication
+                // group.
+                logger.trace("addPeerRecoveryRetentionLeaseForSolePrimary: adding lease [{}]", leaseId);
+                innerAddRetentionLease(leaseId, Math.max(0L, checkpoints.get(shardAllocationId).globalCheckpoint + 1),
+                    PEER_RECOVERY_RETENTION_LEASE_SOURCE);
                 hasAllPeerRecoveryRetentionLeases = true;
+            } else {
+                /*
+                 * We got here here via a rolling upgrade from an older version that doesn't create peer recovery retention
+                 * leases for every shard copy, but in this case we do not expect any leases to exist.
+                 */
+                assert hasAllPeerRecoveryRetentionLeases == false : routingTable + " vs " + retentionLeases;
+                logger.debug("{} becoming primary of {} with missing lease: {}", primaryShard, routingTable, retentionLeases);
             }
+        } else if (hasAllPeerRecoveryRetentionLeases == false && routingTable.assignedShards().stream().allMatch(shardRouting ->
+            retentionLeases.contains(getPeerRecoveryRetentionLeaseId(shardRouting))
+                || checkpoints.get(shardRouting.allocationId().getId()).tracked == false)) {
+            // Although this index is old enough not to have all the expected peer recovery retention leases, in fact it does, so we
+            // don't need to do any more work.
+            hasAllPeerRecoveryRetentionLeases = true;
         }
     }
 
@@ -1356,9 +1350,7 @@ private synchronized void setHasAllPeerRecoveryRetentionLeases() {
      * prior to {@link Version#V_7_4_0} that does not create peer-recovery retention leases.
      */
     public synchronized void createMissingPeerRecoveryRetentionLeases(ActionListener<Void> listener) {
-        if (indexSettings().isSoftDeleteEnabled()
-            && indexSettings().getIndexMetaData().getState() == IndexMetaData.State.OPEN
-            && hasAllPeerRecoveryRetentionLeases == false) {
+        if (indexSettings().isSoftDeleteEnabled() && hasAllPeerRecoveryRetentionLeases == false) {
 
             final List<ShardRouting> shardRoutings = routingTable.assignedShards();
             final GroupedActionListener<ReplicationResponse> groupedActionListener = new GroupedActionListener<>(ActionListener.wrap(vs -> {

server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.WriteResponse;
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -117,6 +118,9 @@ public ClusterBlockLevel indexBlockLevel() {
     }
 
     public static final class Request extends ReplicatedWriteRequest<Request> {
+        // allow adding retention leases for peer recovery on closed indices
+        private static final IndicesOptions INDICES_OPTIONS =
+            IndicesOptions.fromOptions(false, false, false, false, false, false, true, false);
 
         private RetentionLeases retentionLeases;
 
@@ -141,6 +145,11 @@ public void writeTo(final StreamOutput out) throws IOException {
             retentionLeases.writeTo(out);
         }
 
+        @Override
+        public IndicesOptions indicesOptions() {
+            return INDICES_OPTIONS;
+        }
+
         @Override
         public String toString() {
             return "RetentionLeaseSyncAction.Request{" +

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

@@ -38,7 +38,6 @@
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.ThreadedActionListener;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
-import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.CheckedSupplier;
@@ -154,8 +153,7 @@ public void recoverToTarget(ActionListener<RecoveryResponse> listener) {
                 IOUtils.closeWhileHandlingException(releaseResources, () -> wrappedListener.onFailure(e));
             };
 
-            final boolean useRetentionLeases = shard.indexSettings().isSoftDeleteEnabled()
-                && shard.indexSettings().getIndexMetaData().getState() != IndexMetaData.State.CLOSE;
+            final boolean useRetentionLeases = shard.indexSettings().isSoftDeleteEnabled();
             final SetOnce<RetentionLease> retentionLeaseRef = new SetOnce<>();
 
             runUnderPrimaryPermit(() -> {

server/src/test/java/org/elasticsearch/gateway/ReplicaShardAllocatorIT.java

@@ -282,6 +282,40 @@ public void testPreferCopyWithHighestMatchingOperations() throws Exception {
         assertThat(internalCluster().nodesInclude(indexName), allOf(hasItem(nodeWithHigherMatching), not(hasItem(nodeWithLowerMatching))));
     }
 
+    public void testPeerRecoveryForClosedIndices() throws Exception {
+        String indexName = "peer_recovery_closed_indices";
+        internalCluster().ensureAtLeastNumDataNodes(1);
+        createIndex(indexName, Settings.builder()
+            .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+            .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
+            .put(IndexService.GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.getKey(), "100ms")
+            .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "100ms")
+            .build());
+        indexRandom(randomBoolean(), randomBoolean(), randomBoolean(), IntStream.range(0, randomIntBetween(1, 100))
+            .mapToObj(n -> client().prepareIndex(indexName).setSource("num", n)).collect(Collectors.toList()));
+        ensureActivePeerRecoveryRetentionLeasesAdvanced(indexName);
+        assertAcked(client().admin().indices().prepareClose(indexName));
+        int numberOfReplicas = randomIntBetween(1, 2);
+        internalCluster().ensureAtLeastNumDataNodes(2 + numberOfReplicas);
+        assertAcked(client().admin().indices().prepareUpdateSettings(indexName)
+            .setSettings(Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas)));
+        ensureGreen(indexName);
+        ensureActivePeerRecoveryRetentionLeasesAdvanced(indexName);
+        assertAcked(client().admin().cluster().prepareUpdateSettings()
+            .setPersistentSettings(Settings.builder().put("cluster.routing.allocation.enable", "primaries").build()));
+        internalCluster().fullRestart();
+        ensureYellow(indexName);
+        if (randomBoolean()) {
+            assertAcked(client().admin().indices().prepareOpen(indexName));
+            client().admin().indices().prepareForceMerge(indexName).get();
+        }
+        assertAcked(client().admin().cluster().prepareUpdateSettings()
+            .setPersistentSettings(Settings.builder().putNull("cluster.routing.allocation.enable").build()));
+        ensureGreen(indexName);
+        assertNoOpRecoveries(indexName);
+    }
+
     private void ensureActivePeerRecoveryRetentionLeasesAdvanced(String indexName) throws Exception {
         assertBusy(() -> {
             Index index = resolveIndex(indexName);