Allow more legit cases in Metadata.Builder.validateDataStreams (#65791)

This change simplifies logic and allow more legit cases in Metadata.Builder.validateDataStreams.
It will only show conflict on names that are in form of .ds-<data stream name>-<[0-9]+> and will allow any names like .ds-<data stream name>-something-else-<[0-9]+>.
This fixes problem with rollover when you have 2 data streams with names like a and a-b - currently if a-b has generation greater than a you won't be able to rollover a anymore.

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

@@ -77,6 +77,7 @@ import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -89,6 +90,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
 
     public static final String ALL = "_all";
     public static final String UNKNOWN_CLUSTER_UUID = "_na_";
+    public static final Pattern NUMBER_PATTERN = Pattern.compile("[0-9]+$");
 
     public enum XContentContext {
         /* Custom metadata should be returns as part of API call */
@@ -1481,29 +1483,18 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
         static void validateDataStreams(SortedMap<String, IndexAbstraction> indicesLookup, @Nullable DataStreamMetadata dsMetadata) {
             if (dsMetadata != null) {
                 for (DataStream ds : dsMetadata.dataStreams().values()) {
-                    Map<String, IndexAbstraction> conflicts =
-                        indicesLookup.subMap(DataStream.BACKING_INDEX_PREFIX + ds.getName() + "-",
-                            DataStream.BACKING_INDEX_PREFIX + ds.getName() + ".") // '.' is the char after '-'
-                            .entrySet().stream()
-                            .filter(entry -> {
-                                if (entry.getValue().getType() != IndexAbstraction.Type.CONCRETE_INDEX) {
-                                    return true;
-                                } else {
-                                    int indexNameCounter;
-                                    try {
-                                        indexNameCounter = IndexMetadata.parseIndexNameCounter(entry.getKey());
-                                    } catch (IllegalArgumentException e) {
-                                        // index name is not in the %s-%d+ format so it will not crash with backing indices
-                                        return false;
-                                    }
-                                    return indexNameCounter > ds.getGeneration();
-                                }
-                            }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                    String prefix = DataStream.BACKING_INDEX_PREFIX + ds.getName() + "-";
+                    Set<String> conflicts =
+                        indicesLookup.subMap(prefix, DataStream.BACKING_INDEX_PREFIX + ds.getName() + ".") // '.' is the char after '-'
+                            .keySet().stream()
+                            .filter(s -> NUMBER_PATTERN.matcher(s.substring(prefix.length())).matches())
+                            .filter(s -> IndexMetadata.parseIndexNameCounter(s) > ds.getGeneration())
+                            .collect(Collectors.toSet());
 
                     if (conflicts.size() > 0) {
                         throw new IllegalStateException("data stream [" + ds.getName() +
                             "] could create backing indices that conflict with " + conflicts.size() + " existing index(s) or alias(s)" +
-                            " including '" + conflicts.keySet().iterator().next() + "'");
+                            " including '" + conflicts.iterator().next() + "'");
                     }
                 }
             }

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

@@ -1148,11 +1148,25 @@ public class MetadataTests extends ESTestCase {
 
             )
             .build();
-        // don't expect any exception when validating against non-backing indinces that don't conform to the backing indices naming
+        // 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));
     }
 
+    public void testValidateDataStreamsAllowsNamesThatStartsWithPrefix() {
+        String dataStreamName = "foo-datastream";
+        Metadata metadata = Metadata.builder(createIndices(10, 10, dataStreamName).metadata)
+            .put(
+                new IndexMetadata.Builder(DataStream.BACKING_INDEX_PREFIX + dataStreamName + "-something-100012")
+                    .settings(settings(Version.CURRENT))
+                    .numberOfShards(1)
+                    .numberOfReplicas(1)
+            ).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;