Allow for the data lifecycle to be explicitly nullified (#95979)

This PR allows a user of an index template to explicitly reset to `null` either the whole data lifecycle config or just the retention of a component template and overwrite what previous component templates have set.

docs/changelog/95979.yaml

@@ -0,0 +1,5 @@
+pr: 95979
+summary: Allow for the data lifecycle and the retention to be explicitly nullified
+area: DLM
+type: enhancement
+issues: []

modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java

@@ -11,6 +11,7 @@ package org.elasticsearch.datastreams;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.metadata.Template;
@@ -32,11 +33,13 @@ import java.util.List;
 import java.util.Set;
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.generateTsdbMapping;
+import static org.elasticsearch.cluster.metadata.MetadataIndexTemplateService.composeDataLifecycles;
 import static org.elasticsearch.common.settings.Settings.builder;
 import static org.elasticsearch.datastreams.MetadataDataStreamRolloverServiceTests.createSettingsProvider;
 import static org.elasticsearch.indices.ShardLimitValidator.SETTING_CLUSTER_MAX_SHARDS_PER_NODE;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -140,6 +143,47 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         }
     }
 
+    public void testLifecycleComposition() {
+        // No lifecycles result to null
+        {
+            List<DataLifecycle> lifecycles = List.of();
+            assertThat(composeDataLifecycles(lifecycles), nullValue());
+        }
+        // One lifecycle results to this lifecycle as the final
+        {
+            DataLifecycle lifecycle = switch (randomInt(2)) {
+                case 0 -> new DataLifecycle();
+                case 1 -> new DataLifecycle(DataLifecycle.Retention.NULL);
+                default -> new DataLifecycle(randomMillisUpToYear9999());
+            };
+            List<DataLifecycle> lifecycles = List.of(lifecycle);
+            assertThat(composeDataLifecycles(lifecycles), equalTo(lifecycle));
+        }
+        // If the last lifecycle is missing a property we keep the latest from the previous ones
+        {
+            DataLifecycle lifecycleWithRetention = new DataLifecycle(randomMillisUpToYear9999());
+            List<DataLifecycle> lifecycles = List.of(lifecycleWithRetention, new DataLifecycle());
+            assertThat(
+                composeDataLifecycles(lifecycles).getEffectiveDataRetention(),
+                equalTo(lifecycleWithRetention.getEffectiveDataRetention())
+            );
+        }
+        // If both lifecycle have all properties, then the latest one overwrites all the others
+        {
+            DataLifecycle lifecycle1 = new DataLifecycle(randomMillisUpToYear9999());
+            DataLifecycle lifecycle2 = new DataLifecycle(randomMillisUpToYear9999());
+            List<DataLifecycle> lifecycles = List.of(lifecycle1, lifecycle2);
+            assertThat(composeDataLifecycles(lifecycles), equalTo(lifecycle2));
+        }
+        // If the last lifecycle is explicitly null, the result is also null
+        {
+            DataLifecycle lifecycle1 = new DataLifecycle(randomMillisUpToYear9999());
+            DataLifecycle lifecycle2 = new DataLifecycle(randomMillisUpToYear9999());
+            List<DataLifecycle> lifecycles = List.of(lifecycle1, lifecycle2, Template.NO_LIFECYCLE);
+            assertThat(composeDataLifecycles(lifecycles), nullValue());
+        }
+    }
+
     private MetadataIndexTemplateService getMetadataIndexTemplateService() {
         var indicesService = getInstanceFromNode(IndicesService.class);
         var clusterService = getInstanceFromNode(ClusterService.class);

modules/dlm/src/internalClusterTest/java/org/elasticsearch/dlm/CrudDataLifecycleIT.java

@@ -10,6 +10,8 @@ package org.elasticsearch.dlm;
 
 import org.elasticsearch.action.datastreams.CreateDataStreamAction;
 import org.elasticsearch.cluster.metadata.DataLifecycle;
+import org.elasticsearch.cluster.metadata.Template;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.datastreams.DataStreamsPlugin;
 import org.elasticsearch.dlm.action.DeleteDataLifecycleAction;
 import org.elasticsearch.dlm.action.GetDataLifecycleAction;
@@ -65,11 +67,11 @@ public class CrudDataLifecycleIT extends ESIntegTestCase {
             GetDataLifecycleAction.Response response = client().execute(GetDataLifecycleAction.INSTANCE, getDataLifecycleRequest).get();
             assertThat(response.getDataStreamLifecycles().size(), equalTo(3));
             assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
-            assertThat(response.getDataStreamLifecycles().get(0).lifecycle(), equalTo(lifecycle));
+            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
             assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
-            assertThat(response.getDataStreamLifecycles().get(1).lifecycle(), equalTo(lifecycle));
+            assertDataLifecycle(response.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
             assertThat(response.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("without-lifecycle"));
-            assertThat(response.getDataStreamLifecycles().get(2).lifecycle(), is(nullValue()));
+            assertThat(response.getDataStreamLifecycles().get(2).lifecycle(), nullValue());
             assertThat(response.getRolloverConfiguration(), nullValue());
         }
 
@@ -79,9 +81,9 @@ public class CrudDataLifecycleIT extends ESIntegTestCase {
             GetDataLifecycleAction.Response response = client().execute(GetDataLifecycleAction.INSTANCE, getDataLifecycleRequest).get();
             assertThat(response.getDataStreamLifecycles().size(), equalTo(2));
             assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
-            assertThat(response.getDataStreamLifecycles().get(0).lifecycle(), equalTo(lifecycle));
+            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
             assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
-            assertThat(response.getDataStreamLifecycles().get(1).lifecycle(), equalTo(lifecycle));
+            assertDataLifecycle(response.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
             assertThat(response.getRolloverConfiguration(), nullValue());
         }
 
@@ -93,7 +95,7 @@ public class CrudDataLifecycleIT extends ESIntegTestCase {
             GetDataLifecycleAction.Response response = client().execute(GetDataLifecycleAction.INSTANCE, getDataLifecycleRequest).get();
             assertThat(response.getDataStreamLifecycles().size(), equalTo(2));
             assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
-            assertThat(response.getDataStreamLifecycles().get(0).lifecycle(), equalTo(lifecycle));
+            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
             assertThat(response.getRolloverConfiguration(), nullValue());
         }
 
@@ -106,14 +108,22 @@ public class CrudDataLifecycleIT extends ESIntegTestCase {
         ).get();
         assertThat(responseWithRollover.getDataStreamLifecycles().size(), equalTo(3));
         assertThat(responseWithRollover.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
-        assertThat(responseWithRollover.getDataStreamLifecycles().get(0).lifecycle(), equalTo(lifecycle));
+        assertDataLifecycle(responseWithRollover.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
         assertThat(responseWithRollover.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
-        assertThat(responseWithRollover.getDataStreamLifecycles().get(1).lifecycle(), equalTo(lifecycle));
+        assertDataLifecycle(responseWithRollover.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
         assertThat(responseWithRollover.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("without-lifecycle"));
         assertThat(responseWithRollover.getDataStreamLifecycles().get(2).lifecycle(), is(nullValue()));
         assertThat(responseWithRollover.getRolloverConfiguration(), notNullValue());
     }
 
+    private void assertDataLifecycle(DataLifecycle dataLifecycle, DataLifecycle expected) {
+        if (expected.equals(Template.NO_LIFECYCLE)) {
+            assertThat(dataLifecycle, nullValue());
+        } else {
+            assertThat(dataLifecycle, equalTo(expected));
+        }
+    }
+
     public void testPutLifecycle() throws Exception {
         putComposableIndexTemplate("id1", null, List.of("my-data-stream*"), null, null, null);
         // Create index without a lifecycle
@@ -132,17 +142,17 @@ public class CrudDataLifecycleIT extends ESIntegTestCase {
 
         // Set lifecycle
         {
-            DataLifecycle lifecycle = randomDataLifecycle();
+            TimeValue dataRetention = randomBoolean() ? null : TimeValue.timeValueMillis(randomMillisUpToYear9999());
             PutDataLifecycleAction.Request putDataLifecycleRequest = new PutDataLifecycleAction.Request(
                 new String[] { "*" },
-                lifecycle.getDataRetention()
+                dataRetention
             );
             assertThat(client().execute(PutDataLifecycleAction.INSTANCE, putDataLifecycleRequest).get().isAcknowledged(), equalTo(true));
             GetDataLifecycleAction.Request getDataLifecycleRequest = new GetDataLifecycleAction.Request(new String[] { "my-data-stream" });
             GetDataLifecycleAction.Response response = client().execute(GetDataLifecycleAction.INSTANCE, getDataLifecycleRequest).get();
             assertThat(response.getDataStreamLifecycles().size(), equalTo(1));
             assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("my-data-stream"));
-            assertThat(response.getDataStreamLifecycles().get(0).lifecycle(), equalTo(lifecycle));
+            assertThat(response.getDataStreamLifecycles().get(0).lifecycle().getEffectiveDataRetention(), equalTo(dataRetention));
         }
     }
 

modules/dlm/src/internalClusterTest/java/org/elasticsearch/dlm/ExplainDataLifecycleIT.java

@@ -119,7 +119,7 @@ public class ExplainDataLifecycleIT extends ESIntegTestCase {
                 assertThat(explainIndex.isManagedByDLM(), is(true));
                 assertThat(explainIndex.getIndexCreationDate(), notNullValue());
                 assertThat(explainIndex.getLifecycle(), notNullValue());
-                assertThat(explainIndex.getLifecycle().getDataRetention(), nullValue());
+                assertThat(explainIndex.getLifecycle().getEffectiveDataRetention(), nullValue());
                 if (internalCluster().numDataNodes() > 1) {
                     // If the number of nodes is 1 then the cluster will be yellow so forcemerge will report an error if it has run
                     assertThat(explainIndex.getError(), nullValue());
@@ -208,7 +208,7 @@ public class ExplainDataLifecycleIT extends ESIntegTestCase {
                 assertThat(explainIndex.isManagedByDLM(), is(true));
                 assertThat(explainIndex.getIndexCreationDate(), notNullValue());
                 assertThat(explainIndex.getLifecycle(), notNullValue());
-                assertThat(explainIndex.getLifecycle().getDataRetention(), nullValue());
+                assertThat(explainIndex.getLifecycle().getEffectiveDataRetention(), nullValue());
                 assertThat(explainIndex.getRolloverDate(), nullValue());
                 assertThat(explainIndex.getTimeSinceRollover(System::currentTimeMillis), nullValue());
                 // index has not been rolled over yet

modules/dlm/src/main/java/org/elasticsearch/dlm/DataLifecycleService.java

@@ -290,7 +290,7 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
             RolloverRequest rolloverRequest = getDefaultRolloverRequest(
                 rolloverConfiguration,
                 dataStream.getName(),
-                dataStream.getLifecycle().getDataRetention()
+                dataStream.getLifecycle().getEffectiveDataRetention()
             );
             transportActionsDeduplicator.executeOnce(
                 rolloverRequest,
@@ -525,7 +525,7 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
         if (dataStream.getLifecycle() == null) {
             return null;
         }
-        return dataStream.getLifecycle().getDataRetention();
+        return dataStream.getLifecycle().getEffectiveDataRetention();
     }
 
     /**

modules/dlm/src/test/java/org/elasticsearch/dlm/DLMFixtures.java

@@ -28,9 +28,9 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.lucene.tests.util.LuceneTestCase.rarely;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
 import static org.elasticsearch.test.ESIntegTestCase.client;
+import static org.elasticsearch.test.ESTestCase.randomInt;
 import static org.elasticsearch.test.ESTestCase.randomIntBetween;
 import static org.junit.Assert.assertTrue;
 
@@ -94,6 +94,11 @@ public class DLMFixtures {
     }
 
     static DataLifecycle randomDataLifecycle() {
-        return rarely() ? new DataLifecycle() : new DataLifecycle(TimeValue.timeValueDays(randomIntBetween(1, 365)));
+        return switch (randomInt(3)) {
+            case 0 -> new DataLifecycle();
+            case 1 -> new DataLifecycle(DataLifecycle.Retention.NULL);
+            case 2 -> Template.NO_LIFECYCLE;
+            default -> new DataLifecycle(TimeValue.timeValueDays(randomIntBetween(1, 365)));
+        };
     }
 }

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -127,12 +127,13 @@ public record TransportVersion(int id) implements Comparable<TransportVersion> {
     public static final TransportVersion V_8_500_004 = registerTransportVersion(8_500_004, "6a00db6a-fd66-42a9-97ea-f6cc53169110");
     public static final TransportVersion V_8_500_005 = registerTransportVersion(8_500_005, "65370d2a-d936-4383-a2e0-8403f708129b");
     public static final TransportVersion V_8_500_006 = registerTransportVersion(8_500_006, "7BB5621A-80AC-425F-BA88-75543C442F23");
+    public static final TransportVersion V_8_500_007 = registerTransportVersion(8_500_007, "77261d43-4149-40af-89c5-7e71e0454fce");
 
     /**
      * Reference to the most recent transport version.
      * This should be the transport version with the highest id.
      */
-    public static final TransportVersion CURRENT = V_8_500_006;
+    public static final TransportVersion CURRENT = V_8_500_007;
 
     /**
      * Reference to the earliest compatible transport version to this version of the codebase.

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

@@ -8,12 +8,14 @@
 
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.admin.indices.rollover.RolloverConfiguration;
 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.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.core.Nullable;
@@ -26,7 +28,6 @@ import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
-import java.util.List;
 import java.util.Objects;
 
 /**
@@ -44,7 +45,6 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
 
     private static final FeatureFlag DLM_FEATURE_FLAG = new FeatureFlag("dlm");
 
-    public static final DataLifecycle EMPTY = new DataLifecycle();
     public static final String DLM_ORIGIN = "data_lifecycle";
 
     public static final ParseField DATA_RETENTION_FIELD = new ParseField("data_retention");
@@ -53,16 +53,18 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
     public static final ConstructingObjectParser<DataLifecycle, Void> PARSER = new ConstructingObjectParser<>(
         "lifecycle",
         false,
-        (args, unused) -> new DataLifecycle((TimeValue) args[0])
+        (args, unused) -> new DataLifecycle((Retention) args[0])
     );
 
     static {
-        PARSER.declareField(
-            ConstructingObjectParser.optionalConstructorArg(),
-            (p, c) -> TimeValue.parseTimeValue(p.textOrNull(), DATA_RETENTION_FIELD.getPreferredName()),
-            DATA_RETENTION_FIELD,
-            ObjectParser.ValueType.STRING_OR_NULL
-        );
+        PARSER.declareField(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> {
+            String value = p.textOrNull();
+            if (value == null) {
+                return Retention.NULL;
+            } else {
+                return new Retention(TimeValue.parseTimeValue(value, DATA_RETENTION_FIELD.getPreferredName()));
+            }
+        }, DATA_RETENTION_FIELD, ObjectParser.ValueType.STRING_OR_NULL);
     }
 
     public static boolean isEnabled() {
@@ -70,13 +72,17 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
     }
 
     @Nullable
-    private final TimeValue dataRetention;
+    private final Retention dataRetention;
 
     public DataLifecycle() {
-        this.dataRetention = null;
+        this((TimeValue) null);
     }
 
     public DataLifecycle(@Nullable TimeValue dataRetention) {
+        this(dataRetention == null ? null : new Retention(dataRetention));
+    }
+
+    public DataLifecycle(@Nullable Retention dataRetention) {
         this.dataRetention = dataRetention;
     }
 
@@ -85,55 +91,25 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
     }
 
     /**
-     * This method composes a series of lifecycles to a final one. The lifecycles are getting composed one level deep,
-     * meaning that the keys present on the latest lifecycle will override the ones of the others. If a key is missing
-     * then it keeps the value of the previous lifecycles. For example, if we have the following two lifecycles:
-     * [
-     *   {
-     *     "lifecycle": {
-     *       "data_retention" : "10d"
-     *     }
-     *   },
-     *   {
-     *     "lifecycle": {
-     *       "data_retention" : "20d"
-     *     }
-     *   }
-     * ]
-     * The result will be { "lifecycle": { "data_retention" : "20d"}} because the second data retention overrides the first.
-     * However, if we have the following two lifecycles:
-     * [
-     *   {
-     *     "lifecycle": {
-     *       "data_retention" : "10d"
-     *     }
-     *   },
-     *   {
-     *   "lifecycle": { }
-     *   }
-     * ]
-     * The result will be { "lifecycle": { "data_retention" : "10d"} } because the latest lifecycle does not have any
-     * information on retention.
-     * @param lifecycles a sorted list of lifecycles in the order that they will be composed
-     * @return the final lifecycle
+     * The least amount of time data should be kept by elasticsearch.
+     * @return the time period or null, null represents that data should never be deleted.
      */
     @Nullable
-    public static DataLifecycle compose(List<DataLifecycle> lifecycles) {
-        DataLifecycle.Builder builder = null;
-        for (DataLifecycle current : lifecycles) {
-            if (builder == null) {
-                builder = Builder.newBuilder(current);
-            } else {
-                if (current.dataRetention != null) {
-                    builder.dataRetention(current.getDataRetention());
-                }
-            }
-        }
-        return builder == null ? null : builder.build();
+    public TimeValue getEffectiveDataRetention() {
+        return dataRetention == null ? null : dataRetention.value;
     }
 
+    /**
+     * The configuration as provided by the user about the least amount of time data should be kept by elasticsearch.
+     * This method differentiates between a missing retention and a nullified retention and this is useful for template
+     * composition.
+     * @return one of the following:
+     * - `null`, represents that the user did not provide data retention, this represents the user has no opinion about retention
+     * - `Retention{value = null}`, represents that the user explicitly wants to have infinite retention
+     * - `Retention{value = "10d"}`, represents that the user has requested the data to be kept at least 10d.
+     */
     @Nullable
-    public TimeValue getDataRetention() {
+    Retention getDataRetention() {
         return dataRetention;
     }
 
@@ -148,16 +124,29 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
 
     @Override
     public int hashCode() {
-        return Objects.hashCode(dataRetention);
+        return Objects.hash(dataRetention);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        out.writeOptionalTimeValue(dataRetention);
+        if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_007)) {
+            out.writeOptionalWriteable(dataRetention);
+        } else {
+            out.writeOptionalTimeValue(getEffectiveDataRetention());
+        }
     }
 
     public DataLifecycle(StreamInput in) throws IOException {
-        dataRetention = in.readOptionalTimeValue();
+        if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_007)) {
+            dataRetention = in.readOptionalWriteable(Retention::read);
+        } else {
+            var value = in.readOptionalTimeValue();
+            if (value == null) {
+                dataRetention = null;
+            } else {
+                dataRetention = new Retention(value);
+            }
+        }
     }
 
     public static Diff<DataLifecycle> readDiffFrom(StreamInput in) throws IOException {
@@ -181,11 +170,15 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
         throws IOException {
         builder.startObject();
         if (dataRetention != null) {
-            builder.field(DATA_RETENTION_FIELD.getPreferredName(), dataRetention.getStringRep());
+            if (dataRetention.value() == null) {
+                builder.nullField(DATA_RETENTION_FIELD.getPreferredName());
+            } else {
+                builder.field(DATA_RETENTION_FIELD.getPreferredName(), dataRetention.value().getStringRep());
+            }
         }
         if (rolloverConfiguration != null) {
             builder.field(ROLLOVER_FIELD.getPreferredName());
-            rolloverConfiguration.evaluateAndConvertToXContent(builder, params, dataRetention);
+            rolloverConfiguration.evaluateAndConvertToXContent(builder, params, getEffectiveDataRetention());
         }
         builder.endObject();
         return builder;
@@ -200,9 +193,9 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
      */
     static class Builder {
         @Nullable
-        private TimeValue dataRetention = null;
+        private Retention dataRetention = null;
 
-        Builder dataRetention(@Nullable TimeValue value) {
+        Builder dataRetention(@Nullable Retention value) {
             dataRetention = value;
             return this;
         }
@@ -215,4 +208,23 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
             return new Builder().dataRetention(dataLifecycle.getDataRetention());
         }
     }
+
+    /**
+     * Retention is the least amount of time that the data will be kept by elasticsearch. Public for testing.
+     * @param value is a time period or null. Null represents an explicitly set infinite retention period
+     */
+    public record Retention(@Nullable TimeValue value) implements Writeable {
+
+        // For testing
+        public static final Retention NULL = new Retention(null);
+
+        public static Retention read(StreamInput in) throws IOException {
+            return new Retention(in.readOptionalTimeValue());
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeOptionalTimeValue(value);
+        }
+    }
 }

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

@@ -613,12 +613,12 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
      * is treated differently for the write index (i.e. they first need to be rolled over)
      */
     public List<Index> getIndicesPastRetention(Function<String, IndexMetadata> indexMetadataSupplier, LongSupplier nowSupplier) {
-        if (lifecycle == null || lifecycle.getDataRetention() == null) {
+        if (lifecycle == null || lifecycle.getEffectiveDataRetention() == null) {
             return List.of();
         }
 
         List<Index> indicesPastRetention = getNonWriteIndicesOlderThan(
-            lifecycle.getDataRetention(),
+            lifecycle.getEffectiveDataRetention(),
             indexMetadataSupplier,
             this::isIndexManagedByDLM,
             nowSupplier

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

@@ -1468,7 +1468,57 @@ public class MetadataIndexTemplateService {
         if (template.template() != null && template.template().lifecycle() != null) {
             lifecycles.add(template.template().lifecycle());
         }
-        return DataLifecycle.compose(lifecycles);
+        return composeDataLifecycles(lifecycles);
+    }
+
+    /**
+     * This method composes a series of lifecycles to a final one. The lifecycles are getting composed one level deep,
+     * meaning that the keys present on the latest lifecycle will override the ones of the others. If a key is missing
+     * then it keeps the value of the previous lifecycles. For example, if we have the following two lifecycles:
+     * [
+     *   {
+     *     "lifecycle": {
+     *       "data_retention" : "10d"
+     *     }
+     *   },
+     *   {
+     *     "lifecycle": {
+     *       "data_retention" : "20d"
+     *     }
+     *   }
+     * ]
+     * The result will be { "lifecycle": { "data_retention" : "20d"}} because the second data retention overrides the first.
+     * However, if we have the following two lifecycles:
+     * [
+     *   {
+     *     "lifecycle": {
+     *       "data_retention" : "10d"
+     *     }
+     *   },
+     *   {
+     *   "lifecycle": { }
+     *   }
+     * ]
+     * The result will be { "lifecycle": { "data_retention" : "10d"} } because the latest lifecycle does not have any
+     * information on retention.
+     * @param lifecycles a sorted list of lifecycles in the order that they will be composed
+     * @return the final lifecycle
+     */
+    @Nullable
+    public static DataLifecycle composeDataLifecycles(List<DataLifecycle> lifecycles) {
+        DataLifecycle.Builder builder = null;
+        for (DataLifecycle current : lifecycles) {
+            if (current == Template.NO_LIFECYCLE) {
+                builder = null;
+            } else if (builder == null) {
+                builder = DataLifecycle.Builder.newBuilder(current);
+            } else {
+                if (current.getDataRetention() != null) {
+                    builder.dataRetention(current.getDataRetention());
+                }
+            }
+        }
+        return builder == null ? null : builder.build();
     }
 
     /**

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

@@ -41,6 +41,18 @@ import java.util.Objects;
  * template and a {@link ComponentTemplate}.
  */
 public class Template implements SimpleDiffable<Template>, ToXContentObject {
+
+    // This represents when the data lifecycle was explicitly set to be null, meaning the user wants to remove the
+    // lifecycle.
+    public static final DataLifecycle NO_LIFECYCLE = new DataLifecycle() {
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params, RolloverConfiguration rolloverConfiguration)
+            throws IOException {
+            return builder.nullValue();
+        }
+    };
+
     private static final ParseField SETTINGS = new ParseField("settings");
     private static final ParseField MAPPINGS = new ParseField("mappings");
     private static final ParseField ALIASES = new ParseField("aliases");
@@ -82,7 +94,12 @@ public class Template implements SimpleDiffable<Template>, ToXContentObject {
         }, ALIASES);
         // We adjust the parser to ensure that the error message will be consistent with that of an unknown field.
         if (DataLifecycle.isEnabled()) {
-            PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> DataLifecycle.fromXContent(p), LIFECYCLE);
+            PARSER.declareObjectOrNull(
+                ConstructingObjectParser.optionalConstructorArg(),
+                (p, c) -> DataLifecycle.fromXContent(p),
+                NO_LIFECYCLE,
+                LIFECYCLE
+            );
         }
     }
 
@@ -133,7 +150,16 @@ public class Template implements SimpleDiffable<Template>, ToXContentObject {
             this.aliases = null;
         }
         if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
-            this.lifecycle = in.readOptionalWriteable(DataLifecycle::new);
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_007)) {
+                boolean isExplicitNull = in.readBoolean();
+                if (isExplicitNull) {
+                    this.lifecycle = NO_LIFECYCLE;
+                } else {
+                    this.lifecycle = in.readOptionalWriteable(DataLifecycle::new);
+                }
+            } else {
+                this.lifecycle = in.readOptionalWriteable(DataLifecycle::new);
+            }
         } else {
             this.lifecycle = null;
         }
@@ -180,7 +206,15 @@ public class Template implements SimpleDiffable<Template>, ToXContentObject {
             out.writeMap(this.aliases, StreamOutput::writeString, (stream, aliasMetadata) -> aliasMetadata.writeTo(stream));
         }
         if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
-            out.writeOptionalWriteable(lifecycle);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_007)) {
+                boolean isExplicitNull = lifecycle == NO_LIFECYCLE;
+                out.writeBoolean(isExplicitNull);
+                if (isExplicitNull == false) {
+                    out.writeOptionalWriteable(lifecycle);
+                }
+            } else {
+                out.writeOptionalWriteable(lifecycle);
+            }
         }
     }
 

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

@@ -120,7 +120,12 @@ public class ComponentTemplateTests extends SimpleDiffableSerializationTestCase<
     }
 
     private static DataLifecycle randomLifecycle() {
-        return new DataLifecycle(randomMillisUpToYear9999());
+        return switch (randomIntBetween(0, 3)) {
+            case 0 -> DataLifecycleTests.IMPLICIT_INFINITE_RETENTION;
+            case 1 -> Template.NO_LIFECYCLE;
+            case 2 -> DataLifecycleTests.EXPLICIT_INFINITE_RETENTION;
+            default -> new DataLifecycle(randomMillisUpToYear9999());
+        };
     }
 
     private static Map<String, Object> randomMeta() {
@@ -260,7 +265,7 @@ public class ComponentTemplateTests extends SimpleDiffableSerializationTestCase<
         if (randomBoolean()) {
             aliases = randomAliases();
         }
-        DataLifecycle lifecycle = randomLifecycle();
+        DataLifecycle lifecycle = new DataLifecycle(randomMillisUpToYear9999());
         ComponentTemplate template = new ComponentTemplate(
             new Template(settings, mappings, aliases, lifecycle),
             randomNonNegativeLong(),
@@ -273,7 +278,9 @@ public class ComponentTemplateTests extends SimpleDiffableSerializationTestCase<
             template.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConfiguration);
             String serialized = Strings.toString(builder);
             assertThat(serialized, containsString("rollover"));
-            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getDataRetention()).getConditions().keySet()) {
+            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getEffectiveDataRetention())
+                .getConditions()
+                .keySet()) {
                 assertThat(serialized, containsString(label));
             }
         }

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

