Converting an Existing Data Stream to a System DataStream is Broken (elastic#121392)

Adds support of converting existing data stream to a system data stream as part of existing system_index_metadata_upgrade_service task

Author: alexey-ivanov-es

docs/changelog/121392.yaml

@@ -0,0 +1,5 @@
+pr: 121392
+summary: Include data streams when converting an existing resource to a system resource
+area: Infra/Core
+type: bug
+issues: []

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

@@ -301,6 +301,13 @@ public boolean isFailureStoreIndex(String indexName) {
         return failureIndices.containsIndex(indexName);
     }
 
+    /**
+     * Returns true if the index name provided belongs to this data stream.
+     */
+    public boolean containsIndex(String indexName) {
+        return backingIndices.containsIndex(indexName) || failureIndices.containsIndex(indexName);
+    }
+
     public DataStreamOptions getDataStreamOptions() {
         return dataStreamOptions;
     }
@@ -782,8 +789,9 @@ public DataStream addBackingIndex(Metadata clusterMetadata, Index index) {
         // ensure that no aliases reference index
         ensureNoAliasesOnIndex(clusterMetadata, index);
 
-        List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices);
-        backingIndices.add(0, index);
+        List<Index> backingIndices = new ArrayList<>(this.backingIndices.indices.size() + 1);
+        backingIndices.add(index);
+        backingIndices.addAll(this.backingIndices.indices);
         assert backingIndices.size() == this.backingIndices.indices.size() + 1;
         return copy().setBackingIndices(this.backingIndices.copy().setIndices(backingIndices).build())
             .setGeneration(generation + 1)
@@ -808,8 +816,9 @@ public DataStream addFailureStoreIndex(Metadata clusterMetadata, Index index) {
 
         ensureNoAliasesOnIndex(clusterMetadata, index);
 
-        List<Index> updatedFailureIndices = new ArrayList<>(failureIndices.indices);
-        updatedFailureIndices.add(0, index);
+        List<Index> updatedFailureIndices = new ArrayList<>(failureIndices.indices.size() + 1);
+        updatedFailureIndices.add(index);
+        updatedFailureIndices.addAll(failureIndices.indices);
         assert updatedFailureIndices.size() == failureIndices.indices.size() + 1;
         return copy().setFailureIndices(failureIndices.copy().setIndices(updatedFailureIndices).build())
             .setGeneration(generation + 1)
@@ -1039,7 +1048,7 @@ private boolean isIndexOlderThan(
      * we return false.
      */
     public boolean isIndexManagedByDataStreamLifecycle(Index index, Function<String, IndexMetadata> indexMetadataSupplier) {
-        if (backingIndices.containsIndex(index.getName()) == false && failureIndices.containsIndex(index.getName()) == false) {
+        if (containsIndex(index.getName()) == false) {
             return false;
         }
         IndexMetadata indexMetadata = indexMetadataSupplier.apply(index.getName());