Keep track of aliased indices (#80873)

The PR adds a top level `aliasedIndices` map to `Metadata` that keeps
track of all the aliases that are defined in each `IndexMetadata`. This
`aliasedIndices` map isn't rebuild on each `Metadata#build(...)`
invocation, but rather updated in place each time alias definitions are
added or removed from `IndexMetadata` instances. New `Metadata.Builder`
instances reuse the previous `aliasedIndices` map.

This change is beneficial because it will allow the `indicesLookup`
sorted map to always be build lazily (which is what this PR is also
doing). Prior to this change building the `indicesLookup` sorted map was
required for certain alias validation, but this is no longer the case,
since that validation can now use `aliasedIndices` map. Building the
`indicesLookup` sorted map can get expensive in clusters with many
indices and always being able to delay this computation to after a new
cluster state has been build and published, can reduce load on the
cluster state update thread on the elected master node.

Also two helper methods have been added to `Metadata` class to easily
and efficiently get access to aliases and the indices these aliases
refer to. Logically aliases are often seen as independent resources, but
this isn't the case and getting access to an entire alias requires
checking all `IndexMetadata` instances. This change will make this
easier, without changing how we store aliases in the cluster state.
Changing that would be much more difficult.

server/src/internalClusterTest/java/org/elasticsearch/cluster/ClusterStateDiffIT.java

@@ -570,9 +570,7 @@ public class ClusterStateDiffIT extends ESIntegTestCase {
                         }
                         break;
                     case 2:
-                        builder.settings(
-                            Settings.builder().put(part.getSettings()).put(IndexMetadata.SETTING_INDEX_UUID, UUIDs.randomBase64UUID())
-                        );
+                        builder.settings(Settings.builder().put(part.getSettings()).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0));
                         break;
                     default:
                         throw new IllegalArgumentException("Shouldn't be here");

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

@@ -10,7 +10,6 @@ package org.elasticsearch.cluster.metadata;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.common.ParsingException;
-import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.time.DateFormatters;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.Index;
@@ -26,7 +25,6 @@ import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -251,20 +249,12 @@ public interface IndexAbstraction {
             } else if (writeIndices.size() == 1) {
                 this.writeIndex = writeIndices.get(0).getIndex();
             } else {
-                List<String> writeIndicesStrings = writeIndices.stream().map(i -> i.getIndex().getName()).collect(Collectors.toList());
-                throw new IllegalStateException(
-                    "alias ["
-                        + aliasName
-                        + "] has more than one write index ["
-                        + Strings.collectionToCommaDelimitedString(writeIndicesStrings)
-                        + "]"
-                );
+                throw new IllegalStateException("write indices size can only be 0 or 1, but is [" + writeIndices.size() + "]");
             }
 
             this.isHidden = aliasMetadata.isHidden() == null ? false : aliasMetadata.isHidden();
             this.isSystem = indices.stream().allMatch(IndexMetadata::isSystem);
             dataStreamAlias = false;
-            validateAliasProperties(indices);
         }
 
         public Alias(DataStreamAlias dataStreamAlias, List<Index> indicesOfAllDataStreams, Index writeIndexOfWriteDataStream) {
@@ -321,68 +311,6 @@ public interface IndexAbstraction {
             return null;
         }
 
-        private void validateAliasProperties(List<IndexMetadata> referenceIndexMetadatas) {
-            // Validate hidden status
-            final Map<Boolean, List<IndexMetadata>> groupedByHiddenStatus = referenceIndexMetadatas.stream()
-                .collect(Collectors.groupingBy(idxMeta -> Boolean.TRUE.equals(idxMeta.getAliases().get(aliasName).isHidden())));
-            if (isNonEmpty(groupedByHiddenStatus.get(true)) && isNonEmpty(groupedByHiddenStatus.get(false))) {
-                List<String> hiddenOn = groupedByHiddenStatus.get(true)
-                    .stream()
-                    .map(idx -> idx.getIndex().getName())
-                    .collect(Collectors.toList());
-                List<String> nonHiddenOn = groupedByHiddenStatus.get(false)
-                    .stream()
-                    .map(idx -> idx.getIndex().getName())
-                    .collect(Collectors.toList());
-                throw new IllegalStateException(
-                    "alias ["
-                        + aliasName
-                        + "] has is_hidden set to true on indices ["
-                        + Strings.collectionToCommaDelimitedString(hiddenOn)
-                        + "] but does not have is_hidden set to true on indices ["
-                        + Strings.collectionToCommaDelimitedString(nonHiddenOn)
-                        + "]; alias must have the same is_hidden setting "
-                        + "on all indices"
-                );
-            }
-
-            // Validate system status
-
-            final Map<Boolean, List<IndexMetadata>> groupedBySystemStatus = referenceIndexMetadatas.stream()
-                .collect(Collectors.groupingBy(IndexMetadata::isSystem));
-            // If the alias has either all system or all non-system, then no more validation is required
-            if (isNonEmpty(groupedBySystemStatus.get(false)) && isNonEmpty(groupedBySystemStatus.get(true))) {
-                final List<String> newVersionSystemIndices = groupedBySystemStatus.get(true)
-                    .stream()
-                    .filter(i -> i.getCreationVersion().onOrAfter(IndexNameExpressionResolver.SYSTEM_INDEX_ENFORCEMENT_VERSION))
-                    .map(i -> i.getIndex().getName())
-                    .sorted() // reliable error message for testing
-                    .collect(Collectors.toList());
-
-                if (newVersionSystemIndices.isEmpty() == false) {
-                    final List<String> nonSystemIndices = groupedBySystemStatus.get(false)
-                        .stream()
-                        .map(i -> i.getIndex().getName())
-                        .sorted() // reliable error message for testing
-                        .collect(Collectors.toList());
-                    throw new IllegalStateException(
-                        "alias ["
-                            + aliasName
-                            + "] refers to both system indices "
-                            + newVersionSystemIndices
-                            + " and non-system indices: "
-                            + nonSystemIndices
-                            + ", but aliases must refer to either system or"
-                            + " non-system indices, not both"
-                    );
-                }
-            }
-        }
-
-        private boolean isNonEmpty(List<IndexMetadata> idxMetas) {
-            return (Objects.isNull(idxMetas) || idxMetas.isEmpty()) == false;
-        }
-
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;

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

@@ -203,6 +203,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
     private final Settings settings;
     private final DiffableStringMap hashesOfConsistentSettings;
     private final ImmutableOpenMap<String, IndexMetadata> indices;
+    private final ImmutableOpenMap<String, Set<Index>> aliasedIndices;
     private final ImmutableOpenMap<String, IndexTemplateMetadata> templates;
     private final ImmutableOpenMap<String, Custom> customs;
 
@@ -233,6 +234,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         int totalNumberOfShards,
         int totalOpenIndexShards,
         ImmutableOpenMap<String, IndexMetadata> indices,
+        ImmutableOpenMap<String, Set<Index>> aliasedIndices,
         ImmutableOpenMap<String, IndexTemplateMetadata> templates,
         ImmutableOpenMap<String, Custom> customs,
         String[] allIndices,
@@ -254,6 +256,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         this.settings = settings;
         this.hashesOfConsistentSettings = hashesOfConsistentSettings;
         this.indices = indices;
+        this.aliasedIndices = aliasedIndices;
         this.customs = customs;
         this.templates = templates;
         this.totalNumberOfShards = totalNumberOfShards;
@@ -282,6 +285,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             totalNumberOfShards,
             totalOpenIndexShards,
             indices,
+            aliasedIndices,
             templates,
             customs,
             allIndices,
@@ -339,15 +343,6 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         return this.oldestIndexVersion;
     }
 
