Fix IndexMetaData loads after rollover (#33394)

When we rollover and index we write the conditions of the rollover that
the old index met into the old index. Loading this index metadata
requires a working `NamedXContentRegistry` that has been populated with
parsers from the rollover infrastructure. We had a few loads that didn't
use a working `NamedXContentRegistry` and so would fail if they ever
encountered an index that had been rolled over. Here are the locations
of the loads and how I fixed them:

* IndexFolderUpgrader - removed entirely. It existed to support opening
indices made in Elasticsearch 2.x. Since we only need this change as far
back as 6.4.1 which will supports reading from indices created as far
back as 5.0.0 we should be good here.
* TransportNodesListGatewayStartedShards - wired the
`NamedXContentRegistry` into place.
* TransportNodesListShardStoreMetaData - wired the
`NamedXContentRegistry` into place.
* OldIndexUtils - removed entirely. It existed to support the zip based
index backwards compatibility tests which we've since replaced with code
that actually runs old versions of Elasticsearch.

In addition to fixing the actual problem I added full cluster restart
integration tests for rollover which would have caught this problem and
I added an extra assertion to IndexMetaData's deserialization code which
will trip if we try to deserialize and index's metadata without a fully
formed `NamedXContentRegistry`. It won't catch if use the *wrong*
`NamedXContentRegistry` but it is better than nothing.

Closes #33316

Author: nik9000

qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java

@@ -477,6 +477,62 @@ public void testShrinkAfterUpgrade() throws IOException {
         }
     }
 