@@ -313,7 +313,9 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
             template.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConfiguration);
             String serialized = Strings.toString(builder);
             assertThat(serialized, containsString("rollover"));
-            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getDataRetention()).getConditions().keySet()) {
+            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getEffectiveDataRetention())
+                .getConditions()
+                .keySet()) {
                 assertThat(serialized, containsString(label));
             }
         }

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

@@ -16,14 +16,15 @@ import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
-import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.Matchers.containsString;
@@ -32,6 +33,9 @@ import static org.hamcrest.Matchers.nullValue;
 
 public class DataLifecycleTests extends AbstractXContentSerializingTestCase<DataLifecycle> {
 
+    public static final DataLifecycle EXPLICIT_INFINITE_RETENTION = new DataLifecycle(DataLifecycle.Retention.NULL);
+    public static final DataLifecycle IMPLICIT_INFINITE_RETENTION = new DataLifecycle((TimeValue) null);
+
     @Override
     protected Writeable.Reader<DataLifecycle> instanceReader() {
         return DataLifecycle::new;
@@ -39,19 +43,28 @@ public class DataLifecycleTests extends AbstractXContentSerializingTestCase<Data
 
     @Override
     protected DataLifecycle createTestInstance() {
-        if (randomBoolean()) {
-            return new DataLifecycle();
-        } else {
-            return new DataLifecycle(randomMillisUpToYear9999());
-        }
+        return switch (randomInt(2)) {
+            case 0 -> IMPLICIT_INFINITE_RETENTION;
+            case 1 -> EXPLICIT_INFINITE_RETENTION;
+            default -> new DataLifecycle(randomMillisUpToYear9999());
+        };
     }
 
     @Override
     protected DataLifecycle mutateInstance(DataLifecycle instance) throws IOException {
-        if (instance.getDataRetention() == null) {
-            return new DataLifecycle(randomMillisUpToYear9999());
+        if (IMPLICIT_INFINITE_RETENTION.equals(instance)) {
+            return randomBoolean() ? EXPLICIT_INFINITE_RETENTION : new DataLifecycle(randomMillisUpToYear9999());
         }
-        return new DataLifecycle(instance.getDataRetention().millis() + randomMillisUpToYear9999());
+        if (EXPLICIT_INFINITE_RETENTION.equals(instance)) {
+            return randomBoolean() ? IMPLICIT_INFINITE_RETENTION : new DataLifecycle(randomMillisUpToYear9999());
+        }
+        return switch (randomInt(2)) {
+            case 0 -> IMPLICIT_INFINITE_RETENTION;
+            case 1 -> EXPLICIT_INFINITE_RETENTION;
+            default -> new DataLifecycle(
+                randomValueOtherThan(instance.getEffectiveDataRetention().millis(), ESTestCase::randomMillisUpToYear9999)
+            );
+        };
     }
 
     @Override
@@ -67,7 +80,7 @@ public class DataLifecycleTests extends AbstractXContentSerializingTestCase<Data
             dataLifecycle.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConfiguration);
             String serialized = Strings.toString(builder);
             assertThat(serialized, containsString("rollover"));
-            for (String label : rolloverConfiguration.resolveRolloverConditions(dataLifecycle.getDataRetention())
+            for (String label : rolloverConfiguration.resolveRolloverConditions(dataLifecycle.getEffectiveDataRetention())
                 .getConditions()
                 .keySet()) {
                 assertThat(serialized, containsString(label));
@@ -79,34 +92,6 @@ public class DataLifecycleTests extends AbstractXContentSerializingTestCase<Data
         }
     }
 
-    public void testLifecycleComposition() {
-        // No lifecycles result to null
-        {
-            List<DataLifecycle> lifecycles = List.of();
-            assertThat(DataLifecycle.compose(lifecycles), nullValue());
-        }
-        // One lifecycle results to this lifecycle as the final
-        {
-            DataLifecycle lifecycle = createTestInstance();
-            List<DataLifecycle> lifecycles = List.of(lifecycle);
-            assertThat(DataLifecycle.compose(lifecycles), equalTo(lifecycle));
-        }
-        // If the last lifecycle is missing a property we keep the latest from the previous ones
-        {
-            DataLifecycle lifecycleWithRetention = new DataLifecycle(randomMillisUpToYear9999());
-            List<DataLifecycle> lifecycles = List.of(lifecycleWithRetention, new DataLifecycle());
-            assertThat(DataLifecycle.compose(lifecycles).getDataRetention(), equalTo(lifecycleWithRetention.getDataRetention()));
-        }
-        // If both lifecycle have all properties, then the latest one overwrites all the others
-        {
-            DataLifecycle lifecycle1 = new DataLifecycle(randomMillisUpToYear9999());
-            DataLifecycle lifecycle2 = new DataLifecycle(randomMillisUpToYear9999());
-            List<DataLifecycle> lifecycles = List.of(lifecycle1, lifecycle2);
-            assertThat(DataLifecycle.compose(lifecycles), equalTo(lifecycle2));
-        }
-
-    }
-
     public void testDefaultClusterSetting() {
         ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
         RolloverConfiguration rolloverConfiguration = clusterSettings.get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING);

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

@@ -1224,7 +1224,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             creationAndRolloverTimes,
             settings(Version.CURRENT),
             new DataLifecycle() {
-                public TimeValue getDataRetention() {
+                public TimeValue getEffectiveDataRetention() {
                     return testRetentionReference.get();
                 }
             }
@@ -1464,7 +1464,9 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             dataStream.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConfiguration);
             String serialized = Strings.toString(builder);
             assertThat(serialized, containsString("rollover"));
-            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getDataRetention()).getConditions().keySet()) {
+            for (String label : rolloverConfiguration.resolveRolloverConditions(lifecycle.getEffectiveDataRetention())
+                .getConditions()
+                .keySet()) {
                 assertThat(serialized, containsString(label));
             }
         }

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

