[ML] Fix DatafeedJobsIT#testDatafeedTimingStats_DatafeedRecreated (elastic#137086) (elastic#137105)

Problem:

The test was failing intermittently with AssertionError: Expected: a value greater than <0L> but: <0L> was equal to <0L> due to a race condition between data processing completion and asynchronous datafeed timing stats persistence.

Root Cause:
Datafeed timing stats are persisted asynchronously. The test's single assertBusy() block checked both data processing completion AND timing stats simultaneously. Data processing could complete before timing stats persistence finished, causing search_count to still appear as 0.

Solution:
Split the assertion into two phases:

Phase 1: Wait for data processing to complete (processedRecordCount == numDocs)
Phase 2: Wait for timing stats to be persisted (search_count > 0)
This eliminates the race condition while maintaining the test's validation goals and testing realistic client-observable behavior.

Author: valeriy42

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

@@ -7,7 +7,6 @@
 package org.elasticsearch.xpack.ml.integration;
 
 import org.apache.logging.log4j.Level;
-import org.apache.lucene.tests.util.LuceneTestCase;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ResourceNotFoundException;
@@ -232,7 +231,6 @@ public void testLookbackOnlyRuntimeMapping() throws Exception {
         waitUntilJobIsClosed(jobBuilder.getId());
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/63973")
     public void testDatafeedTimingStats_DatafeedRecreated() throws Exception {
         client().admin().indices().prepareCreate("data").setMapping("time", "type=date").get();
         long numDocs = randomIntBetween(32, 2048);
@@ -254,11 +252,17 @@ public void testDatafeedTimingStats_DatafeedRecreated() throws Exception {
             // Datafeed did not do anything yet, hence search_count is equal to 0.
             assertDatafeedStats(datafeedId, DatafeedState.STOPPED, job.getId(), equalTo(0L));
             startDatafeed(datafeedId, 0L, now.toEpochMilli());
-            assertBusy(() -> {
-                assertThat(getDataCounts(job.getId()).getProcessedRecordCount(), equalTo(numDocs));
-                // Datafeed processed numDocs documents so search_count must be greater than 0.
-                assertDatafeedStats(datafeedId, DatafeedState.STOPPED, job.getId(), greaterThan(0L));
-            }, 60, TimeUnit.SECONDS);
+
+            // First, wait for data processing to complete
+            assertBusy(() -> { assertThat(getDataCounts(job.getId()).getProcessedRecordCount(), equalTo(numDocs)); }, 60, TimeUnit.SECONDS);
+            // Then, wait for datafeed timing stats to be persisted
+            // Datafeed processed numDocs documents so search_count must be greater than 0.
+            assertBusy(
+                () -> { assertDatafeedStats(datafeedId, DatafeedState.STOPPED, job.getId(), greaterThan(0L)); },
+                30,
+                TimeUnit.SECONDS
+            );
+
             deleteDatafeed(datafeedId);
             waitUntilJobIsClosed(job.getId());
         };
@@ -788,7 +792,6 @@ private void startRealtime(String jobId, Integer maxEmptySearches) throws Except
         }, 30, TimeUnit.SECONDS);
     }
 
-    @LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/105239")
     public void testStartDatafeed_GivenTimeout_Returns408() throws Exception {
         client().admin().indices().prepareCreate("data-1").setMapping("time", "type=date").get();
         long numDocs = 100;