diff --git a/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/FailureStoreQueryParamIT.java b/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/FailureStoreQueryParamIT.java
new file mode 100644
index 0000000000000..1aee1e5ecc7da
--- /dev/null
+++ b/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/FailureStoreQueryParamIT.java
@@ -0,0 +1,184 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.hamcrest.MatcherAssert;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * This should be a yaml test, but in order to write one we would need to expose the new parameter in the rest-api-spec.
+ * We do not want to do that until the feature flag is removed. For this reason, we temporarily, test the affected APIs here.
+ * Please convert this to a yaml test when the feature flag is removed.
+ */
+public class FailureStoreQueryParamIT extends DisabledSecurityDataStreamTestCase {
+
+    private static final String DATA_STREAM_NAME = "failure-data-stream";
+    private String backingIndex;
+    private String failureStoreIndex;
+
+    @SuppressWarnings("unchecked")
+    @Before
+    public void setup() throws IOException {
+        Request putComposableIndexTemplateRequest = new Request("POST", "/_index_template/ds-template");
+        putComposableIndexTemplateRequest.setJsonEntity("""
+            {
+              "index_patterns": ["failure-data-stream"],
+              "template": {
+                "settings": {
+                  "number_of_replicas": 0
+                }
+              },
+              "data_stream": {
+                "failure_store": true
+              }
+            }
+            """);
+        assertOK(client().performRequest(putComposableIndexTemplateRequest));
+
+        assertOK(client().performRequest(new Request("PUT", "/_data_stream/" + DATA_STREAM_NAME)));
+        ensureGreen(DATA_STREAM_NAME);
+
+        final Response dataStreamResponse = client().performRequest(new Request("GET", "/_data_stream/" + DATA_STREAM_NAME));
+        List<Object> dataStreams = (List<Object>) entityAsMap(dataStreamResponse).get("data_streams");
+        MatcherAssert.assertThat(dataStreams.size(), is(1));
+        Map<String, Object> dataStream = (Map<String, Object>) dataStreams.get(0);
+        MatcherAssert.assertThat(dataStream.get("name"), equalTo(DATA_STREAM_NAME));
+        List<String> backingIndices = getBackingIndices(dataStream);
+        MatcherAssert.assertThat(backingIndices.size(), is(1));
+        List<String> failureStore = getFailureStore(dataStream);
+        MatcherAssert.assertThat(failureStore.size(), is(1));
+        backingIndex = backingIndices.get(0);
+        failureStoreIndex = failureStore.get(0);
+    }
+
+    public void testGetIndexApi() throws IOException {
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME));
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(2));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME + "?failure_store=exclude"));
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME + "?failure_store=only"));
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public void testGetIndexStatsApi() throws IOException {
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME + "/_stats"));
+            Map<String, Object> indices = (Map<String, Object>) entityAsMap(indicesResponse).get("indices");
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_stats?failure_store=include")
+            );
+            Map<String, Object> indices = (Map<String, Object>) entityAsMap(indicesResponse).get("indices");
+            MatcherAssert.assertThat(indices.size(), is(2));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_stats?failure_store=only")
+            );
+            Map<String, Object> indices = (Map<String, Object>) entityAsMap(indicesResponse).get("indices");
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+    }
+
+    public void testGetIndexSettingsApi() throws IOException {
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME + "/_settings"));
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_settings?failure_store=include")
+            );
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(2));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_settings?failure_store=only")
+            );
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+    }
+
+    public void testGetIndexMappingApi() throws IOException {
+        {
+            final Response indicesResponse = client().performRequest(new Request("GET", "/" + DATA_STREAM_NAME + "/_mapping"));
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_mapping?failure_store=include")
+            );
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(2));
+            MatcherAssert.assertThat(indices.containsKey(backingIndex), is(true));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+        {
+            final Response indicesResponse = client().performRequest(
+                new Request("GET", "/" + DATA_STREAM_NAME + "/_mapping?failure_store=only")
+            );
+            Map<String, Object> indices = entityAsMap(indicesResponse);
+            MatcherAssert.assertThat(indices.size(), is(1));
+            MatcherAssert.assertThat(indices.containsKey(failureStoreIndex), is(true));
+        }
+    }
+
+    private List<String> getBackingIndices(Map<String, Object> response) {
+        return getIndices(response, "indices");
+    }
+
+    private List<String> getFailureStore(Map<String, Object> response) {
+        return getIndices(response, "failure_indices");
+
+    }
+
+    @SuppressWarnings("unchecked")
+    private List<String> getIndices(Map<String, Object> response, String fieldName) {
+        List<Map<String, String>> indices = (List<Map<String, String>>) response.get(fieldName);
+        return indices.stream().map(index -> index.get("index_name")).toList();
+    }
+}
diff --git a/server/src/main/java/org/elasticsearch/TransportVersions.java b/server/src/main/java/org/elasticsearch/TransportVersions.java
index 7b3ca0e2f069a..ab8e811237607 100644
--- a/server/src/main/java/org/elasticsearch/TransportVersions.java
+++ b/server/src/main/java/org/elasticsearch/TransportVersions.java
@@ -191,6 +191,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion NESTED_KNN_MORE_INNER_HITS = def(8_577_00_0);
     public static final TransportVersion REQUIRE_DATA_STREAM_ADDED = def(8_578_00_0);
     public static final TransportVersion ML_INFERENCE_COHERE_EMBEDDINGS_ADDED = def(8_579_00_0);