-    public boolean hasAlias(String alias) {
-        IndexAbstraction indexAbstraction = getIndicesLookup().get(alias);
-        if (indexAbstraction != null) {
-            return indexAbstraction.getType() == IndexAbstraction.Type.ALIAS;
-        } else {
-            return false;
-        }
-    }
-
     public boolean equalsAliases(Metadata other) {
         for (IndexMetadata otherIndex : other.indices().values()) {
             IndexMetadata thisIndex = index(otherIndex.getIndex());
@@ -383,6 +378,10 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         return indicesLookup;
     }
 
+    public boolean sameIndicesLookup(Metadata other) {
+        return this.indicesLookup == other.indicesLookup;
+    }
+
     /**
      * Finds the specific index aliases that point to the requested concrete indices directly
      * or that match with the indices via wildcards.
@@ -787,6 +786,36 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         return indices();
     }
 
+    /**
+     * Returns whether an alias exists with provided alias name.
+     *
+     * @param aliasName The provided alias name
+     * @return whether an alias exists with provided alias name
+     */
+    public boolean hasAlias(String aliasName) {
+        return aliasedIndices.containsKey(aliasName) || dataStreamAliases().containsKey(aliasName);
+    }
+
+    /**
+     * Returns all the indices that the alias with the provided alias name refers to.
+     * These are aliased indices. Not that, this only return indices that have been aliased
+     * and not indices that are behind a data stream or data stream alias.
+     *
+     * @param aliasName The provided alias name
+     * @return all aliased indices by the alias with the provided alias name
+     */
+    public Set<Index> aliasedIndices(String aliasName) {
+        Objects.requireNonNull(aliasName);
+        return aliasedIndices.getOrDefault(aliasName, Set.of());
+    }
+
+    /**
+     * @return the names of all indices aliases.
+     */
+    public Set<String> aliasedIndices() {
+        return aliasedIndices.keySet();
+    }
+
     public ImmutableOpenMap<String, IndexTemplateMetadata> templates() {
         return this.templates;
     }
@@ -1136,6 +1165,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         private DiffableStringMap hashesOfConsistentSettings = DiffableStringMap.EMPTY;
 
         private final ImmutableOpenMap.Builder<String, IndexMetadata> indices;
+        private final ImmutableOpenMap.Builder<String, Set<Index>> aliasedIndices;
         private final ImmutableOpenMap.Builder<String, IndexTemplateMetadata> templates;
         private final ImmutableOpenMap.Builder<String, Custom> customs;
 
@@ -1155,15 +1185,17 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             this.hashesOfConsistentSettings = metadata.hashesOfConsistentSettings;
             this.version = metadata.version;
             this.indices = ImmutableOpenMap.builder(metadata.indices);
+            this.aliasedIndices = ImmutableOpenMap.builder(metadata.aliasedIndices);
             this.templates = ImmutableOpenMap.builder(metadata.templates);
             this.customs = ImmutableOpenMap.builder(metadata.customs);
-            previousIndicesLookup = metadata.getIndicesLookup();
+            this.previousIndicesLookup = metadata.indicesLookup;
             this.mappingsByHash = new HashMap<>(metadata.mappingsByHash);
         }
 
         private Builder(Map<String, MappingMetadata> mappingsByHash) {
             clusterUUID = UNKNOWN_CLUSTER_UUID;
             indices = ImmutableOpenMap.builder();
+            aliasedIndices = ImmutableOpenMap.builder();
             templates = ImmutableOpenMap.builder();
             customs = ImmutableOpenMap.builder();
             indexGraveyard(IndexGraveyard.builder().build()); // create new empty index graveyard to initialize
@@ -1177,6 +1209,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             dedupeMapping(indexMetadataBuilder);
             IndexMetadata indexMetadata = indexMetadataBuilder.build();
             IndexMetadata previous = indices.put(indexMetadata.getIndex().getName(), indexMetadata);
+            updateAliases(previous, indexMetadata);
             if (unsetPreviousIndicesLookup(previous, indexMetadata)) {
                 previousIndicesLookup = null;
             }
@@ -1193,6 +1226,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
                 indexMetadata = IndexMetadata.builder(indexMetadata).version(indexMetadata.getVersion() + 1).build();
             }
             IndexMetadata previous = indices.put(indexMetadata.getIndex().getName(), indexMetadata);