+    /**
+     * Test upgrading after a rollover. Specifically:
+     * <ol>
+     *  <li>Create an index with a write alias
+     *  <li>Write some documents to the write alias
+     *  <li>Roll over the index
+     *  <li>Make sure the document count is correct
+     *  <li>Upgrade
+     *  <li>Write some more documents to the write alias
+     *  <li>Make sure the document count is correct
+     * </ol>
+     */
+    public void testRollover() throws IOException {
+        if (runningAgainstOldCluster) {
+            Request createIndex = new Request("PUT", "/" + index + "-000001");
+            createIndex.setJsonEntity("{"
+                    + "  \"aliases\": {"
+                    + "    \"" + index + "_write\": {}"
+                    + "  }"
+                    + "}");
+            client().performRequest(createIndex);
+        }
+
+        int bulkCount = 10;
+        StringBuilder bulk = new StringBuilder();
+        for (int i = 0; i < bulkCount; i++) {
+            bulk.append("{\"index\":{}}\n");
+            bulk.append("{\"test\":\"test\"}\n");
+        }
+        Request bulkRequest = new Request("POST", "/" + index + "_write/doc/_bulk");
+        bulkRequest.setJsonEntity(bulk.toString());
+        bulkRequest.addParameter("refresh", "");
+        assertThat(EntityUtils.toString(client().performRequest(bulkRequest).getEntity()), containsString("\"errors\":false"));
+
+        if (runningAgainstOldCluster) {
+            Request rolloverRequest = new Request("POST", "/" + index + "_write/_rollover");
+            rolloverRequest.setJsonEntity("{"
+                    + "  \"conditions\": {"
+                    + "    \"max_docs\": 5"
+                    + "  }"
+                    + "}");
+            client().performRequest(rolloverRequest);
+
+            assertThat(EntityUtils.toString(client().performRequest(new Request("GET", "/_cat/indices?v")).getEntity()),
+                    containsString("testrollover-000002"));
+        }
+
+        Request countRequest = new Request("POST", "/" + index + "-*/_search");
+        countRequest.addParameter("size", "0");
+        Map<String, Object> count = entityAsMap(client().performRequest(countRequest));
+        assertNoFailures(count);
+
+        int expectedCount = bulkCount + (runningAgainstOldCluster ? 0 : bulkCount);
+        assertEquals(expectedCount, (int) XContentMapValues.extractValue("hits.total", count));
+    }
+
     void assertBasicSearchWorks(int count) throws IOException {
         logger.info("--> testing basic search");
         {
@@ -947,7 +1003,7 @@ private void checkSnapshot(String snapshotName, int count, Version tookOnVersion
         Request writeToRestoredRequest = new Request("POST", "/restored_" + index + "/doc/_bulk");
         writeToRestoredRequest.addParameter("refresh", "true");
         writeToRestoredRequest.setJsonEntity(bulk.toString());
-        client().performRequest(writeToRestoredRequest);
+        assertThat(EntityUtils.toString(client().performRequest(writeToRestoredRequest).getEntity()), containsString("\"errors\":false"));
 
         // And count to make sure the add worked
         // Make sure search finds all documents

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

@@ -45,6 +45,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -1346,6 +1347,8 @@ public void toXContent(XContentBuilder builder, IndexMetaData state) throws IOEx
 
         @Override
         public IndexMetaData fromXContent(XContentParser parser) throws IOException {
+            assert parser.getXContentRegistry() != NamedXContentRegistry.EMPTY
+                    : "loading index metadata requires a working named xcontent registry";
             return Builder.fromXContent(parser);
         }
     };

server/src/main/java/org/elasticsearch/common/util/IndexFolderUpgrader.java

@@ -35,7 +35,6 @@
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.IndexFolderUpgrader;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.plugins.MetaDataUpgrader;
@@ -84,7 +83,6 @@ public GatewayMetaState(Settings settings, NodeEnvironment nodeEnv, MetaStateSer
         if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings)) {
             try {
                 ensureNoPre019State();
-                IndexFolderUpgrader.upgradeIndicesIfNeeded(settings, nodeEnv);
                 final MetaData metaData = metaStateService.loadFullState();
                 final MetaData upgradedMetaData = upgradeMetaData(metaData, metaDataIndexUpgradeService, metaDataUpgrader);
                 // We finished global state validation and successfully checked all indices for backward compatibility

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

@@ -69,15 +69,18 @@ public class TransportNodesListGatewayStartedShards extends
     public static final String ACTION_NAME = "internal:gateway/local/started_shards";
     private final NodeEnvironment nodeEnv;
     private final IndicesService indicesService;
+    private final NamedXContentRegistry namedXContentRegistry;
 
     @Inject
     public TransportNodesListGatewayStartedShards(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                                   TransportService transportService, ActionFilters actionFilters,
-                                                  NodeEnvironment env, IndicesService indicesService) {
+                                                  NodeEnvironment env, IndicesService indicesService,
+                                                  NamedXContentRegistry namedXContentRegistry) {
         super(settings, ACTION_NAME, threadPool, clusterService, transportService, actionFilters,
             Request::new, NodeRequest::new, ThreadPool.Names.FETCH_SHARD_STARTED, NodeGatewayStartedShards.class);
         this.nodeEnv = env;
         this.indicesService = indicesService;
+        this.namedXContentRegistry = namedXContentRegistry;
     }
 
     @Override
@@ -112,15 +115,15 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) {
         try {
             final ShardId shardId = request.getShardId();
             logger.trace("{} loading local shard state info", shardId);
-            ShardStateMetaData shardStateMetaData = ShardStateMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
+            ShardStateMetaData shardStateMetaData = ShardStateMetaData.FORMAT.loadLatestState(logger, namedXContentRegistry,
                 nodeEnv.availableShardPaths(request.shardId));
             if (shardStateMetaData != null) {
                 IndexMetaData metaData = clusterService.state().metaData().index(shardId.getIndex());
                 if (metaData == null) {
                     // we may send this requests while processing the cluster state that recovered the index
                     // sometimes the request comes in before the local node processed that cluster state
                     // in such cases we can load it from disk
-                    metaData = IndexMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
+                    metaData = IndexMetaData.FORMAT.loadLatestState(logger, namedXContentRegistry,
                         nodeEnv.indexPaths(shardId.getIndex()));
                 }
                 if (metaData == null) {

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

@@ -67,17 +67,19 @@ public class TransportNodesListShardStoreMetaData extends TransportNodesAction<T
     public static final String ACTION_NAME = "internal:cluster/nodes/indices/shard/store";
 
     private final IndicesService indicesService;
-
     private final NodeEnvironment nodeEnv;
+    private final NamedXContentRegistry namedXContentRegistry;
 
     @Inject
     public TransportNodesListShardStoreMetaData(Settings settings, ThreadPool threadPool,
                                                 ClusterService clusterService, TransportService transportService,
-                                                IndicesService indicesService, NodeEnvironment nodeEnv, ActionFilters actionFilters) {
+                                                IndicesService indicesService, NodeEnvironment nodeEnv,
+                                                ActionFilters actionFilters, NamedXContentRegistry namedXContentRegistry) {
         super(settings, ACTION_NAME, threadPool, clusterService, transportService, actionFilters,
             Request::new, NodeRequest::new, ThreadPool.Names.FETCH_SHARD_STORE, NodeStoreFilesMetaData.class);
         this.indicesService = indicesService;
         this.nodeEnv = nodeEnv;
+        this.namedXContentRegistry = namedXContentRegistry;
     }
 
     @Override
@@ -129,7 +131,7 @@ private StoreFilesMetaData listStoreMetaData(ShardId shardId) throws IOException
                 // we may send this requests while processing the cluster state that recovered the index
                 // sometimes the request comes in before the local node processed that cluster state
                 // in such cases we can load it from disk
-                metaData = IndexMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY,
+                metaData = IndexMetaData.FORMAT.loadLatestState(logger, namedXContentRegistry,
                     nodeEnv.indexPaths(shardId.getIndex()));
             }
             if (metaData == null) {