diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamFailureStore.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamFailureStore.java
new file mode 100644
index 0000000000000..d647956e752a3
--- /dev/null
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamFailureStore.java
@@ -0,0 +1,76 @@
+/*
+ * 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.cluster.metadata;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.SimpleDiffable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+/**
+ * Holds the data stream failure store metadata that enable or disable the failure store of a data stream. Currently, it
+ * supports the following configurations:
+ * - enabled
+ */
+public record DataStreamFailureStore(boolean enabled) implements SimpleDiffable<DataStreamFailureStore>, ToXContentObject {
+
+    public static final ParseField ENABLED_FIELD = new ParseField("enabled");
+
+    public static final ConstructingObjectParser<DataStreamFailureStore, Void> PARSER = new ConstructingObjectParser<>(
+        "failure_store",
+        false,
+        (args, unused) -> new DataStreamFailureStore(args[0] == null || (Boolean) args[0])
+    );
+
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), ENABLED_FIELD);
+    }
+
+    public DataStreamFailureStore() {
+        this(true);
+    }
+
+    public DataStreamFailureStore(StreamInput in) throws IOException {
+        this(in.readBoolean());
+    }
+
+    public static Diff<DataStreamFailureStore> readDiffFrom(StreamInput in) throws IOException {
+        return SimpleDiffable.readDiffFrom(DataStreamFailureStore::new, in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeBoolean(enabled);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this, true, true);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(ENABLED_FIELD.getPreferredName(), enabled);
+        builder.endObject();
+        return builder;
+    }
+
+    public static DataStreamFailureStore fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamOptions.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamOptions.java
new file mode 100644
index 0000000000000..9c7d2a986fa48
--- /dev/null
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamOptions.java
@@ -0,0 +1,93 @@
+/*
+ * 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.cluster.metadata;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.SimpleDiffable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+/**
+ * Holds data stream dedicated configuration options such as failure store, (in the future lifecycle). Currently, it
+ * supports the following configurations:
+ * - failure store
+ */
+public record DataStreamOptions(@Nullable DataStreamFailureStore failureStore)
+    implements
+        SimpleDiffable<DataStreamOptions>,
+        ToXContentObject {
+
+    public static final ParseField FAILURE_STORE_FIELD = new ParseField("failure_store");
+
+    public static final ConstructingObjectParser<DataStreamOptions, Void> PARSER = new ConstructingObjectParser<>(
+        "options",
+        false,
+        (args, unused) -> new DataStreamOptions((DataStreamFailureStore) args[0])
+    );
+
+    static {
+        PARSER.declareField(
+            ConstructingObjectParser.optionalConstructorArg(),
+            (p, c) -> DataStreamFailureStore.fromXContent(p),
+            FAILURE_STORE_FIELD,
+            ObjectParser.ValueType.OBJECT_OR_NULL
+        );
+    }
+
+    public DataStreamOptions() {
+        this(null);
+    }
+
+    public static DataStreamOptions read(StreamInput in) throws IOException {
+        return new DataStreamOptions(in.readOptionalWriteable(DataStreamFailureStore::new));
+    }
+
+    @Nullable
+    public DataStreamFailureStore getFailureStore() {
+        return failureStore;
+    }
+
+    public static Diff<DataStreamOptions> readDiffFrom(StreamInput in) throws IOException {
+        return SimpleDiffable.readDiffFrom(DataStreamOptions::read, in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeOptionalWriteable(failureStore);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this, true, true);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (failureStore != null) {
+            builder.field(FAILURE_STORE_FIELD.getPreferredName(), failureStore);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public static DataStreamOptions fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamFailureStoreTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamFailureStoreTests.java
new file mode 100644
index 0000000000000..f5334f903af6b
--- /dev/null
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamFailureStoreTests.java
@@ -0,0 +1,42 @@
+/*
+ * 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.cluster.metadata;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class DataStreamFailureStoreTests extends AbstractXContentSerializingTestCase<DataStreamFailureStore> {
+
+    @Override
+    protected Writeable.Reader<DataStreamFailureStore> instanceReader() {
+        return DataStreamFailureStore::new;
+    }
+
+    @Override
+    protected DataStreamFailureStore createTestInstance() {
+        return randomFailureStore();
+    }
+
+    @Override
+    protected DataStreamFailureStore mutateInstance(DataStreamFailureStore instance) throws IOException {
+        return new DataStreamFailureStore(instance.enabled() == false);
+    }
+
+    @Override
+    protected DataStreamFailureStore doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamFailureStore.fromXContent(parser);
+    }
+
+    static DataStreamFailureStore randomFailureStore() {
+        return new DataStreamFailureStore(randomBoolean());
+    }
+}
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamOptionsTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamOptionsTests.java
new file mode 100644
index 0000000000000..8a7cf2329b863
--- /dev/null
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamOptionsTests.java
@@ -0,0 +1,44 @@
+/*
+ * 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.cluster.metadata;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class DataStreamOptionsTests extends AbstractXContentSerializingTestCase<DataStreamOptions> {
+
+    @Override
+    protected Writeable.Reader<DataStreamOptions> instanceReader() {
+        return DataStreamOptions::read;
+    }
+
+    @Override
+    protected DataStreamOptions createTestInstance() {
+        return new DataStreamOptions(randomBoolean() ? null : DataStreamFailureStoreTests.randomFailureStore());
+    }
+
+    @Override
+    protected DataStreamOptions mutateInstance(DataStreamOptions instance) throws IOException {
+        var failureStore = instance.getFailureStore();
+        if (failureStore == null) {
+            failureStore = DataStreamFailureStoreTests.randomFailureStore();
+        } else {
+            failureStore = randomBoolean() ? null : new DataStreamFailureStore(failureStore.enabled() == false);
+        }
+        return new DataStreamOptions(failureStore);
+    }
+
+    @Override
+    protected DataStreamOptions doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamOptions.fromXContent(parser);
+    }
+}