+    public static final TransportVersion ADD_FAILURE_STORE_OPTIONS = def(8_579_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,
diff --git a/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java b/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
index 5a168e1695e55..0aefe5ed65c1f 100644
--- a/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
+++ b/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
@@ -10,6 +10,7 @@
 import joptsimple.internal.Strings;
 
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -44,9 +45,12 @@
  * @param generalOptions, applies to all the resolved indices and defines if throttled will be included and if certain type of
  *                        aliases or indices are allowed, or they will throw an error.
  */
-public record IndicesOptions(ConcreteTargetOptions concreteTargetOptions, WildcardOptions wildcardOptions, GeneralOptions generalOptions)
-    implements
-        ToXContentFragment {
+public record IndicesOptions(
+    ConcreteTargetOptions concreteTargetOptions,
+    WildcardOptions wildcardOptions,
+    GeneralOptions generalOptions,
+    FailureStoreOptions failureStoreOptions
+) implements ToXContentFragment {
 
     public static IndicesOptions.Builder newBuilder() {
         return new Builder();
@@ -381,6 +385,73 @@ public static Builder newBuilder(GeneralOptions generalOptions) {
         }
     }
 
+    public record FailureStoreOptions(boolean includeBackingIndices, boolean includeFailureIndices) implements ToXContentFragment {
+
+        public static final FailureStoreOptions DEFAULT = newBuilder().build();
+
+        public static Builder newBuilder() {
+            return new Builder();
+        }
+
+        public static Builder newBuilder(FailureStoreOptions failureStoreOptions) {
+            return new Builder(failureStoreOptions);
+        }
+
+        public static FailureStoreOptions fromRequest(RestRequest request, FailureStoreOptions defaultOption) {
+            String failureStoreString = request.param("failure_store");
+            if (failureStoreString == null) {
+                return defaultOption == null ? DEFAULT : defaultOption;
+            }
+            return switch (failureStoreString) {
+                case "only" -> new FailureStoreOptions(false, true);
+                case "exclude" -> DEFAULT;
+                case "include" -> new FailureStoreOptions(true, true);
+                default -> throw new IllegalArgumentException(
+                    "parameter [failure_store] does not support value [" + failureStoreString + "]"
+                );
+            };
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            String value;
+            if (includeFailureIndices && includeBackingIndices == false) {
+                value = "only";
+            } else if (includeFailureIndices) {
+                value = "include";
+            } else {
+                value = "exclude";
+            }
+            return builder.field("failure_store", value);
+        }
+
+        public static class Builder {
+            private boolean includeBackingIndices;
+            private boolean includeFailureIndices;
+
+            Builder() {}
+
+            Builder(FailureStoreOptions failureStoreOptions) {
+                includeBackingIndices = failureStoreOptions.includeBackingIndices();
+                includeFailureIndices = failureStoreOptions.includeFailureIndices();
+            }
+
+            public Builder includeBackingIndices(boolean includeBackingIndices) {
+                this.includeBackingIndices = includeBackingIndices;
+                return this;
+            }
+
+            public Builder includeFailureIndices(boolean includeFailureIndices) {
+                this.includeFailureIndices = includeFailureIndices;
+                return this;
+            }
+
+            public FailureStoreOptions build() {
+                return new FailureStoreOptions(includeBackingIndices, includeFailureIndices);
+            }
+        }
+    }
+
     /**
      * This class is maintained for backwards compatibility and performance purposes. We use it for serialisation along with {@link Option}.
      */
@@ -411,7 +482,9 @@ private enum Option {
         ERROR_WHEN_ALIASES_TO_MULTIPLE_INDICES,
 
         ERROR_WHEN_CLOSED_INDICES,
-        EXCLUDE_THROTTLED;
+        EXCLUDE_THROTTLED,
+        EXCLUDE_BACKING_INDICES,
+        INCLUDE_FAILURE_STORE;
     }
 
     private static final DeprecationLogger DEPRECATION_LOGGER = DeprecationLogger.getLogger(IndicesOptions.class);
@@ -539,6 +612,14 @@ public boolean ignoreThrottled() {
         return generalOptions().removeThrottled();
     }
 
+    public boolean includeBackingIndices() {
+        return failureStoreOptions().includeBackingIndices();
+    }
+
+    public boolean includeFailureIndices() {
+        return failureStoreOptions().includeFailureIndices();
+    }
+
     public void writeIndicesOptions(StreamOutput out) throws IOException {
         EnumSet<Option> backwardsCompatibleOptions = EnumSet.noneOf(Option.class);
         if (allowNoIndices()) {
@@ -559,6 +640,14 @@ public void writeIndicesOptions(StreamOutput out) throws IOException {
         if (ignoreUnavailable()) {
             backwardsCompatibleOptions.add(Option.ALLOW_UNAVAILABLE_CONCRETE_TARGETS);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ADD_FAILURE_STORE_OPTIONS)) {
+            if (failureStoreOptions.includeBackingIndices() == false) {
+                backwardsCompatibleOptions.add(Option.EXCLUDE_BACKING_INDICES);
+            }
+            if (failureStoreOptions.includeFailureIndices) {
+                backwardsCompatibleOptions.add(Option.INCLUDE_FAILURE_STORE);
+            }
+        }
         out.writeEnumSet(backwardsCompatibleOptions);
 
         EnumSet<WildcardStates> states = EnumSet.noneOf(WildcardStates.class);
@@ -586,12 +675,17 @@ public static IndicesOptions readIndicesOptions(StreamInput in) throws IOExcepti
             .allowAliasToMultipleIndices(options.contains(Option.ERROR_WHEN_ALIASES_TO_MULTIPLE_INDICES) == false)
             .removeThrottled(options.contains(Option.EXCLUDE_THROTTLED))
             .build();
+        FailureStoreOptions failureStoreOptions = FailureStoreOptions.newBuilder()
+            .includeBackingIndices(options.contains(Option.EXCLUDE_BACKING_INDICES) == false)
+            .includeFailureIndices(options.contains(Option.INCLUDE_FAILURE_STORE))
+            .build();
         return new IndicesOptions(
             options.contains(Option.ALLOW_UNAVAILABLE_CONCRETE_TARGETS)
                 ? ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS
                 : ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
             wildcardOptions,
-            generalOptions
+            generalOptions,
+            failureStoreOptions
         );
     }
 
@@ -599,6 +693,7 @@ public static class Builder {
         private ConcreteTargetOptions concreteTargetOptions = ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS;
         private WildcardOptions wildcardOptions = WildcardOptions.DEFAULT;
         private GeneralOptions generalOptions = GeneralOptions.DEFAULT;
+        private FailureStoreOptions failureStoreOptions = FailureStoreOptions.DEFAULT;
 
         Builder() {}
 
@@ -633,8 +728,18 @@ public Builder generalOptions(GeneralOptions.Builder generalOptions) {
             return this;
         }
 
+        public Builder failureStoreOptions(FailureStoreOptions failureStoreOptions) {
+            this.failureStoreOptions = failureStoreOptions;
+            return this;
+        }
+
+        public Builder failureStoreOptions(FailureStoreOptions.Builder failureStoreOptions) {
+            this.failureStoreOptions = failureStoreOptions.build();
+            return this;
+        }
+
         public IndicesOptions build() {
-            return new IndicesOptions(concreteTargetOptions, wildcardOptions, generalOptions);
+            return new IndicesOptions(concreteTargetOptions, wildcardOptions, generalOptions, failureStoreOptions);
         }
     }
 
@@ -720,6 +825,34 @@ public static IndicesOptions fromOptions(
         boolean forbidClosedIndices,
         boolean ignoreAliases,
         boolean ignoreThrottled
+    ) {
+        return fromOptions(
+            ignoreUnavailable,
+            allowNoIndices,
+            expandToOpenIndices,
+            expandToClosedIndices,
+            expandToHiddenIndices,
+            allowAliasesToMultipleIndices,
+            forbidClosedIndices,
+            ignoreAliases,
+            ignoreThrottled,
+            true,
+            false
+        );
+    }
+
+    public static IndicesOptions fromOptions(
+        boolean ignoreUnavailable,
+        boolean allowNoIndices,
+        boolean expandToOpenIndices,
+        boolean expandToClosedIndices,
+        boolean expandToHiddenIndices,
+        boolean allowAliasesToMultipleIndices,
+        boolean forbidClosedIndices,
+        boolean ignoreAliases,
+        boolean ignoreThrottled,
+        boolean includeBackingIndices,
+        boolean includeFailureStore
     ) {
         final WildcardOptions wildcards = WildcardOptions.newBuilder()
             .includeOpen(expandToOpenIndices)
@@ -733,10 +866,15 @@ public static IndicesOptions fromOptions(
             .allowClosedIndices(forbidClosedIndices == false)
             .removeThrottled(ignoreThrottled)
             .build();
+        final FailureStoreOptions failureStoreOptions = FailureStoreOptions.newBuilder()
+            .includeBackingIndices(includeBackingIndices)
+            .includeFailureIndices(includeFailureStore)
+            .build();
         return new IndicesOptions(
             ignoreUnavailable ? ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS : ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
             wildcards,
-            generalOptions
+            generalOptions,
+            failureStoreOptions
         );
     }
 
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
index ff31c6fe950d7..23a7c37070847 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
@@ -277,6 +277,11 @@ public boolean rolloverOnWrite() {
         return rolloverOnWrite;
     }
 
+    @Nullable
+    public Index getFailureStoreWriteIndex() {
+        return failureIndices.isEmpty() ? null : failureIndices.get(failureIndices.size() - 1);
+    }
+
     /**
      * @param timestamp The timestamp used to select a backing index based on its start and end time.
      * @param metadata  The metadata that is used to fetch the start and end times for backing indices of this data stream.
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java
index 0446b479b191d..1f3cc39dcacb9 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java
@@ -359,12 +359,22 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
                     concreteIndicesResult.add(writeIndex);
                 }
             } else if (indexAbstraction.getType() == Type.DATA_STREAM && context.isResolveToWriteIndex()) {
-                Index writeIndex = indexAbstraction.getWriteIndex();
-                if (addIndex(writeIndex, null, context)) {
-                    concreteIndicesResult.add(writeIndex);
+                if (shouldIncludeBackingIndices(context.getOptions())) {
+                    Index writeIndex = indexAbstraction.getWriteIndex();
+                    if (addIndex(writeIndex, null, context)) {
+                        concreteIndicesResult.add(writeIndex);
+                    }
+                }
+                DataStream dataStream = (DataStream) indexAbstraction;
+                if (shouldIncludeFailureStoreIndices(context.getOptions(), dataStream)) {
+                    Index failureStoreWriteIndex = dataStream.getFailureStoreWriteIndex();
+                    if (failureStoreWriteIndex != null && addIndex(failureStoreWriteIndex, null, context)) {
+                        concreteIndicesResult.add(failureStoreWriteIndex);
+                    }
                 }
             } else {
-                if (indexAbstraction.getIndices().size() > 1 && context.getOptions().allowAliasesToMultipleIndices() == false) {
+                if (resolvesToMoreThanOneIndices(indexAbstraction, context)
+                    && context.getOptions().allowAliasesToMultipleIndices() == false) {
                     String[] indexNames = new String[indexAbstraction.getIndices().size()];
                     int i = 0;
                     for (Index indexName : indexAbstraction.getIndices()) {
@@ -380,9 +390,27 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
                     );
                 }
 
-                for (Index index : indexAbstraction.getIndices()) {
-                    if (shouldTrackConcreteIndex(context, context.getOptions(), index)) {
-                        concreteIndicesResult.add(index);
+                if (indexAbstraction.getType() == Type.DATA_STREAM) {
+                    DataStream dataStream = (DataStream) indexAbstraction;
+                    if (shouldIncludeBackingIndices(context.getOptions())) {
+                        for (Index index : indexAbstraction.getIndices()) {
+                            if (shouldTrackConcreteIndex(context, context.getOptions(), index)) {
+                                concreteIndicesResult.add(index);
+                            }
+                        }
+                    }
+                    if (shouldIncludeFailureStoreIndices(context.getOptions(), dataStream)) {
+                        for (Index index : dataStream.getFailureIndices()) {
+                            if (shouldTrackConcreteIndex(context, context.getOptions(), index)) {
+                                concreteIndicesResult.add(index);
+                            }
+                        }
+                    }
+                } else {
+                    for (Index index : indexAbstraction.getIndices()) {
+                        if (shouldTrackConcreteIndex(context, context.getOptions(), index)) {
+                            concreteIndicesResult.add(index);
+                        }
                     }
                 }
             }
@@ -395,6 +423,21 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
         return concreteIndicesResult.toArray(Index.EMPTY_ARRAY);
     }
 
+    private static boolean resolvesToMoreThanOneIndices(IndexAbstraction indexAbstraction, Context context) {
+        if (indexAbstraction.getType() == Type.DATA_STREAM) {
+            DataStream dataStream = (DataStream) indexAbstraction;
+            int count = 0;
+            if (shouldIncludeBackingIndices(context.getOptions())) {
+                count += dataStream.getIndices().size();
+            }
+            if (shouldIncludeFailureStoreIndices(context.getOptions(), dataStream)) {
+                count += dataStream.getFailureIndices().size();
+            }
+            return count > 1;
+        }
+        return indexAbstraction.getIndices().size() > 1;
+    }
+
     private void checkSystemIndexAccess(Context context, Set<Index> concreteIndices) {
         final Predicate<String> systemIndexAccessPredicate = context.getSystemIndexAccessPredicate();
         if (systemIndexAccessPredicate == ALWAYS_TRUE) {
@@ -1321,7 +1364,18 @@ private static Stream<String> expandToOpenClosed(Context context, Stream<IndexAb
                 } else if (context.isPreserveDataStreams() && indexAbstraction.getType() == Type.DATA_STREAM) {
                     return Stream.of(indexAbstraction.getName());
                 } else {
-                    Stream<IndexMetadata> indicesStateStream = indexAbstraction.getIndices().stream().map(context.state.metadata()::index);
+                    Stream<IndexMetadata> indicesStateStream = Stream.of();
+                    if (shouldIncludeBackingIndices(context.getOptions())) {
+                        indicesStateStream = indexAbstraction.getIndices().stream().map(context.state.metadata()::index);
+                    }
+                    if (indexAbstraction.getType() == Type.DATA_STREAM
+                        && shouldIncludeFailureStoreIndices(context.getOptions(), (DataStream) indexAbstraction)) {
+                        DataStream dataStream = (DataStream) indexAbstraction;
+                        indicesStateStream = Stream.concat(
+                            indicesStateStream,
+                            dataStream.getFailureIndices().stream().map(context.state.metadata()::index)
+                        );
+                    }
                     if (excludeState != null) {
                         indicesStateStream = indicesStateStream.filter(indexMeta -> indexMeta.getState() != excludeState);
                     }
@@ -1363,6 +1417,9 @@ private static List<String> resolveEmptyOrTrivialWildcardWithAllowedSystemIndice
         }
 
         private static String[] resolveEmptyOrTrivialWildcardToAllIndices(IndicesOptions options, Metadata metadata) {
+            if (shouldIncludeBackingIndices(options) == false) {
+                return Strings.EMPTY_ARRAY;
+            }
             if (options.expandWildcardsOpen() && options.expandWildcardsClosed() && options.expandWildcardsHidden()) {
                 return metadata.getConcreteAllIndices();
             } else if (options.expandWildcardsOpen() && options.expandWildcardsClosed()) {
@@ -1381,6 +1438,14 @@ private static String[] resolveEmptyOrTrivialWildcardToAllIndices(IndicesOptions
         }
     }
 
+    private static boolean shouldIncludeBackingIndices(IndicesOptions options) {
+        return DataStream.isFailureStoreEnabled() == false || options.includeBackingIndices();
+    }
+
+    private static boolean shouldIncludeFailureStoreIndices(IndicesOptions options, DataStream dataStream) {
+        return DataStream.isFailureStoreEnabled() && options.includeFailureIndices() && dataStream.isFailureStore();
+    }
+
     public static final class DateMathExpressionResolver {
 
         private static final DateFormatter DEFAULT_DATE_FORMATTER = DateFormatter.forPattern("uuuu.MM.dd");
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
index db10bdd985d59..a9017c582ab97 100644
--- a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
+++ b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.settings.Settings;
@@ -64,6 +65,15 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final GetIndexRequest getIndexRequest = new GetIndexRequest();
         getIndexRequest.indices(indices);
         getIndexRequest.indicesOptions(IndicesOptions.fromRequest(request, getIndexRequest.indicesOptions()));
+        if (DataStream.isFailureStoreEnabled()) {
+            IndicesOptions.FailureStoreOptions failureStoreOptions = IndicesOptions.FailureStoreOptions.fromRequest(
+                request,
+                getIndexRequest.indicesOptions().failureStoreOptions()
+            );
+            getIndexRequest.indicesOptions(
+                IndicesOptions.newBuilder(getIndexRequest.indicesOptions()).failureStoreOptions(failureStoreOptions).build()
+            );
+        }
         getIndexRequest.local(request.paramAsBoolean("local", getIndexRequest.local()));
         getIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getIndexRequest.masterNodeTimeout()));
         getIndexRequest.humanReadable(request.paramAsBoolean("human", false));
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
index 065399076c12a..f4d571cab0b8c 100644
--- a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
+++ b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.core.RestApiVersion;
@@ -87,6 +88,15 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final TimeValue timeout = request.paramAsTime("master_timeout", getMappingsRequest.masterNodeTimeout());
         getMappingsRequest.masterNodeTimeout(timeout);
         getMappingsRequest.local(request.paramAsBoolean("local", getMappingsRequest.local()));
+        if (DataStream.isFailureStoreEnabled()) {
+            IndicesOptions.FailureStoreOptions failureStoreOptions = IndicesOptions.FailureStoreOptions.fromRequest(
+                request,
+                getMappingsRequest.indicesOptions().failureStoreOptions()
+            );
+            getMappingsRequest.indicesOptions(
+                IndicesOptions.newBuilder(getMappingsRequest.indicesOptions()).failureStoreOptions(failureStoreOptions).build()
+            );
+        }
         final HttpChannel httpChannel = request.getHttpChannel();
         return channel -> new RestCancellableNodeClient(client, httpChannel).admin()
             .indices()
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
index af72e66f6127d..abe799255b62b 100644
--- a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
+++ b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
@@ -56,6 +57,15 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             .names(names);
         getSettingsRequest.local(request.paramAsBoolean("local", getSettingsRequest.local()));
         getSettingsRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getSettingsRequest.masterNodeTimeout()));
+        if (DataStream.isFailureStoreEnabled()) {
+            IndicesOptions.FailureStoreOptions failureStoreOptions = IndicesOptions.FailureStoreOptions.fromRequest(
+                request,
+                getSettingsRequest.indicesOptions().failureStoreOptions()
+            );
+            getSettingsRequest.indicesOptions(
+                IndicesOptions.newBuilder(getSettingsRequest.indicesOptions()).failureStoreOptions(failureStoreOptions).build()
+            );
+        }
         return channel -> client.admin().indices().getSettings(getSettingsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
index f0aa614d73677..673d7fa926211 100644
--- a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
+++ b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
@@ -14,6 +14,7 @@
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.core.RestApiVersion;
@@ -88,6 +89,15 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         assert indicesStatsRequest.indicesOptions() == IndicesOptions.strictExpandOpenAndForbidClosed()
             : "IndicesStats default indices options changed";
         indicesStatsRequest.indicesOptions(IndicesOptions.fromRequest(request, defaultIndicesOption));
+        if (DataStream.isFailureStoreEnabled()) {
+            IndicesOptions.FailureStoreOptions failureStore = IndicesOptions.FailureStoreOptions.fromRequest(
+                request,
+                indicesStatsRequest.indicesOptions().failureStoreOptions()
+            );
+            indicesStatsRequest.indicesOptions(
+                IndicesOptions.newBuilder(indicesStatsRequest.indicesOptions()).failureStoreOptions(failureStore).build()
+            );
+        }
         indicesStatsRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
         // level parameter validation
         ClusterStatsLevel.of(request, ClusterStatsLevel.INDICES);
diff --git a/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java b/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
index ccfcb9b505e92..1f636e36f1ba8 100644
--- a/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
+++ b/server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
@@ -78,7 +78,13 @@ protected void documentation(StringBuilder sb) {
     @Override
     public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
-        final IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, IndicesOptions.strictExpand());
+        final IndicesOptions.FailureStoreOptions failureStoreOptions = IndicesOptions.FailureStoreOptions.fromRequest(
+            request,
+            IndicesOptions.FailureStoreOptions.newBuilder().includeFailureIndices(true).build()
+        );
+        final IndicesOptions indicesOptions = IndicesOptions.newBuilder(IndicesOptions.fromRequest(request, IndicesOptions.strictExpand()))
+            .failureStoreOptions(failureStoreOptions)
+            .build();
         final TimeValue masterNodeTimeout = request.paramAsTime("master_timeout", DEFAULT_MASTER_NODE_TIMEOUT);
         final boolean includeUnloadedSegments = request.paramAsBoolean("include_unloaded_segments", false);
 
@@ -121,9 +127,11 @@ public RestResponse buildResponse(Void ignored) throws Exception {
                     .execute(listeners.acquire(indexSettingsRef::set));
 
                 // The other requests just provide additional detail, and wildcards may be resolved differently depending on the type of
-                // request in the presence of security plugins, so set the IndicesOptions for all the sub-requests to be as inclusive as
-                // possible.
-                final IndicesOptions subRequestIndicesOptions = IndicesOptions.lenientExpandHidden();
+                // request in the presence of security plugins, so set the IndicesOptions and DataStreamOptions for all the sub-requests
+                // to be as inclusive as possible.
+                final IndicesOptions subRequestIndicesOptions = IndicesOptions.newBuilder(IndicesOptions.lenientExpandHidden())
+                    .failureStoreOptions(failureStoreOptions)
+                    .build();
 
                 client.admin()
                     .cluster()
diff --git a/server/src/test/java/org/elasticsearch/action/support/IndicesOptionsTests.java b/server/src/test/java/org/elasticsearch/action/support/IndicesOptionsTests.java
index e1ee873865439..4a93f71fefd76 100644
--- a/server/src/test/java/org/elasticsearch/action/support/IndicesOptionsTests.java
+++ b/server/src/test/java/org/elasticsearch/action/support/IndicesOptionsTests.java
@@ -9,6 +9,7 @@
 package org.elasticsearch.action.support;
 
 import org.elasticsearch.action.support.IndicesOptions.ConcreteTargetOptions;
+import org.elasticsearch.action.support.IndicesOptions.FailureStoreOptions;
 import org.elasticsearch.action.support.IndicesOptions.GeneralOptions;
 import org.elasticsearch.action.support.IndicesOptions.WildcardOptions;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -41,17 +42,7 @@ public class IndicesOptionsTests extends ESTestCase {
     public void testSerialization() throws Exception {
         int iterations = randomIntBetween(5, 20);
         for (int i = 0; i < iterations; i++) {
-            IndicesOptions indicesOptions = IndicesOptions.fromOptions(
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean(),
-                randomBoolean()
-            );
+            IndicesOptions indicesOptions = randomIndicesOptions();
 
             BytesStreamOutput output = new BytesStreamOutput();
             indicesOptions.writeIndicesOptions(output);
@@ -72,6 +63,20 @@ public void testSerialization() throws Exception {
         }
     }
 
+    public static IndicesOptions randomIndicesOptions() {
+        return IndicesOptions.fromOptions(
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean()
+        );
+    }
+
     public void testFromOptions() {
         final boolean ignoreUnavailable = randomBoolean();
         final boolean allowNoIndices = randomBoolean();
@@ -113,17 +118,7 @@ public void testFromOptionsWithDefaultOptions() {
         boolean expandToOpenIndices = randomBoolean();
         boolean expandToClosedIndices = randomBoolean();
 
-        IndicesOptions defaultOptions = IndicesOptions.fromOptions(
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean()
-        );
+        IndicesOptions defaultOptions = randomIndicesOptions();
 
         IndicesOptions indicesOptions = IndicesOptions.fromOptions(
             ignoreUnavailable,
@@ -176,17 +171,7 @@ public void testFromParameters() {
         boolean allowNoIndices = randomBoolean();
         String allowNoIndicesString = Boolean.toString(allowNoIndices);
 
-        IndicesOptions defaultOptions = IndicesOptions.fromOptions(
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean()
-        );
+        IndicesOptions defaultOptions = randomIndicesOptions();
 
         IndicesOptions updatedOptions = IndicesOptions.fromParameters(
             expandWildcardsString,
@@ -207,17 +192,7 @@ public void testFromParameters() {
     }
 
     public void testEqualityAndHashCode() {
-        IndicesOptions indicesOptions = IndicesOptions.fromOptions(
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean(),
-            randomBoolean()
-        );
+        IndicesOptions indicesOptions = randomIndicesOptions();
 
         EqualsHashCodeTestUtils.checkEqualsAndHashCode(
             indicesOptions,
@@ -345,8 +320,8 @@ public void testToXContent() throws IOException {
             randomBoolean()
         );
         GeneralOptions generalOptions = new GeneralOptions(randomBoolean(), randomBoolean(), randomBoolean());
-
-        IndicesOptions indicesOptions = new IndicesOptions(concreteTargetOptions, wildcardOptions, generalOptions);
+        FailureStoreOptions failureStoreOptions = new FailureStoreOptions(randomBoolean(), randomBoolean());
+        IndicesOptions indicesOptions = new IndicesOptions(concreteTargetOptions, wildcardOptions, generalOptions, failureStoreOptions);
 
         XContentType type = randomFrom(XContentType.values());
         BytesReference xContentBytes = toXContentBytes(indicesOptions, type);
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
index e72bab96e54a1..c097da694323e 100644
--- a/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
@@ -53,6 +53,7 @@
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.backingIndexEqualTo;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createBackingIndex;
+import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createFailureIndex;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_HIDDEN_SETTING;
 import static org.elasticsearch.common.util.set.Sets.newHashSet;
@@ -2291,11 +2292,7 @@ public void testIgnoreThrottled() {
         {
             Index[] indices = indexNameExpressionResolver.concreteIndices(
                 state,
-                new IndicesOptions(
-                    IndicesOptions.ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
-                    IndicesOptions.WildcardOptions.DEFAULT,
-                    IndicesOptions.GeneralOptions.newBuilder().removeThrottled(true).build()
-                ),
+                IndicesOptions.newBuilder().generalOptions(IndicesOptions.GeneralOptions.newBuilder().removeThrottled(true)).build(),
                 "ind*",
                 "test-index"
             );
@@ -2697,6 +2694,173 @@ public void testDataStreams() {
         }
     }
 
+    public void testDataStreamsWithFailureStore() {
+        final String dataStreamName = "my-data-stream";
+        IndexMetadata index1 = createBackingIndex(dataStreamName, 1, epochMillis).build();
+        IndexMetadata index2 = createBackingIndex(dataStreamName, 2, epochMillis).build();
+        IndexMetadata failureIndex1 = createFailureIndex(dataStreamName, 1, epochMillis).build();
+        IndexMetadata failureIndex2 = createFailureIndex(dataStreamName, 2, epochMillis).build();
+        IndexMetadata otherIndex = indexBuilder("my-other-index", Settings.EMPTY).state(State.OPEN).build();
+
+        Metadata.Builder mdBuilder = Metadata.builder()
+            .put(index1, false)
+            .put(index2, false)
+            .put(failureIndex1, false)
+            .put(failureIndex2, false)
+            .put(otherIndex, false)
+            .put(
+                newInstance(
+                    dataStreamName,
+                    List.of(index1.getIndex(), index2.getIndex()),
+                    List.of(failureIndex1.getIndex(), failureIndex2.getIndex())
+                )
+            );
+        ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(mdBuilder).build();
+
+        // Test default with an exact data stream name
+        {
+            IndicesOptions indicesOptions = IndicesOptions.STRICT_EXPAND_OPEN;
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-data-stream");
+            assertThat(result.length, equalTo(2));
+            assertThat(result[0].getName(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis)));
+            assertThat(result[1].getName(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis)));
+        }
+
+        // Test include failure store with an exact data stream name
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(IndicesOptions.FailureStoreOptions.newBuilder().includeFailureIndices(true))
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-data-stream");
+            assertThat(result.length, equalTo(4));
+            assertThat(result[0].getName(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis)));
+            assertThat(result[1].getName(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis)));
+            assertThat(result[2].getName(), equalTo(DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis)));
+            assertThat(result[3].getName(), equalTo(DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis)));
+        }
+
+        // Test only failure store with an exact data stream name
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(IndicesOptions.FailureStoreOptions.newBuilder().includeBackingIndices(false))
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-data-stream");
+            assertThat(result.length, equalTo(2));
+            assertThat(result[0].getName(), equalTo(DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis)));
+            assertThat(result[1].getName(), equalTo(DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis)));
+        }
+
+        // Test default without any expressions
+        {
+            IndicesOptions indicesOptions = IndicesOptions.STRICT_EXPAND_OPEN;
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true);
+            assertThat(result.length, equalTo(3));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis),
+                    otherIndex.getIndex().getName()
+                )
+            );
+        }
+
+        // Test include failure store without any expressions
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(IndicesOptions.FailureStoreOptions.newBuilder().includeFailureIndices(true))
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true);
+            assertThat(result.length, equalTo(5));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis),
+                    otherIndex.getIndex().getName()
+                )
+            );
+        }
+
+        // Test only failure store without any expressions
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(
+                    IndicesOptions.FailureStoreOptions.newBuilder().includeBackingIndices(false).includeFailureIndices(true).build()
+                )
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true);
+            assertThat(result.length, equalTo(2));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis)
+                )
+            );
+        }
+
+        // Test default with wildcard expression
+        {
+            IndicesOptions indicesOptions = IndicesOptions.STRICT_EXPAND_OPEN;
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-*");
+            assertThat(result.length, equalTo(3));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis),
+                    otherIndex.getIndex().getName()
+                )
+            );
+        }
+
+        // Test include failure store with wildcard expression
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(IndicesOptions.FailureStoreOptions.newBuilder().includeFailureIndices(true))
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-*");
+            assertThat(result.length, equalTo(5));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultBackingIndexName(dataStreamName, 1, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis),
+                    otherIndex.getIndex().getName()
+                )
+            );
+        }
+
+        // Test only failure store with wildcard expression
+        {
+            IndicesOptions indicesOptions = IndicesOptions.newBuilder()
+                .failureStoreOptions(
+                    IndicesOptions.FailureStoreOptions.newBuilder().includeBackingIndices(false).includeFailureIndices(true)
+                )
+                .build();
+            Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-*");
+            assertThat(result.length, equalTo(2));
+            List<String> indexNames = Arrays.stream(result).map(Index::getName).toList();
+            assertThat(
+                indexNames,
+                containsInAnyOrder(
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 2, epochMillis),
+                    DataStream.getDefaultFailureStoreName(dataStreamName, 1, epochMillis)
+                )
+            );
+        }
+    }
+
     public void testDataStreamAliases() {
         String dataStream1 = "my-data-stream-1";
         IndexMetadata index1 = createBackingIndex(dataStream1, 1, epochMillis).build();
diff --git a/test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java b/test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java
index d0b30bff92f3e..8f9d68176fdda 100644
--- a/test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java
+++ b/test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java
@@ -90,6 +90,10 @@ public static DataStream newInstance(String name, List<Index> indices) {
         return newInstance(name, indices, indices.size(), null);
     }
 
+    public static DataStream newInstance(String name, List<Index> indices, List<Index> failureIndices) {
+        return newInstance(name, indices, indices.size(), null, false, null, failureIndices);
+    }
+
     public static DataStream newInstance(String name, List<Index> indices, long generation, Map<String, Object> metadata) {
         return newInstance(name, indices, generation, metadata, false);
     }
@@ -124,7 +128,20 @@ public static DataStream newInstance(
         @Nullable DataStreamLifecycle lifecycle,
         List<Index> failureStores
     ) {
-        return new DataStream(name, indices, generation, metadata, false, replicated, false, false, null, lifecycle, false, failureStores);
+        return new DataStream(
+            name,
+            indices,
+            generation,
+            metadata,
+            false,
+            replicated,
+            false,
+            false,
+            null,
+            lifecycle,
+            failureStores.isEmpty() == false,
+            failureStores
+        );
     }
 
     public static String getLegacyDefaultBackingIndexName(
@@ -169,6 +186,13 @@ public static IndexMetadata.Builder createBackingIndex(String dataStreamName, in
             .numberOfReplicas(NUMBER_OF_REPLICAS);
     }
 
+    public static IndexMetadata.Builder createFailureIndex(String dataStreamName, int generation, long epochMillis) {
+        return IndexMetadata.builder(DataStream.getDefaultFailureStoreName(dataStreamName, generation, epochMillis))
+            .settings(SETTINGS)
+            .numberOfShards(NUMBER_OF_SHARDS)
+            .numberOfReplicas(NUMBER_OF_REPLICAS);
+    }
+
     public static IndexMetadata.Builder getIndexMetadataBuilderForIndex(Index index) {
         return IndexMetadata.builder(index.getName())
             .settings(Settings.builder().put(SETTINGS.build()).put(SETTING_INDEX_UUID, index.getUUID()))