@@ -1499,86 +1499,125 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         final MetadataIndexTemplateService service = getMetadataIndexTemplateService();
         ClusterState state = ClusterState.EMPTY_STATE;
 
-        DataLifecycle dataLifecycle1 = new DataLifecycle(TimeValue.timeValueDays(1));
-        ComponentTemplate ct1 = new ComponentTemplate(new Template(null, null, null, dataLifecycle1), null, null);
-        DataLifecycle dataLifecycle2 = new DataLifecycle(TimeValue.timeValueDays(30));
-        ComponentTemplate ct2 = new ComponentTemplate(new Template(null, null, null, dataLifecycle2), null, null);
-        ComponentTemplate ctNoLifecycle = new ComponentTemplate(new Template(null, null, null, null), null, null);
-
-        state = service.addComponentTemplate(state, true, "ct_1", ct1);
-        state = service.addComponentTemplate(state, true, "ct_2", ct2);
-        state = service.addComponentTemplate(state, true, "ct_no_lifecycle", ctNoLifecycle);
-        {
-            // Respect the order the templates are defined
-            ComposableIndexTemplate it = new ComposableIndexTemplate(
-                List.of("i1*"),
-                new Template(null, null, null),
-                List.of("ct_2", "ct_1"),
-                0L,
-                1L,
-                null,
-                new ComposableIndexTemplate.DataStreamTemplate(),
-                null
-            );
-            state = service.addIndexTemplateV2(state, true, "my-template-1", it);
-
-            DataLifecycle resolvedLifecycle = MetadataIndexTemplateService.resolveLifecycle(state.metadata(), "my-template-1");
-            assertThat(resolvedLifecycle, equalTo(dataLifecycle1));
-        }
-        {
-            // If the lifecycle is missing of the latest template we take the one from the previous
-            ComposableIndexTemplate it = new ComposableIndexTemplate(
-                List.of("i2*"),
-                new Template(null, null, null),
-                List.of("ct_1", "ct_no_lifecycle"),
-                0L,
-                1L,
-                null,
-                new ComposableIndexTemplate.DataStreamTemplate(),
-                null
-            );
-            state = service.addIndexTemplateV2(state, true, "my-template-2", it);
-
-            DataLifecycle resolvedLifecycle = MetadataIndexTemplateService.resolveLifecycle(state.metadata(), "my-template-2");
-
-            // Based on precedence only the latest
-            assertThat(resolvedLifecycle, equalTo(dataLifecycle1));
-        }
-        {
-            ComposableIndexTemplate it = new ComposableIndexTemplate(
-                List.of("i3*"),
-                new Template(null, null, null, dataLifecycle2),
-                List.of("ct_no_lifecycle", "ct_1"),
-                0L,
-                1L,
-                null,
-                new ComposableIndexTemplate.DataStreamTemplate(),
-                null
-            );
-            state = service.addIndexTemplateV2(state, true, "my-template-3", it);
+        DataLifecycle lifecycle30d = new DataLifecycle(TimeValue.timeValueDays(30));
+        String ct30d = "ct_30d";
+        state = addComponentTemplate(service, state, ct30d, lifecycle30d);
+
+        DataLifecycle lifecycle45d = new DataLifecycle(TimeValue.timeValueDays(45));
+        String ct45d = "ct_45d";
+        state = addComponentTemplate(service, state, ct45d, lifecycle45d);
+
+        DataLifecycle lifecycleNullRetention = new DataLifecycle.Builder().dataRetention(DataLifecycle.Retention.NULL).build();
+        String ctNullRetention = "ct_null_retention";
+        state = addComponentTemplate(service, state, ctNullRetention, lifecycleNullRetention);
+
+        String ctEmptyLifecycle = "ct_empty_lifecycle";
+        state = addComponentTemplate(service, state, ctEmptyLifecycle, DataLifecycleTests.IMPLICIT_INFINITE_RETENTION);
+
+        String ctNullLifecycle = "ct_null_lifecycle";
+        state = addComponentTemplate(service, state, ctNullLifecycle, Template.NO_LIFECYCLE);
+
+        String ctNoLifecycle = "ct_no_lifecycle";
+        state = addComponentTemplate(service, state, ctNoLifecycle, null);
+
+        // Component A: -
+        // Component B: "lifecycle": {}
+        // Composable Z: -
+        // Result: "lifecycle": {}
+        assertLifecycleResolution(
+            service,
+            state,
+            List.of(ctNoLifecycle, ctEmptyLifecycle),
+            null,
+            DataLifecycleTests.IMPLICIT_INFINITE_RETENTION
+        );
+
+        // Component A: "lifecycle": {}
+        // Component B: "lifecycle": {"retention": "30d"}
+        // Composable Z: -
+        // Result: "lifecycle": {"retention": "30d"}
+        assertLifecycleResolution(service, state, List.of(ctEmptyLifecycle, ct30d), null, lifecycle30d);
+
+        // Component A: "lifecycle": {"retention": "30d"}
+        // Component B: "lifecycle": {"retention": "45d"}
+        // Composable Z: "lifecycle": {}
+        // Result: "lifecycle": {"retention": "45d"}
+        assertLifecycleResolution(service, state, List.of(ct30d, ct45d), DataLifecycleTests.IMPLICIT_INFINITE_RETENTION, lifecycle45d);
+
+        // Component A: "lifecycle": {}
+        // Component B: "lifecycle": {"retention": "45d"}
+        // Composable Z: "lifecycle": {"retention": "30d"}
+        // Result: "lifecycle": {"retention": "30d"}
+        assertLifecycleResolution(service, state, List.of(ctEmptyLifecycle, ct45d), lifecycle30d, lifecycle30d);
+
+        // Component A: "lifecycle": {"retention": "30d"}
+        // Component B: "lifecycle": {"retention": null}
+        // Composable Z: -
+        // Result: "lifecycle": {"retention": null}, here the result of the composition is with retention explicitly
+        // nullified, but effectively this is equivalent to "lifecycle": {} when there is no further composition.
+        assertLifecycleResolution(service, state, List.of(ct30d, ctNullRetention), null, lifecycleNullRetention);
+
+        // Component A: "lifecycle": {}
+        // Component B: "lifecycle": {"retention": "45d"}
+        // Composable Z: "lifecycle": {"retention": null}
+        // Result: "lifecycle": {"retention": null} , here the result of the composition is with retention explicitly
+        // nullified, but effectively this is equivalent to "lifecycle": {} when there is no further composition.
+        assertLifecycleResolution(service, state, List.of(ctEmptyLifecycle, ct45d), lifecycleNullRetention, lifecycleNullRetention);
+
+        // Component A: "lifecycle": {"retention": "30d"}
+        // Component B: "lifecycle": {"retention": "45d"}
+        // Composable Z: "lifecycle": null
+        // Result: null aka unmanaged
+        assertLifecycleResolution(service, state, List.of(ct30d, ct45d), Template.NO_LIFECYCLE, null);
+
+        // Component A: "lifecycle": {"retention": "30d"}
+        // Component B: "lifecycle": null
+        // Composable Z: -
+        // Result: null aka unmanaged
+        assertLifecycleResolution(service, state, List.of(ct30d, ctNullLifecycle), null, null);
+
+        // Component A: "lifecycle": {"retention": "30d"}
+        // Component B: "lifecycle": null
+        // Composable Z: "lifecycle": {"retention": "45d"}
+        // Result: "lifecycle": {"retention": "45d"}
+        assertLifecycleResolution(service, state, List.of(ct30d, ctNullLifecycle), lifecycle45d, lifecycle45d);
+    }
 
