[ML] fixing ML composite agg restart test (#78329) (#78403)

The test failure is due to data generation. This isn't caused by a bug, but is instead a test set up failure.

The issue is that scroll is seeing the data as a partial result, and aggs is not. Consequently, the first bucket is sometimes only present in scroll, but missing in aggs.

This is acceptable.

This commit fixes the test to make it more deterministic.

closes #76075

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

@@ -44,6 +44,7 @@
 import org.elasticsearch.xpack.core.ml.job.config.JobState;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.DataCounts;
 import org.elasticsearch.xpack.core.ml.job.results.Bucket;
+import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.hamcrest.Matcher;
 import org.junit.After;
 
@@ -350,12 +351,32 @@ public void testStopAndRestartCompositeDatafeed() throws Exception {
             .addMapping("type", "time", "type=date")
             .get();
         long numDocs = randomIntBetween(32, 2048);
+        final long intervalMillis = TimeValue.timeValueHours(1).millis();
         long now = System.currentTimeMillis();
         long oneWeekAgo = now - 604800000;
         long twoWeeksAgo = oneWeekAgo - 604800000;
-        indexDocs(logger, indexName, numDocs, twoWeeksAgo, oneWeekAgo);
+        indexDocs(
+            logger,
+            indexName,
+            1,
+            Intervals.alignToCeil(twoWeeksAgo, intervalMillis),
+            Intervals.alignToCeil(twoWeeksAgo, intervalMillis) + 1
+        );
+        indexDocs(
+            logger,
+            indexName,
+            numDocs - 1,
+            Intervals.alignToCeil(twoWeeksAgo, intervalMillis) + intervalMillis,
+            Intervals.alignToFloor(oneWeekAgo, intervalMillis)
+        );
         long numDocs2 = randomIntBetween(32, 2048);
-        indexDocs(logger, indexName, numDocs2, oneWeekAgo, now);
+        indexDocs(
+            logger,
+            indexName,
+            numDocs2,
+            Intervals.alignToCeil(oneWeekAgo, intervalMillis),
+            Intervals.alignToFloor(now, intervalMillis)
+        );
         client().admin().cluster().prepareHealth(indexName).setWaitForYellowStatus().get();
 
         String scrollJobId = "stop-restart-scroll";
@@ -464,7 +485,10 @@ public void testStopAndRestartCompositeDatafeed() throws Exception {
             Bucket scrollBucket = scrollBuckets.get(i);
             Bucket compositeBucket = compositeBuckets.get(i);
             try {
-                assertThat(compositeBucket.getTimestamp(), equalTo(scrollBucket.getTimestamp()));
+                assertThat("scroll buckets " + scrollBuckets + " composite buckets " + compositeBuckets,
+                    compositeBucket.getTimestamp(),
+                    equalTo(scrollBucket.getTimestamp())
+                );
                 assertThat(
                     "composite bucket [" + compositeBucket.getTimestamp() + "] [" + compositeBucket.getEventCount() + "] does not equal"
                         + " scroll bucket [" + scrollBucket.getTimestamp() + "] [" + scrollBucket.getEventCount() + "]",