Fix DatafeedJobsIT.testDatafeedTimingStats_DatafeedRecreated (elastic#137928)

This commit fixes two separate issues in the test.

Firstly, the assertion that search_count stats should be equal to zero
may fail the second time the openAndRunJob runnable is invoked if the
stats haven't been updated since the previous datafeed with the same ID
was deleted. Wrapping the assertion in an assertBusy() call prevents
this.

Secondly, deleting the datafeed may fail if the master node which
processes the delete datafeed action hasn't finished updating its state
to reflect the fact that the datafeed has been stopped yet, but the node
that processes the datafeed stats request has. Wrapping the
datafeed deletion in an assertBusy() call allows it to be retried if
this race condition is encountered.

Closes elastic#137207

(cherry picked from commit 2bb729b)

Author: DonalEvans

muted-tests.yml

@@ -398,9 +398,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.inference.rerank.RerankOperatorTests
   method: testSimpleCircuitBreaking
   issue: https://github.com/elastic/elasticsearch/issues/133619
-- class: org.elasticsearch.xpack.ml.integration.DatafeedJobsIT
-  method: testDatafeedTimingStats_DatafeedRecreated
-  issue: https://github.com/elastic/elasticsearch/issues/137207
 - class: org.elasticsearch.upgrades.SearchStatesIT
   method: testBWCSearchStates
   issue: https://github.com/elastic/elasticsearch/issues/137681

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/DatafeedJobsIT.java

@@ -250,7 +250,7 @@ public void testDatafeedTimingStats_DatafeedRecreated() throws Exception {
 
             putDatafeed(datafeedConfig);
             // Datafeed did not do anything yet, hence search_count is equal to 0.
-            assertDatafeedStats(datafeedId, DatafeedState.STOPPED, job.getId(), equalTo(0L));
+            assertBusy(() -> assertDatafeedStats(datafeedId, DatafeedState.STOPPED, job.getId(), equalTo(0L)));
             startDatafeed(datafeedId, 0L, now.toEpochMilli());
 
             // First, wait for data processing to complete
@@ -263,7 +263,14 @@ public void testDatafeedTimingStats_DatafeedRecreated() throws Exception {
                 TimeUnit.SECONDS
             );
 
-            deleteDatafeed(datafeedId);
+            DeleteDatafeedAction.Request request = new DeleteDatafeedAction.Request(datafeedId);
+            assertBusy(() -> {
+                try {
+                    client().execute(DeleteDatafeedAction.INSTANCE, request).actionGet();
+                } catch (Exception e) {
+                    throw new AssertionError("Datafeed could not be deleted", e);
+                }
+            });
             waitUntilJobIsClosed(job.getId());
         };