[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

Author: edsavage

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

@@ -51,7 +51,6 @@ public void testPersistJob() throws Exception {
     }
 
     // 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 void testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData() t
         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 void testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData() t
         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 void testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData() t
         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 void testPersistJobOnGracefulShutdown_givenTimeAdvancedAfterNoNewData() t
         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")) {