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/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/test/java/org/elasticsearch/indices/state/CloseIndexIT.java

@@ -23,6 +23,8 @@
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequestBuilder;
 import org.elasticsearch.action.admin.indices.close.CloseIndexResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -37,6 +39,8 @@
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.seqno.RetentionLease;
+import org.elasticsearch.index.seqno.RetentionLeases;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.indices.IndicesService;
@@ -62,7 +66,9 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
@@ -482,6 +488,39 @@ public void testResyncPropagatePrimaryTerm() throws Exception {
         }
     }
 
+    public void testAddAndSyncRetentionLeases() throws Exception {
+        internalCluster().ensureAtLeastNumDataNodes(2);
+        final String indexName = "closed_index_sync_retention_leases";
+        createIndex(indexName, Settings.builder()
+            .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomIntBetween(0, 1))
+            .build());
+        indexRandom(randomBoolean(), randomBoolean(), randomBoolean(), IntStream.range(0, randomIntBetween(0, 50))
+            .mapToObj(n -> client().prepareIndex(indexName, "_doc", Integer.toString(n)).setSource("num", n)).collect(toList()));
+        ensureGreen(indexName);
+        assertAcked(client().admin().indices().prepareClose(indexName));
+        assertIndexIsClosed(indexName);
+        ensureGreen(indexName);
+        List<RetentionLease> addedRetentionLeases = new ArrayList<>();
+        for (String nodeName : internalCluster().nodesInclude(indexName)) {
+            IndexShard indexShard = internalCluster().getInstance(IndicesService.class, nodeName).
+                indexService(resolveIndex(indexName)).getShard(0);
+            if (indexShard.routingEntry().primary()) {
+                int numLeases = randomIntBetween(1, 10);
+                for (int i = 0; i < numLeases; i++) {
+                    PlainActionFuture<ReplicationResponse> future = new PlainActionFuture<>();
+                    addedRetentionLeases.add(indexShard.addRetentionLease("test-lease-" + i, randomNonNegativeLong(), "test", future));
+                    future.actionGet();
+                }
+            }
+        }
+        for (String nodeName : internalCluster().nodesInclude(indexName)) {
+            RetentionLeases retentionLeases = internalCluster().getInstance(IndicesService.class, nodeName).
+                indexService(resolveIndex(indexName)).getShard(0).getRetentionLeases();
+            assertThat(addedRetentionLeases, everyItem(is(in(retentionLeases.leases()))));
+        }
+    }
+
     private static void closeIndices(final String... indices) {
         closeIndices(client().admin().indices().prepareClose(indices));
     }