ES-125727 Fix for [CI] ReindexDataStreamTransportActionIT testAlreadyUpToDateDataStream failing (#126123) (#126135)

(cherry picked from commit d816845)

# Conflicts:
#	muted-tests.yml

Author: lukewhiting

muted-tests.yml

@@ -255,9 +255,6 @@ tests:
 - class: org.elasticsearch.smoketest.MlWithSecurityIT
   method: test {yaml=ml/trained_model_cat_apis/Test cat trained models}
   issue: https://github.com/elastic/elasticsearch/issues/125750
-- class: org.elasticsearch.xpack.migrate.action.ReindexDataStreamTransportActionIT
-  method: testAlreadyUpToDateDataStream
-  issue: https://github.com/elastic/elasticsearch/issues/125727
 - class: org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderIT
   method: testEnterpriseDownloaderTask
   issue: https://github.com/elastic/elasticsearch/issues/126124

x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamTransportActionIT.java

@@ -69,10 +69,7 @@ public void testAlreadyUpToDateDataStream() throws Exception {
             dataStreamName
         );
         final int backingIndexCount = createDataStream(dataStreamName);
-        AcknowledgedResponse response = client().execute(
-            new ActionType<AcknowledgedResponse>(ReindexDataStreamAction.NAME),
-            reindexDataStreamRequest
-        ).actionGet();
+        client().execute(new ActionType<AcknowledgedResponse>(ReindexDataStreamAction.NAME), reindexDataStreamRequest).actionGet();
         String persistentTaskId = "reindex-data-stream-" + dataStreamName;
         AtomicReference<ReindexDataStreamTask> runningTask = new AtomicReference<>();
         for (TransportService transportService : internalCluster().getInstances(TransportService.class)) {
@@ -91,12 +88,14 @@ public void testAlreadyUpToDateDataStream() throws Exception {
             );
         }
         ReindexDataStreamTask task = runningTask.get();
-        assertNotNull(task);
-        assertThat(task.getStatus().complete(), equalTo(true));
-        assertNull(task.getStatus().exception());
-        assertThat(task.getStatus().pending(), equalTo(0));
-        assertThat(task.getStatus().inProgress(), equalTo(Set.of()));
-        assertThat(task.getStatus().errors().size(), equalTo(0));
+        assertBusy(() -> {
+            assertNotNull(task);
+            assertThat(task.getStatus().complete(), equalTo(true));
+            assertNull(task.getStatus().exception());
+            assertThat(task.getStatus().pending(), equalTo(0));
+            assertThat(task.getStatus().inProgress(), equalTo(Set.of()));
+            assertThat(task.getStatus().errors().size(), equalTo(0));
+        });
 
         assertBusy(() -> {
             GetMigrationReindexStatusAction.Response statusResponse = client().execute(