diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java b/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
index b369b98aa5d..6b65721fd3a 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
@@ -1226,6 +1226,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;
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java
index c3d1d30309a..e2547c5d086 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java
+++ b/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));
             }
         }
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java
index eca8a9a8510..c0417dacc8f 100644
--- a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java
@@ -26,8 +26,6 @@ import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
-import org.elasticsearch.common.collect.List;
-import org.elasticsearch.common.collect.Set;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.Index;
@@ -42,7 +40,12 @@ import org.elasticsearch.test.VersionUtils;
 import org.hamcrest.core.IsNull;
 import org.junit.Before;
 
+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;
@@ -78,7 +81,8 @@ public class MetadataDeleteIndexServiceTests extends ESTestCase {
     public void testDeleteSnapshotting() {
         String index = randomAlphaOfLength(5);
         Snapshot snapshot = new Snapshot("doesn't matter", new SnapshotId("snapshot name", "snapshot uuid"));
-        SnapshotsInProgress snaps = SnapshotsInProgress.of(List.of(new SnapshotsInProgress.Entry(snapshot, true, false,
+        SnapshotsInProgress snaps = SnapshotsInProgress.of(
+            org.elasticsearch.common.collect.List.of(new SnapshotsInProgress.Entry(snapshot, true, false,
                 SnapshotsInProgress.State.INIT, singletonList(new IndexId(index, "doesn't matter")),
                 System.currentTimeMillis(), (long) randomIntBetween(0, 1000), ImmutableOpenMap.of(),
                 SnapshotInfoTests.randomUserMetadata(), VersionUtils.randomVersion(random()))));
@@ -115,12 +119,13 @@ public class MetadataDeleteIndexServiceTests extends ESTestCase {
         int numBackingIndices = randomIntBetween(2, 5);
         String dataStreamName = randomAlphaOfLength(6).toLowerCase(Locale.ROOT);
         ClusterState before = DataStreamTestHelper.getClusterStateWithDataStreams(
-            List.of(new Tuple<>(dataStreamName, numBackingIndices)), List.of());
+            org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamName, numBackingIndices)),
+            org.elasticsearch.common.collect.List.of());
 
         int numIndexToDelete = randomIntBetween(1, numBackingIndices - 1);
 
         Index indexToDelete = before.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, numIndexToDelete)).getIndex();
-        ClusterState after = service.deleteIndices(before, Set.of(indexToDelete));
+        ClusterState after = service.deleteIndices(before, org.elasticsearch.common.collect.Set.of(indexToDelete));
 
         assertThat(after.metadata().getIndices().get(indexToDelete.getName()), IsNull.nullValue());
         assertThat(after.metadata().getIndices().size(), equalTo(numBackingIndices - 1));
@@ -128,14 +133,43 @@ 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(
+            org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamName, numBackingIndices)),
+            org.elasticsearch.common.collect.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);
         ClusterState before = DataStreamTestHelper.getClusterStateWithDataStreams(
-            List.of(new Tuple<>(dataStreamName, numBackingIndices)), List.of());
+            org.elasticsearch.common.collect.List.of(new Tuple<>(dataStreamName, numBackingIndices)),
+            org.elasticsearch.common.collect.List.of());
 
         Index indexToDelete = before.metadata().index(DataStream.getDefaultBackingIndexName(dataStreamName, numBackingIndices)).getIndex();
-        Exception e = expectThrows(IllegalArgumentException.class, () -> service.deleteIndices(before, Set.of(indexToDelete)));
+        Exception e = expectThrows(IllegalArgumentException.class,
+            () -> service.deleteIndices(before, org.elasticsearch.common.collect.Set.of(indexToDelete)));
 
         assertThat(e.getMessage(), containsString("index [" + indexToDelete.getName() + "] is the write index for data stream [" +
             dataStreamName + "] and cannot be deleted"));