diff --git a/qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java b/qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java
index b34f677e1c15b..36cb6bcadd73b 100644
--- a/qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java
+++ b/qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.Strings;
@@ -41,6 +42,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Base64;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -59,8 +61,11 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 /**
  * Tests to run before and after a full cluster restart. This is run twice,
@@ -951,6 +956,97 @@ public void testSoftDeletes() throws Exception {
         }
     }
 
+    /**
+     * This test creates an index in the old cluster and then closes it. When the cluster is fully restarted in a newer version,
+     * it verifies that the index exists and is replicated if the old version supports replication.
+     */
+    public void testClosedIndices() throws Exception {
+        if (isRunningAgainstOldCluster()) {
+            createIndex(index, Settings.builder()
+                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1)
+                .build());
+            ensureGreen(index);
+
+            int numDocs = 0;
+            if (randomBoolean()) {
+                numDocs = between(1, 100);
+                for (int i = 0; i < numDocs; i++) {
+                    final Request request = new Request("POST", "/" + index + "/_doc/" + i);
+                    request.setJsonEntity(Strings.toString(JsonXContent.contentBuilder().startObject().field("field", "v1").endObject()));
+                    assertOK(client().performRequest(request));
+                    if (rarely()) {
+                        refresh();
+                    }
+                }
+                refresh();
+            }
+
+            assertTotalHits(numDocs, entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search"))));
+            saveInfoDocument(index + "_doc_count", Integer.toString(numDocs));
+            closeIndex(index);
+        }
+
+        if (getOldClusterVersion().onOrAfter(Version.V_8_0_0)) {
+            ensureGreenLongWait(index);
+            assertClosedIndex(index, true);
+        } else {
+            assertClosedIndex(index, false);
+        }
+
+        if (isRunningAgainstOldCluster() == false) {
+            openIndex(index);
+            ensureGreen(index);
+
+            final int expectedNumDocs = Integer.parseInt(loadInfoDocument(index + "_doc_count"));
+            assertTotalHits(expectedNumDocs, entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search"))));
+        }
+    }
+
+    /**
+     * Asserts that an index is closed in the cluster state. If `checkRoutingTable` is true, it also asserts
+     * that the index has started shards.
+     */
+    @SuppressWarnings("unchecked")
+    private void assertClosedIndex(final String index, final boolean checkRoutingTable) throws IOException {
+        final Map<String, ?> state = entityAsMap(client().performRequest(new Request("GET", "/_cluster/state")));
+
+        final Map<String, ?> metadata = (Map<String, Object>) XContentMapValues.extractValue("metadata.indices." + index, state);
+        assertThat(metadata, notNullValue());
+        assertThat(metadata.get("state"), equalTo("close"));
+
+        final Map<String, ?> blocks = (Map<String, Object>) XContentMapValues.extractValue("blocks.indices." + index, state);
+        assertThat(blocks, notNullValue());
+        assertThat(blocks.containsKey(String.valueOf(MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID)), is(true));
+
+        final Map<String, ?> settings = (Map<String, Object>) XContentMapValues.extractValue("settings", metadata);
+        assertThat(settings, notNullValue());
+
+        final Map<String, ?> routingTable = (Map<String, Object>) XContentMapValues.extractValue("routing_table.indices." + index, state);
+        if (checkRoutingTable) {
+            assertThat(routingTable, notNullValue());
+            assertThat(Booleans.parseBoolean((String) XContentMapValues.extractValue("index.verified_before_close", settings)), is(true));
+            final String numberOfShards = (String) XContentMapValues.extractValue("index.number_of_shards", settings);
+            assertThat(numberOfShards, notNullValue());
+            final int nbShards = Integer.parseInt(numberOfShards);
+            assertThat(nbShards, greaterThanOrEqualTo(1));
+
+            for (int i = 0; i < nbShards; i++) {
+                final Collection<Map<String, ?>> shards =
+                    (Collection<Map<String, ?>>) XContentMapValues.extractValue("shards." + i, routingTable);
+                assertThat(shards, notNullValue());
+                assertThat(shards.size(), equalTo(2));
+                for (Map<String, ?> shard : shards) {
+                    assertThat(XContentMapValues.extractValue("shard", shard), equalTo(i));
+                    assertThat(XContentMapValues.extractValue("state", shard), equalTo("STARTED"));
+                    assertThat(XContentMapValues.extractValue("index", shard), equalTo(index));
+                }
+            }
+        } else {
+            assertThat(routingTable, nullValue());
+            assertThat(XContentMapValues.extractValue("index.verified_before_close", settings), nullValue());
+        }
+    }
+
     private void checkSnapshot(final String snapshotName, final int count, final Version tookOnVersion) throws IOException {
         // Check the snapshot metadata, especially the version
         Request listSnapshotRequest = new Request("GET", "/_snapshot/repo/" + snapshotName);
diff --git a/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java b/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java
index 295aee8b869ff..402c0c4859b77 100644
--- a/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java
+++ b/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java
@@ -24,15 +24,20 @@
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
+import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.rest.action.document.RestIndexAction;
 import org.elasticsearch.test.rest.yaml.ObjectPath;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.function.Predicate;
@@ -43,7 +48,9 @@
 import static org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider.SETTING_ALLOCATION_MAX_RETRY;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 /**
  * In depth testing of the recovery mechanism during a rolling restart.
@@ -310,4 +317,144 @@ public void testRecoveryWithSoftDeletes() throws Exception {
         }
         ensureGreen(index);
     }
+
+    /**
+     * This test creates an index in the non upgraded cluster and closes it. It then checks that the index
+     * is effectively closed and potentially replicated (if the version the index was created on supports
+     * the replication of closed indices) during the rolling upgrade.
+     */
+    public void testRecoveryClosedIndex() throws Exception {
+        final String indexName = "closed_index_created_on_old";
+        if (CLUSTER_TYPE == ClusterType.OLD) {
+            createIndex(indexName, Settings.builder()
+                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 1)
+                // if the node with the replica is the first to be restarted, while a replica is still recovering
+                // then delayed allocation will kick in. When the node comes back, the master will search for a copy
+                // but the recovering copy will be seen as invalid and the cluster health won't return to GREEN
+                // before timing out
+                .put(INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), "100ms")
+                .put(SETTING_ALLOCATION_MAX_RETRY.getKey(), "0") // fail faster
+                .build());
+            ensureGreen(indexName);
+            closeIndex(indexName);
+        }
+
+        final Version indexVersionCreated = indexVersionCreated(indexName);
+        if (indexVersionCreated.onOrAfter(Version.V_8_0_0)) {
+            // index was created on a version that supports the replication of closed indices,
+            // so we expect the index to be closed and replicated
+            ensureGreen(indexName);
+            assertClosedIndex(indexName, true);
+        } else {
+            assertClosedIndex(indexName, false);
+        }
+    }
+
+    /**
+     * This test creates and closes a new index at every stage of the rolling upgrade. It then checks that the index
+     * is effectively closed and potentially replicated if the cluster supports replication of closed indices at the
+     * time the index was closed.
+     */
+    public void testCloseIndexDuringRollingUpgrade() throws Exception {
+        final Version minimumNodeVersion = minimumNodeVersion();
+        final String indexName =
+            String.join("_", "index", CLUSTER_TYPE.toString(), Integer.toString(minimumNodeVersion.id)).toLowerCase(Locale.ROOT);
+
+        if (indexExists(indexName) == false) {
+            createIndex(indexName, Settings.builder()
+                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0)
+                .build());
+            ensureGreen(indexName);
+            closeIndex(indexName);
+        }
+
+        if (minimumNodeVersion.onOrAfter(Version.V_8_0_0)) {
+            // index is created on a version that supports the replication of closed indices,
+            // so we expect the index to be closed and replicated
+            ensureGreen(indexName);
+            assertClosedIndex(indexName, true);
+        } else {
+            assertClosedIndex(indexName, false);
+        }
+    }
+
+    /**
+     * Returns the version in which the given index has been created
+     */
+    private static Version indexVersionCreated(final String indexName) throws IOException {
+        final Request request = new Request("GET", "/" + indexName + "/_settings");
+        final String versionCreatedSetting = indexName + ".settings.index.version.created";
+        request.addParameter("filter_path", versionCreatedSetting);
+
+        final Response response = client().performRequest(request);
+        return Version.fromId(Integer.parseInt(ObjectPath.createFromResponse(response).evaluate(versionCreatedSetting)));
+    }
+
+    /**
+     * Returns the minimum node version among all nodes of the cluster
+     */
+    private static Version minimumNodeVersion() throws IOException {
+        final Request request = new Request("GET", "_nodes");
+        request.addParameter("filter_path", "nodes.*.version");
+
+        final Response response = client().performRequest(request);
+        final Map<String, Object> nodes = ObjectPath.createFromResponse(response).evaluate("nodes");
+
+        Version minVersion = null;
+        for (Map.Entry<String, Object> node : nodes.entrySet()) {
+            @SuppressWarnings("unchecked")
+            Version nodeVersion = Version.fromString((String) ((Map<String, Object>) node.getValue()).get("version"));
+            if (minVersion == null || minVersion.after(nodeVersion)) {
+                minVersion = nodeVersion;
+            }
+        }
+        assertNotNull(minVersion);
+        return minVersion;
+    }
+
+    /**
+     * Asserts that an index is closed in the cluster state. If `checkRoutingTable` is true, it also asserts
+     * that the index has started shards.
+     */
+    @SuppressWarnings("unchecked")
+    private void assertClosedIndex(final String index, final boolean checkRoutingTable) throws IOException {
+        final Map<String, ?> state = entityAsMap(client().performRequest(new Request("GET", "/_cluster/state")));
+
+        final Map<String, ?> metadata = (Map<String, Object>) XContentMapValues.extractValue("metadata.indices." + index, state);
+        assertThat(metadata, notNullValue());
+        assertThat(metadata.get("state"), equalTo("close"));
+
+        final Map<String, ?> blocks = (Map<String, Object>) XContentMapValues.extractValue("blocks.indices." + index, state);
+        assertThat(blocks, notNullValue());
+        assertThat(blocks.containsKey(String.valueOf(MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID)), is(true));
+
+        final Map<String, ?> settings = (Map<String, Object>) XContentMapValues.extractValue("settings", metadata);
+        assertThat(settings, notNullValue());
+
+        final int numberOfShards = Integer.parseInt((String) XContentMapValues.extractValue("index.number_of_shards", settings));
+        final int numberOfReplicas = Integer.parseInt((String) XContentMapValues.extractValue("index.number_of_replicas", settings));
+
+        final Map<String, ?> routingTable = (Map<String, Object>) XContentMapValues.extractValue("routing_table.indices." + index, state);
+        if (checkRoutingTable) {
+            assertThat(routingTable, notNullValue());
+            assertThat(Booleans.parseBoolean((String) XContentMapValues.extractValue("index.verified_before_close", settings)), is(true));
+
+            for (int i = 0; i < numberOfShards; i++) {
+                final Collection<Map<String, ?>> shards =
+                    (Collection<Map<String, ?>>) XContentMapValues.extractValue("shards." + i, routingTable);
+                assertThat(shards, notNullValue());
+                assertThat(shards.size(), equalTo(numberOfReplicas + 1));
+                for (Map<String, ?> shard : shards) {
+                    assertThat(XContentMapValues.extractValue("shard", shard), equalTo(i));
+                    assertThat(XContentMapValues.extractValue("state", shard), equalTo("STARTED"));
+                    assertThat(XContentMapValues.extractValue("index", shard), equalTo(index));
+                }
+            }
+        } else {
+            assertThat(routingTable, nullValue());
+            assertThat(XContentMapValues.extractValue("index.verified_before_close", settings), nullValue());
+        }
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java
index ca58b49e51860..4d81bf6e9c557 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java
@@ -52,6 +52,8 @@
 import org.elasticsearch.common.ValidationException;
 import org.elasticsearch.common.collect.ImmutableOpenIntMap;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
@@ -90,6 +92,8 @@ public class MetaDataIndexStateService {
     public static final int INDEX_CLOSED_BLOCK_ID = 4;
     public static final ClusterBlock INDEX_CLOSED_BLOCK = new ClusterBlock(4, "index closed", false,
         false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.READ_WRITE);
+    public static final Setting<Boolean> VERIFIED_BEFORE_CLOSE_SETTING =
+        Setting.boolSetting("index.verified_before_close", false, Setting.Property.IndexScope, Setting.Property.PrivateIndex);
 
     private final ClusterService clusterService;
     private final AllocationService allocationService;
@@ -402,15 +406,22 @@ static ClusterState closeRoutingTable(final ClusterState currentState,
                     continue;
                 }
 
-                logger.debug("closing index {} succeeded", index);
-                metadata.put(IndexMetaData.builder(indexMetaData).state(IndexMetaData.State.CLOSE));
                 blocks.removeIndexBlockWithId(index.getName(), INDEX_CLOSED_BLOCK_ID);
                 blocks.addIndexBlock(index.getName(), INDEX_CLOSED_BLOCK);
+                final IndexMetaData.Builder updatedMetaData = IndexMetaData.builder(indexMetaData).state(IndexMetaData.State.CLOSE);
                 if (removeRoutingTable) {
+                    metadata.put(updatedMetaData);
                     routingTable.remove(index.getName());
                 } else {
+                    metadata.put(updatedMetaData
+                        .settingsVersion(indexMetaData.getSettingsVersion() + 1)
+                        .settings(Settings.builder()
+                            .put(indexMetaData.getSettings())
+                            .put(VERIFIED_BEFORE_CLOSE_SETTING.getKey(), true)));
                     routingTable.addAsFromOpenToClose(metadata.getSafe(index));
                 }
+
+                logger.debug("closing index {} succeeded", index);
                 closedIndices.add(index.getName());
             } catch (final IndexNotFoundException e) {
                 logger.debug("index {} has been deleted since it was blocked before closing, ignoring", index);
@@ -490,7 +501,15 @@ ClusterState openIndices(final Index[] indices, final ClusterState currentState)
         for (IndexMetaData indexMetaData : indicesToOpen) {
             final Index index = indexMetaData.getIndex();
             if (indexMetaData.getState() != IndexMetaData.State.OPEN) {
-                IndexMetaData updatedIndexMetaData = IndexMetaData.builder(indexMetaData).state(IndexMetaData.State.OPEN).build();
+                final Settings.Builder updatedSettings = Settings.builder().put(indexMetaData.getSettings());
+                updatedSettings.remove(VERIFIED_BEFORE_CLOSE_SETTING.getKey());
+
+                IndexMetaData updatedIndexMetaData = IndexMetaData.builder(indexMetaData)
+                    .state(IndexMetaData.State.OPEN)
+                    .settingsVersion(indexMetaData.getSettingsVersion() + 1)
+                    .settings(updatedSettings)
+                    .build();
+
                 // The index might be closed because we couldn't import it due to old incompatible version
                 // We need to check that this index can be upgraded to the current version
                 updatedIndexMetaData = metaDataIndexUpgradeService.upgradeIndexMetaData(updatedIndexMetaData, minIndexCompatibilityVersion);
@@ -554,4 +573,9 @@ public static ClusterBlock createIndexClosingBlock() {
             EnumSet.of(ClusterBlockLevel.WRITE));
     }
 
+    public static boolean isIndexVerifiedBeforeClosed(final IndexMetaData indexMetaData) {
+        return indexMetaData.getState() == IndexMetaData.State.CLOSE
+            && VERIFIED_BEFORE_CLOSE_SETTING.exists(indexMetaData.getSettings())
+            && VERIFIED_BEFORE_CLOSE_SETTING.get(indexMetaData.getSettings());
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java b/server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java
index 32ec3e7ce5ba8..3a49577563929 100644
--- a/server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java
+++ b/server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java
@@ -47,6 +47,8 @@
 import java.util.Map;
 import java.util.function.Predicate;
 
+import static org.elasticsearch.cluster.metadata.MetaDataIndexStateService.isIndexVerifiedBeforeClosed;
+
 /**
  * Represents a global cluster-wide routing table for all indices including the
  * version of the current routing state.
@@ -499,9 +501,9 @@ public Builder addAsNew(IndexMetaData indexMetaData) {
         }
 
         public Builder addAsRecovery(IndexMetaData indexMetaData) {
-            if (indexMetaData.getState() == IndexMetaData.State.OPEN) {
+            if (indexMetaData.getState() == IndexMetaData.State.OPEN || isIndexVerifiedBeforeClosed(indexMetaData)) {
                 IndexRoutingTable.Builder indexRoutingBuilder = new IndexRoutingTable.Builder(indexMetaData.getIndex())
-                        .initializeAsRecovery(indexMetaData);
+                    .initializeAsRecovery(indexMetaData);
                 add(indexRoutingBuilder);
             }
             return this;
@@ -526,7 +528,7 @@ public Builder addAsFromCloseToOpen(IndexMetaData indexMetaData) {
         }
 
         public Builder addAsFromOpenToClose(IndexMetaData indexMetaData) {
-            assert indexMetaData.getState() == IndexMetaData.State.CLOSE;
+            assert isIndexVerifiedBeforeClosed(indexMetaData);
             IndexRoutingTable.Builder indexRoutingBuilder = new IndexRoutingTable.Builder(indexMetaData.getIndex())
                 .initializeAsFromOpenToClose(indexMetaData);
             return add(indexRoutingBuilder);
diff --git a/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java
index 9d936a28846e4..f1e42d2413815 100644
--- a/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java
+++ b/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java
@@ -19,6 +19,7 @@
 package org.elasticsearch.common.settings;
 
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider;
@@ -161,6 +162,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         EngineConfig.INDEX_CODEC_SETTING,
         IndexMetaData.SETTING_WAIT_FOR_ACTIVE_SHARDS,
         IndexSettings.DEFAULT_PIPELINE,
+        MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING,
 
         // validate that built-in similarities don't get redefined
         Setting.groupSetting("index.similarity.", (s) -> {
diff --git a/server/src/test/java/org/elasticsearch/cluster/allocation/ClusterRerouteIT.java b/server/src/test/java/org/elasticsearch/cluster/allocation/ClusterRerouteIT.java
index 95cb09df0f68c..d629804b02d58 100644
--- a/server/src/test/java/org/elasticsearch/cluster/allocation/ClusterRerouteIT.java
+++ b/server/src/test/java/org/elasticsearch/cluster/allocation/ClusterRerouteIT.java
@@ -237,6 +237,11 @@ private void rerouteWithAllocateLocalGateway(Settings commonSettings) throws Exc
                 .setSettings(Settings.builder().put("index.number_of_shards", 1))
                 .execute().actionGet();
 
+        final boolean closed = randomBoolean();
+        if (closed) {
+            client().admin().indices().prepareClose("test").get();
+        }
+
         ClusterState state = client().admin().cluster().prepareState().execute().actionGet().getState();
         assertThat(state.getRoutingNodes().unassigned().size(), equalTo(2));
 
@@ -249,8 +254,11 @@ private void rerouteWithAllocateLocalGateway(Settings commonSettings) throws Exc
         assertThat(state.getRoutingNodes().node(state.nodes().resolveNode(node_1).getId()).iterator().next().state(),
             equalTo(ShardRoutingState.INITIALIZING));
 
-        healthResponse = client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID)
-            .setWaitForYellowStatus().execute().actionGet();
+        healthResponse = client().admin().cluster().prepareHealth()
+            .setIndices("test")
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForYellowStatus()
+            .execute().actionGet();
         assertThat(healthResponse.isTimedOut(), equalTo(false));
 
         logger.info("--> get the state, verify shard 1 primary allocated");
@@ -259,8 +267,10 @@ private void rerouteWithAllocateLocalGateway(Settings commonSettings) throws Exc
         assertThat(state.getRoutingNodes().node(state.nodes().resolveNode(node_1).getId()).iterator().next().state(),
             equalTo(ShardRoutingState.STARTED));
 
-        client().prepareIndex("test", "type", "1").setSource("field", "value")
-            .setRefreshPolicy(RefreshPolicy.IMMEDIATE).get();
+        if (closed == false) {
+            client().prepareIndex("test", "type", "1").setSource("field", "value")
+                .setRefreshPolicy(RefreshPolicy.IMMEDIATE).get();
+        }
         final Index index = resolveIndex("test");
 
         logger.info("--> closing all nodes");
@@ -278,7 +288,10 @@ private void rerouteWithAllocateLocalGateway(Settings commonSettings) throws Exc
         // wait a bit for the cluster to realize that the shard is not there...
         // TODO can we get around this? the cluster is RED, so what do we wait for?
         client().admin().cluster().prepareReroute().get();
-        assertThat(client().admin().cluster().prepareHealth().setWaitForNodes("2").execute().actionGet().getStatus(),
+        assertThat(client().admin().cluster().prepareHealth()
+                .setIndices("test")
+                .setWaitForNodes("2")
+                .execute().actionGet().getStatus(),
             equalTo(ClusterHealthStatus.RED));
         logger.info("--> explicitly allocate primary");
         state = client().admin().cluster().prepareReroute()
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateServiceTests.java
index df07982f445ba..6ba85cd22a36e 100644
--- a/server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateServiceTests.java
@@ -152,7 +152,10 @@ public void testCloseRoutingTableRemovesRoutingTable() {
         }
         for (Index blockedIndex : blockedIndices.keySet()) {
             if (results.get(blockedIndex).isAcknowledged()) {
-                assertThat(state.metaData().index(blockedIndex).getState(), is(IndexMetaData.State.CLOSE));
+                IndexMetaData indexMetaData = state.metaData().index(blockedIndex);
+                assertThat(indexMetaData.getState(), is(IndexMetaData.State.CLOSE));
+                Settings indexSettings = indexMetaData.getSettings();
+                assertThat(indexSettings.hasValue(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(false));
                 assertThat(state.blocks().hasIndexBlock(blockedIndex.getName(), MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(true));
                 assertThat("Index must have only 1 block with [id=" + MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID + "]",
                     state.blocks().indices().getOrDefault(blockedIndex.getName(), emptySet()).stream()
@@ -191,7 +194,6 @@ public void testAddIndexClosedBlocks() {
             ClusterState updatedState = MetaDataIndexStateService.addIndexClosedBlocks(indices, blockedIndices, state);
             assertSame(state, updatedState);
             assertTrue(blockedIndices.isEmpty());
-
         }
         {
             final Map<Index, ClusterBlock> blockedIndices = new HashMap<>();
@@ -302,6 +304,32 @@ public void testValidateShardLimit() {
             currentShards + "]/[" + maxShards + "] maximum shards open;", exception.getMessage());
     }
 
+    public void testIsIndexVerifiedBeforeClosed() {
+        final ClusterState initialState = ClusterState.builder(new ClusterName("testIsIndexMetaDataClosed")).build();
+        {
+            String indexName = "open";
+            ClusterState state = addOpenedIndex(indexName, randomIntBetween(1, 3), randomIntBetween(0, 3), initialState);
+            assertFalse(MetaDataIndexStateService.isIndexVerifiedBeforeClosed(state.getMetaData().index(indexName)));
+        }
+        {
+            String indexName = "closed";
+            ClusterState state = addClosedIndex(indexName, randomIntBetween(1, 3), randomIntBetween(0, 3), initialState);
+            assertTrue(MetaDataIndexStateService.isIndexVerifiedBeforeClosed(state.getMetaData().index(indexName)));
+        }
+        {
+            String indexName = "closed-no-setting";
+            IndexMetaData indexMetaData = IndexMetaData.builder(indexName)
+                .state(IndexMetaData.State.CLOSE)
+                .creationDate(randomNonNegativeLong())
+                .settings(Settings.builder()
+                    .put(SETTING_VERSION_CREATED, Version.CURRENT)
+                    .put(SETTING_NUMBER_OF_SHARDS, randomIntBetween(1, 3))
+                    .put(SETTING_NUMBER_OF_REPLICAS, randomIntBetween(0, 3)))
+                .build();
+            assertFalse(MetaDataIndexStateService.isIndexVerifiedBeforeClosed(indexMetaData));
+        }
+    }
+
     public static ClusterState createClusterForShardLimitTest(int nodesInCluster, int openIndexShards, int openIndexReplicas,
                                                               int closedIndexShards, int closedIndexReplicas, Settings clusterSettings) {
         ImmutableOpenMap.Builder<String, DiscoveryNode> dataNodes = ImmutableOpenMap.builder();
@@ -374,13 +402,18 @@ private static ClusterState addIndex(final ClusterState currentState,
                                          final int numReplicas,
                                          final IndexMetaData.State state,
                                          @Nullable final ClusterBlock block) {
+
+        final Settings.Builder settings = Settings.builder()
+            .put(SETTING_VERSION_CREATED, Version.CURRENT)
+            .put(SETTING_NUMBER_OF_SHARDS, numShards)
+            .put(SETTING_NUMBER_OF_REPLICAS, numReplicas);
+        if (state == IndexMetaData.State.CLOSE) {
+            settings.put(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), true);
+        }
         final IndexMetaData indexMetaData = IndexMetaData.builder(index)
             .state(state)
             .creationDate(randomNonNegativeLong())
-            .settings(Settings.builder()
-                .put(SETTING_VERSION_CREATED, Version.CURRENT)
-                .put(SETTING_NUMBER_OF_SHARDS, numShards)
-                .put(SETTING_NUMBER_OF_REPLICAS, numReplicas))
+            .settings(settings)
             .build();
 
         final ClusterState.Builder clusterStateBuilder = ClusterState.builder(currentState);
@@ -405,12 +438,20 @@ private static ClusterState addIndex(final ClusterState currentState,
     }
 
     private static void assertIsOpened(final String indexName, final ClusterState clusterState) {
-        assertThat(clusterState.metaData().index(indexName).getState(), is(IndexMetaData.State.OPEN));
+        final IndexMetaData indexMetaData = clusterState.metaData().indices().get(indexName);
+        assertThat(indexMetaData.getState(), is(IndexMetaData.State.OPEN));
+        assertThat(indexMetaData.getSettings().hasValue(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(false));
+        assertThat(clusterState.routingTable().index(indexName), notNullValue());
+        assertThat(clusterState.blocks().hasIndexBlock(indexName, MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(false));
         assertThat(clusterState.routingTable().index(indexName), notNullValue());
     }
 
     private static void assertIsClosed(final String indexName, final ClusterState clusterState) {
-        assertThat(clusterState.metaData().index(indexName).getState(), is(IndexMetaData.State.CLOSE));
+        final IndexMetaData indexMetaData = clusterState.metaData().indices().get(indexName);
+        assertThat(indexMetaData.getState(), is(IndexMetaData.State.CLOSE));
+        final Settings indexSettings = indexMetaData.getSettings();
+        assertThat(indexSettings.hasValue(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(true));
+        assertThat(indexSettings.getAsBoolean(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), false), is(true));
         assertThat(clusterState.blocks().hasIndexBlock(indexName, MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(true));
         assertThat("Index " + indexName + " must have only 1 block with [id=" + MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID + "]",
             clusterState.blocks().indices().getOrDefault(indexName, emptySet()).stream()
diff --git a/server/src/test/java/org/elasticsearch/cluster/routing/RoutingTableTests.java b/server/src/test/java/org/elasticsearch/cluster/routing/RoutingTableTests.java
index a6c2fab5c91e4..851fe9c550270 100644
--- a/server/src/test/java/org/elasticsearch/cluster/routing/RoutingTableTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/routing/RoutingTableTests.java
@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.ESAllocationTestCase;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.node.DiscoveryNodes.Builder;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
@@ -38,6 +39,7 @@
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
+import static org.elasticsearch.cluster.routing.ShardRoutingState.UNASSIGNED;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -373,6 +375,36 @@ public void testDistinctNodes() {
         assertFalse(IndexShardRoutingTable.Builder.distinctNodes(Arrays.asList(routing2, routing4)));
     }
 
+    public void testAddAsRecovery() {
+        {
+            final IndexMetaData indexMetaData = createIndexMetaData(TEST_INDEX_1).state(IndexMetaData.State.OPEN).build();
+            final RoutingTable routingTable = new RoutingTable.Builder().addAsRecovery(indexMetaData).build();
+            assertThat(routingTable.hasIndex(TEST_INDEX_1), is(true));
+            assertThat(routingTable.allShards(TEST_INDEX_1).size(), is(this.shardsPerIndex));
+            assertThat(routingTable.index(TEST_INDEX_1).shardsWithState(UNASSIGNED).size(), is(this.shardsPerIndex));
+        }
+        {
+            final IndexMetaData indexMetaData = createIndexMetaData(TEST_INDEX_1).state(IndexMetaData.State.CLOSE).build();
+            final RoutingTable routingTable = new RoutingTable.Builder().addAsRecovery(indexMetaData).build();
+            assertThat(routingTable.hasIndex(TEST_INDEX_1), is(false));
+            expectThrows(IndexNotFoundException.class, () -> routingTable.allShards(TEST_INDEX_1));
+        }
+        {
+            final IndexMetaData indexMetaData = createIndexMetaData(TEST_INDEX_1).build();
+            final IndexMetaData.Builder indexMetaDataBuilder = IndexMetaData.builder(indexMetaData)
+                .state(IndexMetaData.State.CLOSE)
+                .settings(Settings.builder()
+                    .put(indexMetaData.getSettings())
+                    .put(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), true)
+                    .build())
+                .settingsVersion(indexMetaData.getSettingsVersion() + 1);
+            final RoutingTable routingTable = new RoutingTable.Builder().addAsRecovery(indexMetaDataBuilder.build()).build();
+            assertThat(routingTable.hasIndex(TEST_INDEX_1), is(true));
+            assertThat(routingTable.allShards(TEST_INDEX_1).size(), is(this.shardsPerIndex));
+            assertThat(routingTable.index(TEST_INDEX_1).shardsWithState(UNASSIGNED).size(), is(this.shardsPerIndex));
+        }
+    }
+
     /** reverse engineer the in sync aid based on the given indexRoutingTable **/
     public static IndexMetaData updateActiveAllocations(IndexRoutingTable indexRoutingTable, IndexMetaData indexMetaData) {
         IndexMetaData.Builder imdBuilder = IndexMetaData.builder(indexMetaData);
diff --git a/server/src/test/java/org/elasticsearch/gateway/ClusterStateUpdatersTests.java b/server/src/test/java/org/elasticsearch/gateway/ClusterStateUpdatersTests.java
index cae33db90a6bc..eec02438f0031 100644
--- a/server/src/test/java/org/elasticsearch/gateway/ClusterStateUpdatersTests.java
+++ b/server/src/test/java/org/elasticsearch/gateway/ClusterStateUpdatersTests.java
@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.coordination.CoordinationMetaData;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
@@ -241,11 +242,36 @@ public void testUpdateRoutingTable() {
                 .build();
         assertFalse(initialState.routingTable().hasIndex(index));
 
-        final ClusterState newState = updateRoutingTable(initialState);
-
-        assertTrue(newState.routingTable().hasIndex(index));
-        assertThat(newState.routingTable().version(), is(0L));
-        assertThat(newState.routingTable().allShards(index.getName()).size(), is(numOfShards));
+        {
+            final ClusterState newState = updateRoutingTable(initialState);
+            assertTrue(newState.routingTable().hasIndex(index));
+            assertThat(newState.routingTable().version(), is(0L));
+            assertThat(newState.routingTable().allShards(index.getName()).size(), is(numOfShards));
+        }
+        {
+            final ClusterState newState = updateRoutingTable(ClusterState.builder(initialState)
+                .metaData(MetaData.builder(initialState.metaData())
+                    .put(IndexMetaData.builder(initialState.metaData().index("test"))
+                        .state(IndexMetaData.State.CLOSE))
+                    .build())
+                .build());
+            assertFalse(newState.routingTable().hasIndex(index));
+        }
+        {
+            final ClusterState newState = updateRoutingTable(ClusterState.builder(initialState)
+                .metaData(MetaData.builder(initialState.metaData())
+                    .put(IndexMetaData.builder(initialState.metaData().index("test"))
+                        .state(IndexMetaData.State.CLOSE)
+                        .settings(Settings.builder()
+                            .put(initialState.metaData().index("test").getSettings())
+                            .put(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), true)
+                            .build())
+                    ).build())
+                .build());
+            assertTrue(newState.routingTable().hasIndex(index));
+            assertThat(newState.routingTable().version(), is(0L));
+            assertThat(newState.routingTable().allShards(index.getName()).size(), is(numOfShards));
+        }
     }
 
     public void testMixCurrentAndRecoveredState() {
diff --git a/server/src/test/java/org/elasticsearch/gateway/GatewayIndexStateIT.java b/server/src/test/java/org/elasticsearch/gateway/GatewayIndexStateIT.java
index 541a24247473a..5b30b85cb72ec 100644
--- a/server/src/test/java/org/elasticsearch/gateway/GatewayIndexStateIT.java
+++ b/server/src/test/java/org/elasticsearch/gateway/GatewayIndexStateIT.java
@@ -59,7 +59,6 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
 
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0)
 public class GatewayIndexStateIT extends ESIntegTestCase {
@@ -166,7 +165,7 @@ public void testSimpleOpenClose() throws Exception {
 
         stateResponse = client().admin().cluster().prepareState().execute().actionGet();
         assertThat(stateResponse.getState().metaData().index("test").getState(), equalTo(IndexMetaData.State.CLOSE));
-        assertThat(stateResponse.getState().routingTable().index("test"), nullValue());
+        assertThat(stateResponse.getState().routingTable().index("test"), notNullValue());
 
         logger.info("--> trying to index into a closed index ...");
         try {
diff --git a/server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java b/server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java
index ca3f6e694097d..42f29e99982cc 100644
--- a/server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java
+++ b/server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java
@@ -308,7 +308,11 @@ public void testConcurrentClosesAndOpens() throws Exception {
     static void assertIndexIsClosed(final String... indices) {
         final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
         for (String index : indices) {
-            assertThat(clusterState.metaData().indices().get(index).getState(), is(IndexMetaData.State.CLOSE));
+            final IndexMetaData indexMetaData = clusterState.metaData().indices().get(index);
+            assertThat(indexMetaData.getState(), is(IndexMetaData.State.CLOSE));
+            final Settings indexSettings = indexMetaData.getSettings();
+            assertThat(indexSettings.hasValue(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(true));
+            assertThat(indexSettings.getAsBoolean(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), false), is(true));
             assertThat(clusterState.routingTable().index(index), notNullValue());
             assertThat(clusterState.blocks().hasIndexBlock(index, MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(true));
             assertThat("Index " + index + " must have only 1 block with [id=" + MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID + "]",
@@ -320,7 +324,9 @@ static void assertIndexIsClosed(final String... indices) {
     static void assertIndexIsOpened(final String... indices) {
         final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
         for (String index : indices) {
-            assertThat(clusterState.metaData().indices().get(index).getState(), is(IndexMetaData.State.OPEN));
+            final IndexMetaData indexMetaData = clusterState.metaData().indices().get(index);
+            assertThat(indexMetaData.getState(), is(IndexMetaData.State.OPEN));
+            assertThat(indexMetaData.getSettings().hasValue(MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(false));
             assertThat(clusterState.routingTable().index(index), notNullValue());
             assertThat(clusterState.blocks().hasIndexBlock(index, MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(false));
         }
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java
index 2dc59ec4a43fd..76c356d59f7dd 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java
@@ -945,7 +945,7 @@ public void testUpdateAnalysisLeaderIndexSettings() throws Exception {
         }
 
         assertBusy(() -> {
-            assertThat(getFollowTaskSettingsVersion("follower"), equalTo(2L));
+            assertThat(getFollowTaskSettingsVersion("follower"), equalTo(4L));
             assertThat(getFollowTaskMappingVersion("follower"), equalTo(2L));
 
             GetSettingsRequest getSettingsRequest = new GetSettingsRequest();
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/index/engine/FrozenIndexTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/index/engine/FrozenIndexTests.java
index 983b186c4ccf6..9231bad9a8dfe 100644
--- a/x-pack/plugin/core/src/test/java/org/elasticsearch/index/engine/FrozenIndexTests.java
+++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/index/engine/FrozenIndexTests.java
@@ -349,7 +349,7 @@ public void testFreezeIndexIncreasesIndexSettingsVersion() throws ExecutionExcep
         assertAcked(xPackClient.freeze(new TransportFreezeIndexAction.FreezeRequest(index)));
         assertIndexFrozen(index);
         assertThat(client().admin().cluster().prepareState().get().getState().metaData().index(index).getSettingsVersion(),
-            equalTo(settingsVersion + 1));
+            greaterThan(settingsVersion));
     }
 
     public void testFreezeEmptyIndexWithTranslogOps() throws Exception {