Fetch backing indices using transport action

Author: nielsbauman

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

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

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

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

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

@@ -125,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<ClusterState> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
+        SubscribableListener<Void> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
         repository.blockOnDataFiles();
         try {
             clusterAdmin().prepareDeleteSnapshot(TEST_REQUEST_TIMEOUT, "test-repo", "test-snapshot")
@@ -147,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<ClusterState> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
+        SubscribableListener<Void> snapshotDeletionListener = createSnapshotDeletionListener("test-repo");
         repository.blockOnDataFiles();
         try {
             clusterAdmin().prepareDeleteSnapshot(TEST_REQUEST_TIMEOUT, "test-repo", "test-snapshot")
@@ -169,7 +169,7 @@ public void testDeleteSnapshotWhenWaitingForCompletion() throws Exception {
      * @param repositoryName The repository to monitor for deletions
      * @return the listener
      */
-    private SubscribableListener<ClusterState> createSnapshotDeletionListener(String repositoryName) {
+    private SubscribableListener<Void> createSnapshotDeletionListener(String repositoryName) {
         AtomicBoolean deleteHasStarted = new AtomicBoolean(false);
         return ClusterServiceUtils.addTemporaryStateListener(
             internalCluster().getCurrentMasterNodeInstance(ClusterService.class),

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

@@ -1388,7 +1388,7 @@ public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(
                                     && e.isClone()
                                     && e.shardSnapshotStatusByRepoShardId().isEmpty() == false
                             )
-                    ).addListener(l.map(cs -> null));
+                    ).addListener(l);
                     client.admin()
                         .cluster()
                         .prepareCloneSnapshot(TEST_REQUEST_TIMEOUT, repoName, originalSnapshotName, cloneName)
@@ -1401,8 +1401,7 @@ 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.map(cs -> null));
+            ClusterServiceUtils.addTemporaryStateListener(masterClusterService, cs -> SnapshotsInProgress.get(cs).isEmpty()).addListener(l);
         }));
 
         deterministicTaskQueue.runAllRunnableTasks();

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

@@ -263,24 +263,19 @@ public void clusterStateProcessed(ClusterState initialState, ClusterState newSta
         );
     }
 
-    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);
+    public static SubscribableListener<Void> addTemporaryStateListener(ClusterService clusterService, Predicate<ClusterState> predicate) {
+        final var listener = new SubscribableListener<Void>();
+        if (predicate.test(clusterService.state())) {
+            listener.onResponse(null);
             // 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 {
-                    final var state = event.state();
-                    if (predicate.test(state)) {
-                        listener.onResponse(state);
+                    if (predicate.test(event.state())) {
+                        listener.onResponse(null);
                     }
                 } catch (Exception e) {
                     listener.onFailure(e);

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

@@ -865,6 +865,8 @@ public static List<String> waitForDataStreamBackingIndices(String dataStreamName
      * 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) {
+        // We listen to the cluster state on the master node to ensure all other nodes have already acked the new cluster state.
+        // This avoids inconsistencies in subsequent API calls which might hit a non-master node.
         final var clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
         final var listener = ClusterServiceUtils.addTemporaryStateListener(clusterService, clusterState -> {
             final var dataStream = clusterState.metadata().getProject().dataStreams().get(dataStreamName);
@@ -873,17 +875,23 @@ public static List<String> waitForDataStreamIndices(String dataStreamName, int e
             }
             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();
+        safeAwait(listener);
+        final var backingIndexNames = getDataStreamBackingIndexNames(dataStreamName, failureStore);
+        assertEquals("Retrieved number of data stream indices doesn't match expectation", expectedSize, backingIndexNames.size());
+        return backingIndexNames;
     }
 
     /**
      * Returns a list of the data stream's backing index names.
      */
-    public List<String> getDataStreamBackingIndexNames(String dataStreamName) {
+    public static List<String> getDataStreamBackingIndexNames(String dataStreamName) {
+        return getDataStreamBackingIndexNames(dataStreamName, false);
+    }
+
+    /**
+     * Returns a list of the data stream's backing or failure index names.
+     */
+    public static List<String> getDataStreamBackingIndexNames(String dataStreamName, boolean failureStore) {
         GetDataStreamAction.Response response = safeGet(
             client().execute(
                 GetDataStreamAction.INSTANCE,
@@ -893,7 +901,7 @@ public List<String> getDataStreamBackingIndexNames(String dataStreamName) {
         assertThat(response.getDataStreams().size(), equalTo(1));
         DataStream dataStream = response.getDataStreams().getFirst().getDataStream();
         assertThat(dataStream.getName(), equalTo(dataStreamName));
-        return dataStream.getIndices().stream().map(Index::getName).toList();
+        return dataStream.getDataStreamIndices(failureStore).getIndices().stream().map(Index::getName).toList();
     }
 
     /**