Fixing unit tests

Author: masseyke

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java

@@ -360,6 +360,13 @@ private void run(ProjectState projectState) {
         final var project = projectState.metadata();
         int affectedIndices = 0;
         int affectedDataStreams = 0;
+        List<DataStreamLifecycleAction> actions = List.of(
+            this::maybeExecuteRollover,
+            this::timeSeriesIndicesStillWithinTimeBounds,
+            this::maybeExecuteRetention,
+            this::maybeExecuteForceMerge,
+            this::maybeExecuteDownsampling
+        );
         for (DataStream dataStream : project.dataStreams().values()) {
             clearErrorStoreForUnmanagedIndices(project, dataStream);
             var dataLifecycleEnabled = dataStream.getDataLifecycle() != null && dataStream.getDataLifecycle().enabled();
@@ -368,14 +375,6 @@ private void run(ProjectState projectState) {
             if (dataLifecycleEnabled == false && failuresLifecycleEnabled == false) {
                 continue;
             }
-            List<DataStreamLifecycleAction> actions = List.of(
-                this::maybeExecuteRollover,
-                DataStreamLifecycleService::timeSeriesIndicesStillWithinTimeBounds,
-                this::maybeExecuteRetention,
-                this::maybeExecuteForceMerge,
-                this::maybeExecuteDownsampling
-            );
-
             // the following indices should not be considered for the remainder of this service run, for various reasons.
             Set<Index> indicesToExcludeForRemainingRun = new HashSet<>();
             for (DataStreamLifecycleAction action : actions) {
@@ -394,11 +393,20 @@ private void run(ProjectState projectState) {
         );
     }
 
+    private Set<Index> timeSeriesIndicesStillWithinTimeBounds(
+        ProjectState projectState,
+        DataStream dataStream,
+        Set<Index> indicesToExcludeForRemainingRun
+    ) {
+        return timeSeriesIndicesStillWithinTimeBounds(projectState, dataStream, indicesToExcludeForRemainingRun, nowSupplier);
+    }
+
     // visible for testing
     static Set<Index> timeSeriesIndicesStillWithinTimeBounds(
         ProjectState projectState,
         DataStream dataStream,
-        Set<Index> indicesToExcludeForRemainingRun
+        Set<Index> indicesToExcludeForRemainingRun,
+        LongSupplier nowSupplier
     ) {
         // tsds indices that are still within their time bounds (i.e. now < time_series.end_time) - we don't want these indices to be
         // deleted, forcemerged, or downsampled as they're still expected to receive large amounts of writes
@@ -412,6 +420,15 @@ static Set<Index> timeSeriesIndicesStillWithinTimeBounds(
                 Instant configuredEndTime = IndexSettings.TIME_SERIES_END_TIME.get(backingIndex.getSettings());
                 assert configuredEndTime != null
                     : "a time series index must have an end time configured but [" + index.getName() + "] does not";
+                if (nowSupplier.getAsLong() <= configuredEndTime.toEpochMilli()) {
+                    logger.trace(
+                        "Data stream lifecycle will not perform any operations in this run on time series index [{}] because "
+                            + "its configured [{}] end time has not lapsed",
+                        index.getName(),
+                        configuredEndTime
+                    );
+                    tsIndicesWithinBounds.add(index);
+                }
             }
         }
         return tsIndicesWithinBounds;

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

@@ -357,7 +357,8 @@ public void testMergePolicyNotExecutedForTSIndicesWithinTimeBounds() {
         clusterState = ClusterState.builder(clusterState).putProjectMetadata(builder).build();
 
         dataStreamLifecycleService.run(clusterState);
-        // There should be two client requests: one rollover, and one to update the merge policy settings. N.B. The merge policy settings
+        // There should be two client requests: one rollover, and one to update the merge policy settings. One of the non-write indices is
+        // still within the time bounds to be written to, so it is not force merged. N.B. The merge policy settings
         // will always be updated before the force merge is done, see testMergePolicySettingsAreConfiguredBeforeForcemerge.
         assertThat(clientSeenRequests.size(), is(2));
         assertThat(clientSeenRequests.get(0), instanceOf(RolloverRequest.class));
@@ -961,7 +962,13 @@ public void onFailure(Exception e) {
         final var projectId = randomProjectIdOrDefault();
         String targetIndex = randomAlphaOfLength(20);
         DataStreamLifecycleService.UpdateDataStreamLifecycleCustomMetadataTask task =
-            new DataStreamLifecycleService.UpdateDataStreamLifecycleCustomMetadataTask(listener, projectId, targetIndex, threadPool);
+            new DataStreamLifecycleService.UpdateDataStreamLifecycleCustomMetadataTask(
+                listener,
+                projectId,
+                targetIndex,
+                FORCE_MERGE_COMPLETED_TIMESTAMP_METADATA_KEY,
+                Long.toString(threadPool.absoluteTimeInMillis())
+            );
         {
             Exception exception = new RuntimeException("task failed");
             task.onFailure(exception);
@@ -1441,14 +1448,19 @@ public void testTimeSeriesIndicesStillWithinTimeBounds() {
         );
         final ProjectState project = clusterState.projectState(projectId);
         DataStream dataStream = project.metadata().dataStreams().get(dataStreamName);
+        dataStream = dataStream.copy()
+            .setName(dataStreamName)
+            .setGeneration(dataStream.getGeneration() + 1)
+            .setLifecycle(DataStreamLifecycle.dataLifecycleBuilder().dataRetention(TimeValue.ZERO).build())
+            .build();
         {
-            // test for an index for which `now` is outside its time bounds
-            Index firstGenIndex = dataStream.getIndices().get(0);
+            // test for an index for which `now` is outside its time bounds by excluding the two that are not
             Set<Index> indices = DataStreamLifecycleService.timeSeriesIndicesStillWithinTimeBounds(
                 // the end_time for the first generation has lapsed
                 project,
                 dataStream,
-                Set.of()
+                Set.of(dataStream.getIndices().get(1), dataStream.getIndices().get(2)),
+                currentTime::toEpochMilli
             );
             assertThat(indices.size(), is(0));
         }
@@ -1458,21 +1470,43 @@ public void testTimeSeriesIndicesStillWithinTimeBounds() {
                 // the end_time for the first generation has lapsed, but the other 2 generations are still within bounds
                 project,
                 dataStream,
-                Set.of()
+                Set.of(),
+                currentTime::toEpochMilli
             );
             assertThat(indices.size(), is(2));
             assertThat(indices, containsInAnyOrder(dataStream.getIndices().get(1), dataStream.getIndices().get(2)));
         }
 
         {
+            ProjectMetadata.Builder projectMetadataBuilder = ProjectMetadata.builder(project.metadata());
             // non time_series indices are not within time bounds (they don't have any)
-            IndexMetadata indexMeta = IndexMetadata.builder(randomAlphaOfLengthBetween(10, 30))
-                .settings(indexSettings(1, 1).put(IndexMetadata.SETTING_INDEX_VERSION_CREATED.getKey(), IndexVersion.current()))
-                .build();
-
-            ProjectState newProject = project.updateProject(ProjectMetadata.builder(project.metadata()).put(indexMeta, true).build());
+            for (Index index : dataStream.getIndices()) {
+                IndexMetadata indexMeta = IndexMetadata.builder(index.getName())
+                    .settings(
+                        indexSettings(1, 1).put(IndexMetadata.SETTING_INDEX_VERSION_CREATED.getKey(), IndexVersion.current())
+                            .put(IndexSettings.MODE.getKey(), IndexMode.STANDARD)
+                    )
+                    .build();
+                projectMetadataBuilder.put(indexMeta, true);
+            }
 
-            Set<Index> indices = DataStreamLifecycleService.timeSeriesIndicesStillWithinTimeBounds(newProject, dataStream, Set.of());
+            List<Index> standardIndices = dataStream.getIndices()
+                .stream()
+                .map(index -> new Index(index.getName(), index.getUUID()))
+                .toList();
+            ProjectState newProject = project.updateProject(projectMetadataBuilder.build());
+            dataStream = dataStream.copy()
+                .setName(dataStreamName)
+                .setGeneration(dataStream.getGeneration() + 1)
+                .setBackingIndices(DataStream.DataStreamIndices.backingIndicesBuilder(standardIndices).build())
+                .setIndexMode(IndexMode.STANDARD)
+                .build();
+            Set<Index> indices = DataStreamLifecycleService.timeSeriesIndicesStillWithinTimeBounds(
+                newProject,
+                dataStream,
+                Set.of(),
+                currentTime::toEpochMilli
+            );
             assertThat(indices.size(), is(0));
         }
     }