-            DataLifecycle resolvedLifecycle = MetadataIndexTemplateService.resolveLifecycle(state.metadata(), "my-template-3");
+    private ClusterState addComponentTemplate(
+        MetadataIndexTemplateService service,
+        ClusterState state,
+        String name,
+        DataLifecycle dataLifecycle
+    ) throws Exception {
+        ComponentTemplate ct = new ComponentTemplate(new Template(null, null, null, dataLifecycle), null, null);
+        return service.addComponentTemplate(state, true, name, ct);
+    }
 
-            // The index template has higher precedence and overwrites the retention of the others.
-            assertThat(resolvedLifecycle, equalTo(dataLifecycle2));
-        }
-        {
-            ComposableIndexTemplate it = new ComposableIndexTemplate(
-                List.of("i4*"),
-                new Template(null, null, null, new DataLifecycle()),
-                List.of("ct_no_lifecycle", "ct_1"),
-                0L,
-                1L,
-                null,
-                new ComposableIndexTemplate.DataStreamTemplate(),
-                null
-            );
-            state = service.addIndexTemplateV2(state, true, "my-template-4", it);
+    private void assertLifecycleResolution(
+        MetadataIndexTemplateService service,
+        ClusterState state,
+        List<String> composeOf,
+        DataLifecycle lifecycleZ,
+        DataLifecycle expected
+    ) throws Exception {
+        ComposableIndexTemplate it = new ComposableIndexTemplate(
+            List.of(randomAlphaOfLength(10) + "*"),
+            new Template(null, null, null, lifecycleZ),
+            composeOf,
+            0L,
+            1L,
+            null,
+            new ComposableIndexTemplate.DataStreamTemplate(),
+            null
+        );
+        state = service.addIndexTemplateV2(state, true, "my-template", it);
 
-            DataLifecycle resolvedLifecycle = MetadataIndexTemplateService.resolveLifecycle(state.metadata(), "my-template-4");
+        DataLifecycle resolvedLifecycle = MetadataIndexTemplateService.resolveLifecycle(state.metadata(), "my-template");
 
-            // The index template lifecycle does not have a retention so the one from ct_1 remains unchanged.
-            assertThat(resolvedLifecycle, equalTo(dataLifecycle1));
+        if (expected == null) {
+            assertThat(resolvedLifecycle, nullValue());
+        } else {
+            assertThat(resolvedLifecycle, equalTo(expected));
         }
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/DataLifecycleUsageTransportAction.java

@@ -56,7 +56,7 @@ public class DataLifecycleUsageTransportAction extends XPackUsageFeatureTranspor
         final Collection<DataStream> dataStreams = state.metadata().dataStreams().values();
         LongSummaryStatistics retentionStats = dataStreams.stream()
             .filter(ds -> ds.getLifecycle() != null)
-            .collect(Collectors.summarizingLong(ds -> ds.getLifecycle().getDataRetention().getMillis()));
+            .collect(Collectors.summarizingLong(ds -> ds.getLifecycle().getEffectiveDataRetention().getMillis()));
         long dataStreamsWithLifecycles = retentionStats.getCount();
         long minRetention = dataStreamsWithLifecycles == 0 ? 0 : retentionStats.getMin();
         long maxRetention = dataStreamsWithLifecycles == 0 ? 0 : retentionStats.getMax();