Prohibit restoring a data stream alias with a conflicting write data stream (#81217)

Prohibit restoring a data stream alias if its write data stream conflicts with
the write data stream of an existing data stream alias.

Currently, the write data stream of data stream alias in the snapshot is ignored and
the write data stream of the data stream alias currently active in the cluster
is always picked.

In order to resolve this properly, I've merged the merge(...) and renameDataStreams(...)
methods into a restore(...) method, which merges a data stream and does renaming at
the same time. This should be ok, since renaming can only be done as part of restoring.
More importantly, this allows to validate whether an existing alias and
an alias from a snapshot to not have conflicting write data streams while at the same time
allowing that a write data stream to be renamed (because data streams got renamed).

Closes #80976

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

@@ -221,38 +221,60 @@ public class DataStreamAlias extends AbstractDiffable<DataStreamAlias> implement
     }
 
     /**
-     * Returns a new {@link DataStreamAlias} instance containing data streams referenced in this instance
-     * and the other instance. If this instance doesn't have a write data stream then the write index of
-     * the other data stream becomes the write data stream of the returned instance.
+     * Performs alias related restore operations for this instance as part of the entire restore operation.
+     *
+     * If a previous instance is provided then it merges the data streams referenced in this instance and
+     * the previous instance. If this instance doesn't have a write data stream then the write index of
+     * the other data stream becomes the write data stream of the returned instance. If both this and
+     * previous instances have a write data stream then these write data streams need to be the same.
+     *
+     * If a renamePattern and renameReplacement is provided then data streams this instance is referring to
+     * are renamed. Assuming that those data streams match with the specified renamePattern.
+     *
+     * @param previous          Optionally, the alias instance that this alias instance is replacing.
+     * @param renamePattern     Optionally, the pattern that is required to match to rename data streams this alias is referring to.
+     * @param renameReplacement Optionally, the replacement used to rename data streams this alias is referring to.
+     * @return a new alias instance that can be applied in the cluster state
      */
-    public DataStreamAlias merge(DataStreamAlias other) {
-        Set<String> mergedDataStreams = new HashSet<>(other.getDataStreams());
-        mergedDataStreams.addAll(this.getDataStreams());
+    public DataStreamAlias restore(DataStreamAlias previous, String renamePattern, String renameReplacement) {
+        Set<String> mergedDataStreams = previous != null ? new HashSet<>(previous.getDataStreams()) : new HashSet<>();
 
         String writeDataStream = this.writeDataStream;
-        if (writeDataStream == null) {
-            if (other.getWriteDataStream() != null && mergedDataStreams.contains(other.getWriteDataStream())) {
-                writeDataStream = other.getWriteDataStream();
+        if (renamePattern != null && renameReplacement != null) {
+            this.dataStreams.stream().map(s -> s.replaceAll(renamePattern, renameReplacement)).forEach(mergedDataStreams::add);
+            if (writeDataStream != null) {
+                writeDataStream = writeDataStream.replaceAll(renamePattern, renameReplacement);
             }
+        } else {
+            mergedDataStreams.addAll(this.dataStreams);
         }
 
-        return new DataStreamAlias(this.name, List.copyOf(mergedDataStreams), writeDataStream, filter);
-    }
-
-    /**
-     * Returns a new instance with potentially renamed data stream names and write data stream name.
-     * If a data stream name matches with the provided rename pattern then it is renamed according
-     * to the provided rename replacement.
-     */
-    public DataStreamAlias renameDataStreams(String renamePattern, String renameReplacement) {
-        List<String> renamedDataStreams = this.dataStreams.stream()
-            .map(s -> s.replaceAll(renamePattern, renameReplacement))
-            .collect(Collectors.toList());
-        String writeDataStream = this.writeDataStream;
-        if (writeDataStream != null) {
-            writeDataStream = writeDataStream.replaceAll(renamePattern, renameReplacement);
+        if (previous != null) {
+            if (writeDataStream != null && previous.getWriteDataStream() != null) {
+                String previousWriteDataStream = previous.getWriteDataStream();
+                if (renamePattern != null && renameReplacement != null) {
+                    previousWriteDataStream = previousWriteDataStream.replaceAll(renamePattern, renameReplacement);
+                }
+                if (writeDataStream.equals(previousWriteDataStream) == false) {
+                    throw new IllegalArgumentException(
+                        "cannot merge alias ["
+                            + name
+                            + "], write data stream of this ["
+                            + writeDataStream
+                            + "] and write data stream of other ["
+                            + previous.getWriteDataStream()
+                            + "] are different"
+                    );
+                }
+            } else if (writeDataStream == null && previous.getWriteDataStream() != null) {
+                // The write alias should exist in the set of merged data streams. It shouldn't be possible to construct an alias with
+                // a write data stream, that doesn't exist in the list of data streams.
+                assert mergedDataStreams.contains(previous.getWriteDataStream());
+                writeDataStream = previous.getWriteDataStream();
+            }
         }
-        return new DataStreamAlias(this.name, renamedDataStreams, writeDataStream, filter);
+
+        return new DataStreamAlias(this.name, List.copyOf(mergedDataStreams), writeDataStream, filter);
     }
 
     public static Diff<DataStreamAlias> readDiffFrom(StreamInput in) throws IOException {

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -1417,25 +1417,13 @@ public class RestoreService implements ClusterStateApplier {
                     .collect(Collectors.toMap(DataStream::getName, Function.identity()))
             );
             final Map<String, DataStreamAlias> updatedDataStreamAliases = new HashMap<>(currentState.metadata().dataStreamAliases());
-            metadata.dataStreamAliases()
-                .values()
-                .stream()
-                // Optionally rename the data stream names for each alias
-                .map(alias -> {
-                    if (request.renamePattern() != null && request.renameReplacement() != null) {
-                        return alias.renameDataStreams(request.renamePattern(), request.renameReplacement());
-                    } else {
-                        return alias;
-                    }
-                })
-                .forEach(alias -> {
-                    final DataStreamAlias current = updatedDataStreamAliases.putIfAbsent(alias.getName(), alias);
-                    if (current != null) {
-                        // Merge data stream alias from snapshot with an existing data stream aliases in target cluster:
-                        DataStreamAlias newInstance = alias.merge(current);
-                        updatedDataStreamAliases.put(alias.getName(), newInstance);
-                    }
-                });
+            for (DataStreamAlias alias : metadata.dataStreamAliases().values()) {
+                // Merge data stream alias from snapshot with an existing data stream aliases in target cluster:
+                updatedDataStreamAliases.compute(
+                    alias.getName(),
+                    (key, previous) -> alias.restore(previous, request.renamePattern(), request.renameReplacement())
+                );
+            }
             mdBuilder.dataStreams(updatedDataStreams, updatedDataStreamAliases);
         }
 

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

@@ -189,41 +189,83 @@ public class DataStreamAliasTests extends AbstractSerializingTestCase<DataStream
         }
     }
 
-    public void testMerge() {
+    public void testRestore() {
         {
             DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), null, null);
             DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), null, null);
-            DataStreamAlias result = alias1.merge(alias2);
+            DataStreamAlias result = alias1.restore(alias2, null, null);
             assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3"));
             assertThat(result.getWriteDataStream(), nullValue());
         }
         {
             DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
             DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), null, null);
