Fix data stream retrieval in DataStreamLifecycleServiceIT

These tests had the potential to fail when two consecutive GET data
streams requests would hit two different nodes, where one node already
had the cluster state that contained the new backing index and the other
node didn't yet.

Caused by elastic#122852

Fixes elastic#124846
Fixes elastic#124950
Fixes elastic#124999

Author: nielsbauman

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java

@@ -339,9 +339,6 @@ tests:
 - class: org.elasticsearch.xpack.ilm.DataStreamAndIndexLifecycleMixingTests
   method: testUpdateIndexTemplateToDataStreamLifecyclePreference
   issue: https://github.com/elastic/elasticsearch/issues/124837
-- class: org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleServiceIT
-  method: testAutomaticForceMerge
-  issue: https://github.com/elastic/elasticsearch/issues/124846
 - class: org.elasticsearch.backwards.MixedClusterClientYamlTestSuiteIT
   method: test {p0=search.vectors/41_knn_search_bbq_hnsw/Test knn search}
   issue: https://github.com/elastic/elasticsearch/issues/124848
@@ -363,9 +360,6 @@ tests:
 - class: org.elasticsearch.packaging.test.BootstrapCheckTests
   method: test20RunWithBootstrapChecks
   issue: https://github.com/elastic/elasticsearch/issues/124940
-- class: org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleServiceIT
-  method: testErrorRecordingOnRetention
-  issue: https://github.com/elastic/elasticsearch/issues/124950
 - class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
   method: test {lookup-join.MvJoinKeyFromRow SYNC}
   issue: https://github.com/elastic/elasticsearch/issues/124951
@@ -390,9 +384,6 @@ tests:
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test011SecurityEnabledStatus
   issue: https://github.com/elastic/elasticsearch/issues/124990
-- class: org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleServiceIT
-  method: testLifecycleAppliedToFailureStore
-  issue: https://github.com/elastic/elasticsearch/issues/124999
 - class: org.elasticsearch.backwards.MixedClusterClientYamlTestSuiteIT
   method: test {p0=search/610_function_score/Random}
   issue: https://github.com/elastic/elasticsearch/issues/125010

muted-tests.yml

