Omit writing index metadata for non-replicated closed indices on data-only node (#47285)

Fixes a bug related to how "closed replicated indices" (introduced in 7.2) interact with the index
metadata storage mechanism, which has special handling for closed indices (but incorrectly
handles replicated closed indices). On non-master-eligible data nodes, it's possible for the
node's manifest file (which tracks the relevant metadata state that the node should persist) to
become out of sync with what's actually stored on disk, leading to an inconsistency that is then
detected at startup, refusing for the node to start up.

Closes #47276

Author: ywelsch

server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java

@@ -509,20 +509,6 @@ private static Set<Index> getRelevantIndicesOnDataOnlyNode(ClusterState state, C
         for (ShardRouting routing : newRoutingNode) {
             indices.add(routing.index());
         }
-        // we have to check the meta data also: closed indices will not appear in the routing table, but we must still write the state if
-        // we have it written on disk previously
-        for (IndexMetaData indexMetaData : state.metaData()) {
-            boolean isOrWasClosed = indexMetaData.getState().equals(IndexMetaData.State.CLOSE);
-            // if the index is open we might still have to write the state if it just transitioned from closed to open
-            // so we have to check for that as well.
-            IndexMetaData previousMetaData = previousState.metaData().index(indexMetaData.getIndex());
-            if (previousMetaData != null) {
-                isOrWasClosed = isOrWasClosed || previousMetaData.getState().equals(IndexMetaData.State.CLOSE);
-            }
-            if (previouslyWrittenIndices.contains(indexMetaData.getIndex()) && isOrWasClosed) {
-                indices.add(indexMetaData.getIndex());
-            }
-        }
         return indices;
     }
 

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

@@ -199,7 +199,7 @@ public void testGetRelevantIndicesForClosedPrevWrittenIndexOnDataOnlyNode() {
                 clusterStateWithClosedIndex(indexMetaData, false),
                 clusterStateWithAssignedIndex(indexMetaData, false),
                 Collections.singleton(indexMetaData.getIndex()));
-        assertThat(indices.size(), equalTo(1));
+        assertThat(indices.size(), equalTo(0));
     }
 
     public void testGetRelevantIndicesForClosedPrevNotWrittenIndexOnDataOnlyNode() {
@@ -217,7 +217,7 @@ public void testGetRelevantIndicesForWasClosedPrevWrittenIndexOnDataOnlyNode() {
                 clusterStateWithJustOpenedIndex(indexMetaData, false),
                 clusterStateWithClosedIndex(indexMetaData, false),
                 Collections.singleton(indexMetaData.getIndex()));
-        assertThat(indices.size(), equalTo(1));
+        assertThat(indices.size(), equalTo(0));
     }
 
     public void testResolveStatesToBeWritten() throws WriteStateException {

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

@@ -435,6 +435,31 @@ public Settings onNodeStopped(String nodeName) throws Exception {
         }
     }
 
+    /**
+     * Test for https://github.com/elastic/elasticsearch/issues/47276 which checks that the persisted metadata on a data node does not
+     * become inconsistent when using replicated closed indices.
+     */
+    public void testRelocatedClosedIndexIssue() throws Exception {
+        final String indexName = "closed-index";
+        final List<String> dataNodes = internalCluster().startDataOnlyNodes(2);
+        // allocate shard to first data node
+        createIndex(indexName, Settings.builder()
+            .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+            .put("index.routing.allocation.include._name", dataNodes.get(0))
+            .build());
+        indexRandom(randomBoolean(), randomBoolean(), randomBoolean(), IntStream.range(0, randomIntBetween(0, 50))
+            .mapToObj(n -> client().prepareIndex(indexName, "_doc").setSource("num", n)).collect(toList()));
+        assertAcked(client().admin().indices().prepareClose(indexName).setWaitForActiveShards(ActiveShardCount.ALL));
+        // move single shard to second node
+        client().admin().indices().prepareUpdateSettings(indexName).setSettings(Settings.builder()
+            .put("index.routing.allocation.include._name", dataNodes.get(1))).get();
+        ensureGreen(indexName);
+        internalCluster().fullRestart();
+        assertIndexIsClosed(indexName);
+        ensureGreen(indexName);
+    }
+
     public void testResyncPropagatePrimaryTerm() throws Exception {
         internalCluster().ensureAtLeastNumDataNodes(3);
         final String indexName = "closed_indices_promotion";