-            DataStreamAlias result = alias1.merge(alias2);
+            DataStreamAlias result = alias1.restore(alias2, null, null);
             assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3"));
             assertThat(result.getWriteDataStream(), equalTo("ds-2"));
         }
         {
             DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
             DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), "ds-3", null);
-            DataStreamAlias result = alias1.merge(alias2);
+            var e = expectThrows(IllegalArgumentException.class, () -> alias1.restore(alias2, null, null));
+            assertThat(
+                e.getMessage(),
+                equalTo(
+                    "cannot merge alias [my-alias], write data stream of this [ds-2] and write data stream of other [ds-3] are different"
+                )
+            );
+        }
+        {
+            DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
+            DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), "ds-2", null);
+            DataStreamAlias result = alias1.restore(alias2, null, null);
             assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3"));
             assertThat(result.getWriteDataStream(), equalTo("ds-2"));
         }
         {
             DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), null, null);
             DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), "ds-3", null);
-            DataStreamAlias result = alias1.merge(alias2);
+            DataStreamAlias result = alias1.restore(alias2, null, null);
             assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3"));
             assertThat(result.getWriteDataStream(), equalTo("ds-3"));
         }
     }
 
-    public void testRenameDataStreams() {
-        DataStreamAlias alias = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
-        DataStreamAlias result = alias.renameDataStreams("ds-2", "ds-3");
-        assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-3"));
+    public void testRestoreWithRename() {
+        {
+            DataStreamAlias alias = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
+            DataStreamAlias result = alias.restore(null, "ds-2", "ds-3");
+            assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-3"));
+            assertThat(result.getWriteDataStream(), equalTo("ds-3"));
+        }
+        {
+            DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
+            DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-4", "ds-5"), "ds-2", null);
+            DataStreamAlias result = alias1.restore(alias2, "ds-2", "ds-3");
+            assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3", "ds-4", "ds-5"));
+            assertThat(result.getWriteDataStream(), equalTo("ds-3"));
+        }
+        {
+            DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), "ds-2", null);
+            DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-3", "ds-4", "ds-5"), "ds-3", null);
+            DataStreamAlias result = alias1.restore(alias2, "ds-2", "ds-3");
+            assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-3", "ds-4", "ds-5"));
+            assertThat(result.getWriteDataStream(), equalTo("ds-3"));
+        }
+        {
+            DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2", "ds-3"), "ds-3", null);
+            DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-4", "ds-5"), "ds-2", null);
+            DataStreamAlias result = alias1.restore(alias2, "ds-2", "ds-3");
+            assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3", "ds-4", "ds-5"));
+            assertThat(result.getWriteDataStream(), equalTo("ds-3"));
+        }
+    }
+
+    public void testRestoreDataStreamWithWriteDataStreamThatDoesNotExistInOriginalAlias() {
+        DataStreamAlias alias1 = new DataStreamAlias("my-alias", List.of("ds-1", "ds-2"), null, null);
+        DataStreamAlias alias2 = new DataStreamAlias("my-alias", List.of("ds-2", "ds-3"), "ds-3", null);
+        DataStreamAlias result = alias1.restore(alias2, "ds-3", null);
+        assertThat(result.getDataStreams(), containsInAnyOrder("ds-1", "ds-2", "ds-3"));
         assertThat(result.getWriteDataStream(), equalTo("ds-3"));
     }
 }