+            updateAliases(previous, indexMetadata);
             if (unsetPreviousIndicesLookup(previous, indexMetadata)) {
                 previousIndicesLookup = null;
             }
@@ -1246,7 +1280,8 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         public Builder remove(String index) {
             previousIndicesLookup = null;
 
-            indices.remove(index);
+            IndexMetadata previous = indices.remove(index);
+            updateAliases(previous, null);
             return this;
         }
 
@@ -1255,6 +1290,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
 
             indices.clear();
             mappingsByHash.clear();
+            aliasedIndices.clear();
             return this;
         }
 
@@ -1267,6 +1303,67 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             return this;
         }
 
+        void updateAliases(IndexMetadata previous, IndexMetadata current) {
+            if (previous == null && current != null) {
+                for (var cursor : current.getAliases()) {
+                    putAlias(cursor.key, current.getIndex());
+                }
+            } else if (previous != null && current == null) {
+                for (var cursor : previous.getAliases()) {
+                    removeAlias(cursor.key, previous.getIndex());
+                }
+            } else if (previous != null && current != null) {
+                if (Objects.equals(previous.getAliases(), current.getAliases())) {
+                    return;
+                }
+
+                for (var currentCursor : current.getAliases()) {
+                    if (previous.getAliases().containsKey(currentCursor.key) == false) {
+                        putAlias(currentCursor.key, current.getIndex());
+                    }
+                }
+                for (var previousCursor : previous.getAliases()) {
+                    if (current.getAliases().containsKey(previousCursor.key) == false) {
+                        removeAlias(previousCursor.key, current.getIndex());
+                    }
+                }
+            }
+        }
+
+        private Builder putAlias(String alias, Index index) {
+            Objects.requireNonNull(alias);
+            Objects.requireNonNull(index);
+
+            Set<Index> indices = new HashSet<>(aliasedIndices.getOrDefault(alias, Set.of()));
+            if (indices.add(index) == false) {
+                return this; // indices already contained this index
+            }
+            aliasedIndices.put(alias, Collections.unmodifiableSet(indices));
+            return this;
+        }
+
+        private Builder removeAlias(String alias, Index index) {
+            Objects.requireNonNull(alias);
+            Objects.requireNonNull(index);
+
+            Set<Index> indices = aliasedIndices.get(alias);
+            if (indices == null || indices.isEmpty()) {
+                throw new IllegalStateException("Cannot remove non-existent alias [" + alias + "] for index [" + index.getName() + "]");
+            }
+
+            indices = new HashSet<>(indices);
+            if (indices.remove(index) == false) {
+                throw new IllegalStateException("Cannot remove non-existent alias [" + alias + "] for index [" + index.getName() + "]");
+            }
+
+            if (indices.isEmpty()) {
+                aliasedIndices.remove(alias); // for consistency, we don't store empty sets, so null it out
+            } else {
+                aliasedIndices.put(alias, Collections.unmodifiableSet(indices));
+            }
+            return this;
+        }
+
         public Builder put(IndexTemplateMetadata.Builder template) {
             return put(template.build());
         }
@@ -1618,18 +1715,6 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
          * @return a new <code>Metadata</code> instance
          */
         public Metadata build() {
-            return build(true);
-        }
-
-        /**
-         * @param builtIndicesLookupEagerly Controls whether indices lookup should be build as part of the execution of this method
-         *                                  or after when needed. Almost all of the time indices lookup should be built eagerly, however
-         *                                  in certain cases when <code>Metdata</code> instances are build that are not published and
-         *                                  many indices have been defined then it makes sense to skip building indices lookup.
-         *
-         * @return a new <code>Metadata</code> instance
-         */
-        public Metadata build(boolean builtIndicesLookupEagerly) {
             // TODO: We should move these datastructures to IndexNameExpressionResolver, this will give the following benefits:
             // 1) The datastructures will be rebuilt only when needed. Now during serializing we rebuild these datastructures
             // while these datastructures aren't even used.
@@ -1639,7 +1724,6 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             final List<String> visibleOpenIndices = new ArrayList<>();
             final List<String> allClosedIndices = new ArrayList<>();
             final List<String> visibleClosedIndices = new ArrayList<>();
-            final Set<String> indicesAliases = new HashSet<>();
             final ImmutableOpenMap<String, IndexMetadata> indicesMap = indices.build();
             final Set<String> allIndices = indicesMap.keySet();
 
@@ -1662,23 +1746,24 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
                         visibleClosedIndices.add(name);
                     }
                 }
-                indexMetadata.getAliases().keysIt().forEachRemaining(indicesAliases::add);
                 oldestIndexVersionId = Math.min(oldestIndexVersionId, indexMetadata.getCompatibilityVersion().id);
             }
 
