Fix NPE when deleting multiple backing indices on a data stream (#62274)

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

@@ -1150,6 +1150,10 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             return this;
         }
 
+        public DataStream dataStream(String dataStreamName) {
+            return ((DataStreamMetadata) customs.get(DataStreamMetadata.TYPE)).dataStreams().get(dataStreamName);
+        }
+
         public Builder dataStreams(Map<String, DataStream> dataStreams) {
             this.customs.put(DataStreamMetadata.TYPE, new DataStreamMetadata(dataStreams));
             return this;

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

@@ -128,7 +128,7 @@ public class MetadataDeleteIndexService {
             clusterBlocksBuilder.removeIndexBlocks(indexName);
             metadataBuilder.remove(indexName);
             if (backingIndices.containsKey(index)) {
-                DataStream parent = backingIndices.get(index);
+                DataStream parent = metadataBuilder.dataStream(backingIndices.get(index).getName());
                 metadataBuilder.put(parent.removeBackingIndex(index));
             }
         }

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

@@ -41,9 +41,12 @@ import org.hamcrest.core.IsNull;
 import org.junit.Before;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonList;
@@ -129,6 +132,32 @@ public class MetadataDeleteIndexServiceTests extends ESTestCase {
             DataStream.getDefaultBackingIndexName(dataStreamName, numIndexToDelete)), IsNull.nullValue());
     }
 
+    public void testDeleteMultipleBackingIndexForDataStream() {
+        int numBackingIndices = randomIntBetween(3, 5);
+        int numBackingIndicesToDelete = randomIntBetween(2, numBackingIndices - 1);
+        String dataStreamName = randomAlphaOfLength(6).toLowerCase(Locale.ROOT);
+        ClusterState before = DataStreamTestHelper.getClusterStateWithDataStreams(
+            List.of(new Tuple<>(dataStreamName, numBackingIndices)), List.of());
+
+        List<Integer> indexNumbersToDelete =
+            randomSubsetOf(numBackingIndicesToDelete, IntStream.rangeClosed(1, numBackingIndices - 1).boxed().collect(Collectors.toList()));
+
+        Set<Index> indicesToDelete = new HashSet<>();
+        for (int k : indexNumbersToDelete) {
+            indicesToDelete.add(before.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, k)).getIndex());
+        }
+        ClusterState after = service.deleteIndices(before, indicesToDelete);
+
+        DataStream dataStream = after.metadata().dataStreams().get(dataStreamName);
+        assertThat(dataStream, IsNull.notNullValue());
+        assertThat(dataStream.getIndices().size(), equalTo(numBackingIndices - indexNumbersToDelete.size()));
+        for (Index i : indicesToDelete) {
+            assertThat(after.metadata().getIndices().get(i.getName()), IsNull.nullValue());
+            assertFalse(dataStream.getIndices().contains(i));
+        }
+        assertThat(after.metadata().getIndices().size(), equalTo(numBackingIndices - indexNumbersToDelete.size()));
+    }
+
     public void testDeleteCurrentWriteIndexForDataStream() {
         int numBackingIndices = randomIntBetween(1, 5);
         String dataStreamName = randomAlphaOfLength(6).toLowerCase(Locale.ROOT);