[ML][TEST] Fix failing test testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData (#40363)

Ensure that there is at least a 1s delay between the time that state
is persisted by each of the two jobs in the test.

Model snapshot IDs use the current time in epoch seconds to
distinguish themselves, hence snapshots will be overwritten
by another if it occurs in the same 1s window.

Closes #40347

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

@@ -51,7 +51,6 @@ public class PersistJobIT extends MlNativeAutodetectIntegTestCase {
     }
 
     // check that state is persisted after time has been advanced even if no new data is seen in the interim
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40347")
     public void testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData() throws Exception {
         String jobId = "time-advanced-after-no-new-data-test";
 
@@ -60,6 +59,7 @@ public class PersistJobIT extends MlNativeAutodetectIntegTestCase {
         FlushJobAction.Response flushResponse = flushJob(jobId, true);
 
         closeJob(jobId);
+        long job1CloseTime = System.currentTimeMillis() / 1000;
 
         // Check that state has been persisted
         SearchResponse stateDocsResponse1 = client().prepareSearch(AnomalyDetectorsIndex.jobStateIndexPattern())
@@ -71,7 +71,7 @@ public class PersistJobIT extends MlNativeAutodetectIntegTestCase {
         int numQuantileRecords = 0;
         int numStateRecords = 0;
         for (SearchHit hit : stateDocsResponse1.getHits().getHits()) {
-            logger.info(hit.getId());
+            logger.info("1: " + hit.getId());
             if (hit.getId().contains("quantiles")) {
                 ++numQuantileRecords;
             } else if (hit.getId().contains("model_state")) {
@@ -82,6 +82,13 @@ public class PersistJobIT extends MlNativeAutodetectIntegTestCase {
         assertThat(numQuantileRecords, equalTo(1));
         assertThat(numStateRecords, equalTo(1));
 
+        // To generate unique snapshot IDs ensure that there is at least a 1s delay between the
+        // time each job was closed
+        assertBusy(() -> {
+            long timeNow = System.currentTimeMillis() / 1000;
+            assertFalse(job1CloseTime >= timeNow);
+        });
+
         // re-open the job
         openJob(jobId);
 
@@ -104,7 +111,7 @@ public class PersistJobIT extends MlNativeAutodetectIntegTestCase {
         numQuantileRecords = 0;
         numStateRecords = 0;
         for (SearchHit hit : stateDocsResponse2.getHits().getHits()) {
-            logger.info(hit.getId());
+            logger.info("2: " + hit.getId());
             if (hit.getId().contains("quantiles")) {
                 ++numQuantileRecords;
             } else if (hit.getId().contains("model_state")) {