+            var aliasedIndices = this.aliasedIndices.build();
+            for (var cursor : aliasedIndices) {
+                List<IndexMetadata> aliasIndices = cursor.value.stream()
+                    .map(idx -> indicesMap.get(idx.getName()))
+                    .collect(Collectors.toList());
+                validateAlias(cursor.key, aliasIndices);
+            }
             final DataStreamMetadata dataStreamMetadata = (DataStreamMetadata) this.customs.get(DataStreamMetadata.TYPE);
-            ensureNoNameCollisions(indicesAliases, indicesMap, allIndices, dataStreamMetadata);
+            ensureNoNameCollisions(aliasedIndices.keySet(), indicesMap, allIndices, dataStreamMetadata);
+            assert assertDataStreams(indicesMap, dataStreamMetadata);
 
-            SortedMap<String, IndexAbstraction> indicesLookup;
+            SortedMap<String, IndexAbstraction> indicesLookup = null;
             if (previousIndicesLookup != null) {
                 assert previousIndicesLookup.equals(buildIndicesLookup(dataStreamMetadata, indicesMap));
                 indicesLookup = previousIndicesLookup;
-            } else {
-                if (builtIndicesLookupEagerly) {
-                    indicesLookup = buildIndicesLookup(dataStreamMetadata, indicesMap);
-                } else {
-                    indicesLookup = null;
-                }
             }
 
             purgeUnusedEntries(indicesMap);
@@ -1715,6 +1800,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
                 totalNumberOfShards,
                 totalOpenIndexShards,
                 indicesMap,
+                aliasedIndices,
                 templates.build(),
                 customs.build(),
                 allIndicesArray,
@@ -1904,32 +1990,108 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
                 assert existing == null : "duplicate for " + entry.getKey();
             }
 
-            validateDataStreams(indicesLookup, dataStreamMetadata);
             return Collections.unmodifiableSortedMap(indicesLookup);
         }
 