@@ -121,7 +121,7 @@ public void onFailure(Exception e) {
         );
 
         safeAwait(
-            taskCreatedListener.<Void>andThen(
+            taskCreatedListener.<ClusterState>andThen(
                 (l, v) -> ClusterServiceUtils.addTemporaryStateListener(
                     masterClusterService,
                     clusterState -> hasPersistentTask(clusterState) == false

server/src/internalClusterTest/java/org/elasticsearch/persistent/PersistentTaskCreationFailureIT.java

@@ -657,7 +657,7 @@ && switch (shardEntry.getValue().state()) {
         resetMockLog();
     }
 
-    private static SubscribableListener<Void> createSnapshotPausedListener(
+    private static SubscribableListener<ClusterState> createSnapshotPausedListener(
         ClusterService clusterService,
         String repoName,
         String indexName,

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotShutdownIT.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.SnapshotDeletionsInProgress;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -124,7 +125,7 @@ public void testDeleteSnapshotWhenNotWaitingForCompletion() throws Exception {
         createSnapshot("test-repo", "test-snapshot", List.of("test-index"));
         MockRepository repository = getRepositoryOnMaster("test-repo");
         PlainActionFuture<AcknowledgedResponse> listener = new PlainActionFuture<>();
-        SubscribableListener<Void> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
+        SubscribableListener<ClusterState> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
         repository.blockOnDataFiles();
         try {
             clusterAdmin().prepareDeleteSnapshot(TEST_REQUEST_TIMEOUT, "test-repo", "test-snapshot")
@@ -146,7 +147,7 @@ public void testDeleteSnapshotWhenWaitingForCompletion() throws Exception {
         createSnapshot("test-repo", "test-snapshot", List.of("test-index"));
         MockRepository repository = getRepositoryOnMaster("test-repo");
         PlainActionFuture<AcknowledgedResponse> requestCompleteListener = new PlainActionFuture<>();
-        SubscribableListener<Void> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
+        SubscribableListener<ClusterState> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
         repository.blockOnDataFiles();
         try {
             clusterAdmin().prepareDeleteSnapshot(TEST_REQUEST_TIMEOUT, "test-repo", "test-snapshot")
@@ -168,7 +169,7 @@ public void testDeleteSnapshotWhenWaitingForCompletion() throws Exception {
      * @param repositoryName The repository to monitor for deletions
      * @return the listener
      */
-    private SubscribableListener<Void> createSnapshotDeletionListener(String repositoryName) {
+    private SubscribableListener<ClusterState> createSnapshotDeletionListener(String repositoryName) {
         AtomicBoolean deleteHasStarted = new AtomicBoolean(false);
         return ClusterServiceUtils.addTemporaryStateListener(
             internalCluster().getCurrentMasterNodeInstance(ClusterService.class),

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotsServiceIT.java

@@ -1388,7 +1388,7 @@ public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
                                     && e.isClone()
                                     && e.shardSnapshotStatusByRepoShardId().isEmpty() == false
                             )
-                    ).addListener(l);
+                    ).addListener(l.map(cs -> null));
                     client.admin()
                         .cluster()
                         .prepareCloneSnapshot(TEST_REQUEST_TIMEOUT, repoName, originalSnapshotName, cloneName)
@@ -1401,7 +1401,8 @@ public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
         testListener = testListener.andThen(l -> scheduleNow(() -> {
             // Once all snapshots & clones have started, drop the data node and wait for all snapshot activity to complete
             testClusterNodes.disconnectNode(testClusterNodes.randomDataNodeSafe());
-            ClusterServiceUtils.addTemporaryStateListener(masterClusterService, cs -> SnapshotsInProgress.get(cs).isEmpty()).addListener(l);
+            ClusterServiceUtils.addTemporaryStateListener(masterClusterService, cs -> SnapshotsInProgress.get(cs).isEmpty())
+                .addListener(l.map(cs -> null));
         }));
 
         deterministicTaskQueue.runAllRunnableTasks();

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -263,14 +263,24 @@ public void clusterStateProcessed(ClusterState initialState, ClusterState newSta
         );
     }
 
-    public static SubscribableListener<Void> addTemporaryStateListener(ClusterService clusterService, Predicate<ClusterState> predicate) {
-        final var listener = new SubscribableListener<Void>();
+    public static SubscribableListener<ClusterState> addTemporaryStateListener(
+        ClusterService clusterService,
+        Predicate<ClusterState> predicate
+    ) {
+        final var listener = new SubscribableListener<ClusterState>();
+        final var initialState = clusterService.state();
+        if (predicate.test(initialState)) {
+            listener.onResponse(initialState);
+            // No need to add the cluster state listener if the predicate already passes.
+            return listener;
+        }
         final ClusterStateListener clusterStateListener = new ClusterStateListener() {
             @Override
             public void clusterChanged(ClusterChangedEvent event) {
                 try {
-                    if (predicate.test(event.state())) {
-                        listener.onResponse(null);
+                    final var state = event.state();
+                    if (predicate.test(state)) {
+                        listener.onResponse(state);
                     }
                 } catch (Exception e) {
                     listener.onFailure(e);
@@ -284,11 +294,7 @@ public String toString() {
         };
         clusterService.addListener(clusterStateListener);
         listener.addListener(ActionListener.running(() -> clusterService.removeListener(clusterStateListener)));
-        if (predicate.test(clusterService.state())) {
-            listener.onResponse(null);
-        } else {
-            listener.addTimeout(ESTestCase.SAFE_AWAIT_TIMEOUT, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
-        }
+        listener.addTimeout(ESTestCase.SAFE_AWAIT_TIMEOUT, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
         return listener;
     }
 }

test/framework/src/main/java/org/elasticsearch/test/ClusterServiceUtils.java

@@ -854,6 +854,32 @@ private static Settings.Builder getExcludeSettings(int num, Settings.Builder bui
         return builder;
     }
 
+    /**
+     * Waits for the specified data stream to have the expected number of backing indices.
+     */
+    public static List<String> waitForDataStreamBackingIndices(String dataStreamName, int expectedSize) {
+        return waitForDataStreamIndices(dataStreamName, expectedSize, false);
+    }
+
+    /**
+     * Waits for the specified data stream to have the expected number of backing or failure indices.
+     */
+    public static List<String> waitForDataStreamIndices(String dataStreamName, int expectedSize, boolean failureStore) {
+        final var clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
+        final var listener = ClusterServiceUtils.addTemporaryStateListener(clusterService, clusterState -> {
+            final var dataStream = clusterState.metadata().getProject().dataStreams().get(dataStreamName);
+            if (dataStream == null) {
+                return false;
+            }
+            return dataStream.getDataStreamIndices(failureStore).getIndices().size() == expectedSize;
+        });
+        final var state = safeAwait(listener);
+        // We will only reach the return statement when the data stream exists (and has the expected number of indices),
+        // so we can safely retrieve the data stream without worrying about NPEs.
+        final var indices = state.metadata().getProject().dataStreams().get(dataStreamName).getDataStreamIndices(failureStore).getIndices();
+        return indices.stream().map(Index::getName).toList();
+    }
+
     /**
      * Returns a list of the data stream's backing index names.
      */