-        static void validateDataStreams(SortedMap<String, IndexAbstraction> indicesLookup, @Nullable DataStreamMetadata dsMetadata) {
+        private static boolean isNonEmpty(List<IndexMetadata> idxMetas) {
+            return (Objects.isNull(idxMetas) || idxMetas.isEmpty()) == false;
+        }
+
+        private static void validateAlias(String aliasName, List<IndexMetadata> indexMetadatas) {
+            // Validate write indices
+            List<String> writeIndices = indexMetadatas.stream()
+                .filter(idxMeta -> Boolean.TRUE.equals(idxMeta.getAliases().get(aliasName).writeIndex()))
+                .map(im -> im.getIndex().getName())
+                .collect(Collectors.toList());
+            if (writeIndices.size() > 1) {
+                throw new IllegalStateException(
+                    "alias ["
+                        + aliasName
+                        + "] has more than one write index ["
+                        + Strings.collectionToCommaDelimitedString(writeIndices)
+                        + "]"
+                );
+            }
+
+            // Validate hidden status
+            final Map<Boolean, List<IndexMetadata>> groupedByHiddenStatus = indexMetadatas.stream()
+                .collect(Collectors.groupingBy(idxMeta -> Boolean.TRUE.equals(idxMeta.getAliases().get(aliasName).isHidden())));
+            if (isNonEmpty(groupedByHiddenStatus.get(true)) && isNonEmpty(groupedByHiddenStatus.get(false))) {
+                List<String> hiddenOn = groupedByHiddenStatus.get(true)
+                    .stream()
+                    .map(idx -> idx.getIndex().getName())
+                    .collect(Collectors.toList());
+                List<String> nonHiddenOn = groupedByHiddenStatus.get(false)
+                    .stream()
+                    .map(idx -> idx.getIndex().getName())
+                    .collect(Collectors.toList());
+                throw new IllegalStateException(
+                    "alias ["
+                        + aliasName
+                        + "] has is_hidden set to true on indices ["
+                        + Strings.collectionToCommaDelimitedString(hiddenOn)
+                        + "] but does not have is_hidden set to true on indices ["
+                        + Strings.collectionToCommaDelimitedString(nonHiddenOn)
+                        + "]; alias must have the same is_hidden setting "
+                        + "on all indices"
+                );
+            }
+
+            // Validate system status
+            final Map<Boolean, List<IndexMetadata>> groupedBySystemStatus = indexMetadatas.stream()
+                .collect(Collectors.groupingBy(IndexMetadata::isSystem));
+            // If the alias has either all system or all non-system, then no more validation is required
+            if (isNonEmpty(groupedBySystemStatus.get(false)) && isNonEmpty(groupedBySystemStatus.get(true))) {
+                final List<String> newVersionSystemIndices = groupedBySystemStatus.get(true)
+                    .stream()
+                    .filter(i -> i.getCreationVersion().onOrAfter(IndexNameExpressionResolver.SYSTEM_INDEX_ENFORCEMENT_VERSION))
+                    .map(i -> i.getIndex().getName())
+                    .sorted() // reliable error message for testing
+                    .collect(Collectors.toList());
+
+                if (newVersionSystemIndices.isEmpty() == false) {
+                    final List<String> nonSystemIndices = groupedBySystemStatus.get(false)
+                        .stream()
+                        .map(i -> i.getIndex().getName())
+                        .sorted() // reliable error message for testing
+                        .collect(Collectors.toList());
+                    throw new IllegalStateException(
+                        "alias ["
+                            + aliasName
+                            + "] refers to both system indices "
+                            + newVersionSystemIndices
+                            + " and non-system indices: "
+                            + nonSystemIndices
+                            + ", but aliases must refer to either system or"
+                            + " non-system indices, not both"
+                    );
+                }
+            }
+        }
+
+        static boolean assertDataStreams(ImmutableOpenMap<String, IndexMetadata> indices, @Nullable DataStreamMetadata dsMetadata) {
             if (dsMetadata != null) {
                 // Sanity check, because elsewhere a more user friendly error should have occurred:
-                List<String> conflictingAliases = indicesLookup.values()
-                    .stream()
-                    .filter(ia -> ia.getType() == IndexAbstraction.Type.ALIAS)
-                    .filter(ia -> ia.isDataStreamRelated() == false)
-                    .filter(ia -> {
-                        for (Index index : ia.getIndices()) {
-                            if (indicesLookup.get(index.getName()).getParentDataStream() != null) {
-                                return true;
+                List<String> conflictingAliases = null;
+
+                for (var dataStream : dsMetadata.dataStreams().values()) {
+                    for (var index : dataStream.getIndices()) {
+                        IndexMetadata im = indices.get(index.getName());
+                        if (im != null && im.getAliases().isEmpty() == false) {
+                            for (var alias : im.getAliases().values()) {
+                                if (conflictingAliases == null) {
+                                    conflictingAliases = new LinkedList<>();
+                                }
+                                conflictingAliases.add(alias.alias());
                             }
                         }
-
-                        return false;
-                    })
-                    .map(IndexAbstraction::getName)
-                    .collect(Collectors.toList());
-                if (conflictingAliases.isEmpty() == false) {
-                    throw new IllegalStateException("aliases " + conflictingAliases + " cannot refer to backing indices of data streams");
+                    }
+                }
+                if (conflictingAliases != null) {
+                    throw new AssertionError("aliases " + conflictingAliases + " cannot refer to backing indices of data streams");
                 }
             }
+            return true;
         }
 
         public static void toXContent(Metadata metadata, XContentBuilder builder, ToXContent.Params params) throws IOException {
@@ -2128,5 +2290,4 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             return Builder.fromXContent(parser);
         }
     };
-
 }

server/src/main/java/org/elasticsearch/cluster/routing/allocation/IndexMetadataUpdater.java

@@ -131,7 +131,7 @@ public class IndexMetadataUpdater extends RoutingChangesObserver.AbstractRouting
 
         if (metadataBuilder != null) {
             Metadata newMetadata = metadataBuilder.build();
-            assert oldMetadata.getIndicesLookup() == newMetadata.getIndicesLookup();
+            assert oldMetadata.sameIndicesLookup(newMetadata);
             return newMetadata;
         } else {
             return oldMetadata;

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -446,7 +446,6 @@ public class MasterService extends AbstractLifecycleComponent {
 
         if (previousClusterState != newClusterState) {
             // only the master controls the version numbers
-            final var previousIndicesLookup = newClusterState.metadata().getIndicesLookup();
             Builder builder = incrementVersion(newClusterState);
             if (previousClusterState.routingTable() != newClusterState.routingTable()) {
                 builder.routingTable(newClusterState.routingTable().withIncrementedVersion());
@@ -455,8 +454,9 @@ public class MasterService extends AbstractLifecycleComponent {
                 builder.metadata(newClusterState.metadata().withIncrementedVersion());
             }
 
+            final var previousMetadata = newClusterState.metadata();
             newClusterState = builder.build();
-            assert previousIndicesLookup == newClusterState.metadata().getIndicesLookup();
+            assert previousMetadata.sameIndicesLookup(newClusterState.metadata());
         }
 
         return newClusterState;

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexServiceTests.java

@@ -95,6 +95,7 @@ import static org.elasticsearch.cluster.metadata.MetadataCreateIndexService.pars
 import static org.elasticsearch.cluster.metadata.MetadataCreateIndexService.resolveAndValidateAliases;
 import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
 import static org.elasticsearch.indices.ShardLimitValidatorTests.createTestShardLimitService;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasKey;
@@ -1019,6 +1020,14 @@ public class MetadataCreateIndexServiceTests extends ESTestCase {
         assertThat(updatedClusterState.blocks().getIndexBlockWithId("test", INDEX_READ_ONLY_BLOCK.id()), is(INDEX_READ_ONLY_BLOCK));
         assertThat(updatedClusterState.routingTable().index("test"), is(notNullValue()));
         assertThat(allocationRerouted.get(), is(true));
+
+        Metadata metadata = updatedClusterState.metadata();
+        IndexAbstraction alias = metadata.getIndicesLookup().get("alias1");
+        assertNotNull(alias);
+        assertThat(alias.getType(), equalTo(IndexAbstraction.Type.ALIAS));
+        Index index = metadata.index("test").getIndex();
+        assertThat(alias.getIndices(), contains(index));
+        assertThat(metadata.aliasedIndices("alias1"), contains(index));
     }
 
     public void testClusterStateCreateIndexWithMetadataTransaction() {

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java

@@ -37,6 +37,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
@@ -143,6 +144,7 @@ public class MetadataDeleteIndexServiceTests extends ESTestCase {
         assertNull(after.routingTable().index(index));
         assertNull(after.blocks().indices().get(index));
         assertNull(after.metadata().getIndicesLookup().get(alias));
+        assertThat(after.metadata().aliasedIndices(alias), empty());
     }
 
     public void testDeleteBackingIndexForDataStream() {

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexAliasesServiceTests.java

@@ -33,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstanc
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
@@ -77,6 +78,7 @@ public class MetadataIndexAliasesServiceTests extends ESTestCase {
         assertThat(alias.getType(), equalTo(IndexAbstraction.Type.ALIAS));
         assertThat(alias.getIndices(), contains(after.metadata().index(index).getIndex()));
         assertAliasesVersionIncreased(index, before, after);
+        assertThat(after.metadata().aliasedIndices("test"), contains(after.metadata().index(index).getIndex()));
 
         // Remove the alias from it while adding another one
         before = after;
@@ -85,17 +87,21 @@ public class MetadataIndexAliasesServiceTests extends ESTestCase {
             Arrays.asList(new AliasAction.Remove(index, "test", null), new AliasAction.Add(index, "test_2", null, null, null, null, null))
         );
         assertNull(after.metadata().getIndicesLookup().get("test"));
+        assertThat(after.metadata().aliasedIndices("test"), empty());
         alias = after.metadata().getIndicesLookup().get("test_2");
         assertNotNull(alias);
         assertThat(alias.getType(), equalTo(IndexAbstraction.Type.ALIAS));
         assertThat(alias.getIndices(), contains(after.metadata().index(index).getIndex()));
         assertAliasesVersionIncreased(index, before, after);
+        assertThat(after.metadata().aliasedIndices("test_2"), contains(after.metadata().index(index).getIndex()));
 
         // Now just remove on its own
         before = after;
         after = service.applyAliasActions(before, singletonList(new AliasAction.Remove(index, "test_2", randomBoolean())));
         assertNull(after.metadata().getIndicesLookup().get("test"));
+        assertThat(after.metadata().aliasedIndices("test"), empty());
         assertNull(after.metadata().getIndicesLookup().get("test_2"));
+        assertThat(after.metadata().aliasedIndices("test_2"), empty());
         assertAliasesVersionIncreased(index, before, after);
     }
 

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataTests.java

@@ -12,6 +12,7 @@ import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
 import org.elasticsearch.cluster.ClusterModule;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.coordination.CoordinationMetadata;
 import org.elasticsearch.cluster.coordination.CoordinationMetadata.VotingConfigExclusion;
@@ -52,15 +53,15 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.SortedMap;
-import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createBackingIndex;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createFirstBackingIndex;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createTimestampField;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
-import static org.elasticsearch.cluster.metadata.Metadata.Builder.validateDataStreams;
+import static org.elasticsearch.cluster.metadata.Metadata.Builder.assertDataStreams;
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -1178,7 +1179,7 @@ public class MetadataTests extends ESTestCase {
             .put(idx, false)
             .put(newInstance(dataStreamName, createTimestampField("@timestamp"), List.of(idx.getIndex())));
 
-        IllegalStateException e = expectThrows(IllegalStateException.class, b::build);
+        AssertionError e = expectThrows(AssertionError.class, b::build);
         assertThat(e.getMessage(), containsString("aliases [" + conflictingName + "] cannot refer to backing indices of data streams"));
     }
 
@@ -1399,7 +1400,7 @@ public class MetadataTests extends ESTestCase {
     public void testValidateDataStreamsNoConflicts() {
         Metadata metadata = createIndices(5, 10, "foo-datastream").metadata;
         // don't expect any exception when validating a system without indices that would conflict with future backing indices
-        validateDataStreams(metadata.getIndicesLookup(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
+        assertDataStreams(metadata.getIndices(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
     }
 
     public void testValidateDataStreamsIgnoresIndicesWithoutCounter() {
@@ -1423,7 +1424,7 @@ public class MetadataTests extends ESTestCase {
             .build();
         // don't expect any exception when validating against non-backing indices that don't conform to the backing indices naming
         // convention
-        validateDataStreams(metadata.getIndicesLookup(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
+        assertDataStreams(metadata.getIndices(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
     }
 
     public void testValidateDataStreamsAllowsNamesThatStartsWithPrefix() {
@@ -1437,51 +1438,7 @@ public class MetadataTests extends ESTestCase {
             .build();
         // don't expect any exception when validating against (potentially backing) indices that can't create conflict because of
         // additional text before number
-        validateDataStreams(metadata.getIndicesLookup(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
-    }
-
-    public void testValidateDataStreamsAllowsPrefixedBackingIndices() {
-        String dataStreamName = "foo-datastream";
-        int generations = 10;
-        List<IndexMetadata> backingIndices = new ArrayList<>(generations);
-        for (int i = 1; i <= generations; i++) {
-            IndexMetadata idx;
-            if (i % 2 == 0 && i < generations) {
-                idx = IndexMetadata.builder("shrink-" + DataStream.getDefaultBackingIndexName(dataStreamName, i))
-                    .settings(ESTestCase.settings(Version.CURRENT).put("index.hidden", true))
-                    .numberOfShards(1)
-                    .numberOfReplicas(1)
-                    .build();
-            } else {
-                idx = createBackingIndex(dataStreamName, i).build();
-            }
-            backingIndices.add(idx);
-        }
-        DataStream dataStream = newInstance(
-            dataStreamName,
-            createTimestampField("@timestamp"),
-            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
-        );
-
-        IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, List.of());
-        // manually building the indices lookup as going through Metadata.Builder#build would trigger the validate method already
-        SortedMap<String, IndexAbstraction> indicesLookup = new TreeMap<>();
-        for (IndexMetadata indexMeta : backingIndices) {
-            indicesLookup.put(indexMeta.getIndex().getName(), new IndexAbstraction.ConcreteIndex(indexMeta, dataStreamAbstraction));
-        }
-
-        for (int i = 1; i <= generations; i++) {
-            // for the indices that we added in the data stream with a "shrink-" prefix, add the non-prefixed indices to the lookup
-            if (i % 2 == 0 && i < generations) {
-                IndexMetadata indexMeta = createBackingIndex(dataStreamName, i).build();
-                indicesLookup.put(indexMeta.getIndex().getName(), new IndexAbstraction.ConcreteIndex(indexMeta, dataStreamAbstraction));
-            }
-        }
-        DataStreamMetadata dataStreamMetadata = new DataStreamMetadata(Map.of(dataStreamName, dataStream), Map.of());
-
-        // prefixed indices with a lower generation than the data stream's generation are allowed even if the non-prefixed, matching the
-        // data stream backing indices naming pattern, indices are already in the system
-        validateDataStreams(indicesLookup, dataStreamMetadata);
+        assertDataStreams(metadata.getIndices(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
     }
 
     public void testValidateDataStreamsForNullDataStreamMetadata() {
@@ -1490,7 +1447,7 @@ public class MetadataTests extends ESTestCase {
             .build();
 
         try {
-            validateDataStreams(metadata.getIndicesLookup(), null);
+            assertDataStreams(metadata.getIndices(), null);
         } catch (Exception e) {
             fail("did not expect exception when validating a system without any data streams but got " + e.getMessage());
         }
@@ -1895,6 +1852,114 @@ public class MetadataTests extends ESTestCase {
         }
     }
 
+    public void testAliasedIndices() {
+        int numAliases = randomIntBetween(32, 64);
+        int numIndicesPerAlias = randomIntBetween(8, 16);
+
+        Metadata.Builder builder = Metadata.builder();
+        for (int i = 0; i < numAliases; i++) {
+            String aliasName = "alias-" + i;
+            for (int j = 0; j < numIndicesPerAlias; j++) {
+                AliasMetadata.Builder alias = new AliasMetadata.Builder(aliasName);
+                if (j == 0) {
+                    alias.writeIndex(true);
+                }
+
+                String indexName = aliasName + "-" + j;
+                builder.put(
+                    IndexMetadata.builder(indexName)
+                        .settings(settings(Version.CURRENT))
+                        .creationDate(randomNonNegativeLong())
+                        .numberOfShards(1)
+                        .numberOfReplicas(0)
+                        .putAlias(alias)
+                );
+            }
+        }
+
+        Metadata metadata = builder.build();
+        for (int i = 0; i < numAliases; i++) {
+            String aliasName = "alias-" + i;
+            Set<Index> result = metadata.aliasedIndices(aliasName);
+            Index[] expected = IntStream.range(0, numIndicesPerAlias)
+                .mapToObj(j -> aliasName + "-" + j)
+                .map(name -> new Index(name, ClusterState.UNKNOWN_UUID))
+                .toArray(Index[]::new);
+            assertThat(result, containsInAnyOrder(expected));
+        }
+
+        // Add a new alias and index
+        builder = Metadata.builder(metadata);
+        String newAliasName = "alias-new";
+        {
+            builder.put(
+                IndexMetadata.builder(newAliasName + "-1")
+                    .settings(settings(Version.CURRENT))
+                    .creationDate(randomNonNegativeLong())
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder(newAliasName).writeIndex(true))
+            );
+        }
+        metadata = builder.build();
+        assertThat(metadata.aliasedIndices(), hasSize(numAliases + 1));
+        assertThat(metadata.aliasedIndices(newAliasName), contains(new Index(newAliasName + "-1", ClusterState.UNKNOWN_UUID)));
+
+        // Remove the new alias/index
+        builder = Metadata.builder(metadata);
+        {
+            builder.remove(newAliasName + "-1");
+        }
+        metadata = builder.build();
+        assertThat(metadata.aliasedIndices(), hasSize(numAliases));
+        assertThat(metadata.aliasedIndices(newAliasName), empty());
+
+        // Add a new alias that points to existing indices
+        builder = Metadata.builder(metadata);
+        {
+            IndexMetadata.Builder imBuilder = new IndexMetadata.Builder(metadata.index("alias-1-0"));
+            imBuilder.putAlias(new AliasMetadata.Builder(newAliasName));
+            builder.put(imBuilder);
+
+            imBuilder = new IndexMetadata.Builder(metadata.index("alias-2-1"));
+            imBuilder.putAlias(new AliasMetadata.Builder(newAliasName));
+            builder.put(imBuilder);
+
+            imBuilder = new IndexMetadata.Builder(metadata.index("alias-3-2"));
+            imBuilder.putAlias(new AliasMetadata.Builder(newAliasName));
+            builder.put(imBuilder);
+        }
+        metadata = builder.build();
+        assertThat(metadata.aliasedIndices(), hasSize(numAliases + 1));
+        assertThat(
+            metadata.aliasedIndices(newAliasName),
+            containsInAnyOrder(
+                new Index("alias-1-0", ClusterState.UNKNOWN_UUID),
+                new Index("alias-2-1", ClusterState.UNKNOWN_UUID),
+                new Index("alias-3-2", ClusterState.UNKNOWN_UUID)
+            )
+        );
+
+        // Remove the new alias that points to existing indices
+        builder = Metadata.builder(metadata);
+        {
+            IndexMetadata.Builder imBuilder = new IndexMetadata.Builder(metadata.index("alias-1-0"));
+            imBuilder.removeAlias(newAliasName);
+            builder.put(imBuilder);
+
+            imBuilder = new IndexMetadata.Builder(metadata.index("alias-2-1"));
+            imBuilder.removeAlias(newAliasName);
+            builder.put(imBuilder);
+
+            imBuilder = new IndexMetadata.Builder(metadata.index("alias-3-2"));
+            imBuilder.removeAlias(newAliasName);
+            builder.put(imBuilder);
+        }
+        metadata = builder.build();
+        assertThat(metadata.aliasedIndices(), hasSize(numAliases));
+        assertThat(metadata.aliasedIndices(newAliasName), empty());
+    }
+
     public static final String SYSTEM_ALIAS_NAME = "system_alias";
 
     public void testHiddenAliasValidation() {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/CopyExecutionStateStep.java

@@ -97,7 +97,7 @@ public class CopyExecutionStateStep extends ClusterStateActionStep {
         Metadata.Builder newMetadata = Metadata.builder(clusterState.getMetadata())
             .put(IndexMetadata.builder(targetIndexMetadata).putCustom(ILM_CUSTOM_METADATA_KEY, relevantTargetCustomData.build().asMap()));
 
-        return ClusterState.builder(clusterState).metadata(newMetadata.build(false)).build();
+        return ClusterState.builder(clusterState).metadata(newMetadata).build();
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/CopySettingsStep.java

@@ -92,7 +92,7 @@ public class CopySettingsStep extends ClusterStateActionStep {
             .put(
                 IndexMetadata.builder(targetIndexMetadata).settingsVersion(targetIndexMetadata.getSettingsVersion() + 1).settings(settings)
             );
-        return ClusterState.builder(clusterState).metadata(newMetaData.build(false)).build();
+        return ClusterState.builder(clusterState).metadata(newMetaData).build();
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/GenerateSnapshotNameStep.java

@@ -103,7 +103,6 @@ public class GenerateSnapshotNameStep extends ClusterStateActionStep {
             .metadata(
                 Metadata.builder(clusterState.getMetadata())
                     .put(IndexMetadata.builder(indexMetadata).putCustom(ILM_CUSTOM_METADATA_KEY, newCustomData.build().asMap()))
-                    .build(false)
             )
             .build();
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/GenerateUniqueIndexNameStep.java

@@ -98,7 +98,7 @@ public class GenerateUniqueIndexNameStep extends ClusterStateActionStep {
 
         IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(indexMetadata);
         indexMetadataBuilder.putCustom(ILM_CUSTOM_METADATA_KEY, newCustomData.build().asMap());
-        newClusterStateBuilder.metadata(Metadata.builder(clusterState.getMetadata()).put(indexMetadataBuilder).build(false));
+        newClusterStateBuilder.metadata(Metadata.builder(clusterState.getMetadata()).put(indexMetadataBuilder));
         return newClusterStateBuilder.build();
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/InitializePolicyContextStep.java

@@ -69,7 +69,7 @@ public final class InitializePolicyContextStep extends ClusterStateActionStep {
         newCustomData.setIndexCreationDate(indexMetadata.getCreationDate());
         indexMetadataBuilder.putCustom(ILM_CUSTOM_METADATA_KEY, newCustomData.build().asMap());
 
-        newClusterStateBuilder.metadata(Metadata.builder(clusterState.getMetadata()).put(indexMetadataBuilder).build(false));
+        newClusterStateBuilder.metadata(Metadata.builder(clusterState.getMetadata()).put(indexMetadataBuilder));
         return newClusterStateBuilder.build();
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/PhaseCacheManagement.java

@@ -54,7 +54,7 @@ public final class PhaseCacheManagement {
         final IndexMetadata idxMeta = state.metadata().index(index);
         Metadata.Builder metadataBuilder = Metadata.builder(state.metadata());
         refreshPhaseDefinition(metadataBuilder, idxMeta, updatedPolicy);
-        return ClusterState.builder(state).metadata(metadataBuilder.build(false)).build();
+        return ClusterState.builder(state).metadata(metadataBuilder.build()).build();
     }
 
     /**
@@ -122,7 +122,7 @@ public final class PhaseCacheManagement {
     ) {
         Metadata.Builder mb = Metadata.builder(state.metadata());
         if (updateIndicesForPolicy(mb, state, xContentRegistry, client, oldPolicy, newPolicy, licenseState)) {
-            return ClusterState.builder(state).metadata(mb.build(false)).build();
+            return ClusterState.builder(state).metadata(mb).build();
         }
         return state;
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ReplaceDataStreamBackingIndexStep.java

@@ -115,7 +115,7 @@ public class ReplaceDataStreamBackingIndexStep extends ClusterStateActionStep {
 
         Metadata.Builder newMetaData = Metadata.builder(clusterState.getMetadata())
             .put(dataStream.getDataStream().replaceBackingIndex(index, targetIndexMetadata.getIndex()));
-        return ClusterState.builder(clusterState).metadata(newMetaData.build(false)).build();
+        return ClusterState.builder(clusterState).metadata(newMetaData).build();
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/UpdateRolloverLifecycleDateStep.java

@@ -74,9 +74,7 @@ public class UpdateRolloverLifecycleDateStep extends ClusterStateActionStep {
 
         IndexMetadata.Builder newIndexMetadata = IndexMetadata.builder(indexMetadata);
         newIndexMetadata.putCustom(ILM_CUSTOM_METADATA_KEY, newLifecycleState.build().asMap());
-        return ClusterState.builder(currentState)
-            .metadata(Metadata.builder(currentState.metadata()).put(newIndexMetadata).build(false))
-            .build();
+        return ClusterState.builder(currentState).metadata(Metadata.builder(currentState.metadata()).put(newIndexMetadata)).build();
     }
 
     private static String getRolloverTarget(Index index, ClusterState currentState) {

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunner.java

@@ -66,8 +66,6 @@ class IndexLifecycleRunner {
                 builder.failure(task, e);
             }
         }
-        // Trigger indices lookup creation and related validation
-        state.metadata().getIndicesLookup();
         return builder.build(state);
     };
 

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleTransition.java

@@ -430,7 +430,6 @@ public final class IndexLifecycleTransition {
                     IndexMetadata.builder(clusterState.getMetadata().index(index))
                         .putCustom(ILM_CUSTOM_METADATA_KEY, lifecycleState.asMap())
                 )
-                .build(false)
         );
         return newClusterStateBuilder;
     }