[TEST] Force a stop to save rollup state before continuing (#32787)

We only upgrade the ID when the state is saved in one of four scenarios:

- when we reach a checkpoint (every 50 pages)
- when we run out of data
- when explicitly stopped
- on failure

The test was relying on the pre-upgrade to finish, save state and then
the post-upgrade to start, hit the end of data and upgrade ID.  THEN
get the new doc and apply the new ID.

But I think this is vulnerable to timing issues. If the pre-upgrade
portion shutdown before it saved the state, when restarting we would run
through all the data from the beginning with the old ID, meaning both
docs would still have the old scheme.

This change makes the pre-upgrade wait for the job to go back to STARTED
so that we know it persisted the end point.  Post-upgrade, it stops and
restarts the job to ensure the state was persisted and the ID upgraded.

That _should_ rule out the above timing issue.

Closes #32773

x-pack/qa/full-cluster-restart/src/test/java/org/elasticsearch/xpack/restart/FullClusterRestartIT.java

@@ -382,13 +382,29 @@ public class FullClusterRestartIT extends ESRestTestCase {
                 }
             });
 
+            // After we've confirmed the doc, wait until we move back to STARTED so that we know the
+            // state was saved at the end
+            waitForRollUpJob("rollup-id-test", equalTo("started"));
+
         } else {
 
             final Request indexRequest = new Request("POST", "/id-test-rollup/_doc/2");
             indexRequest.setJsonEntity("{\"timestamp\":\"2018-01-02T00:00:01\",\"value\":345}");
             client().performRequest(indexRequest);
 
-            assertRollUpJob("rollup-id-test");
+            // stop the rollup job to force a state save, which will upgrade the ID
+            final Request stopRollupJobRequest = new Request("POST", "_xpack/rollup/job/rollup-id-test/_stop");
+            Map<String, Object> stopRollupJobResponse = entityAsMap(client().performRequest(stopRollupJobRequest));
+            assertThat(stopRollupJobResponse.get("stopped"), equalTo(Boolean.TRUE));
+
+            waitForRollUpJob("rollup-id-test", equalTo("stopped"));
+
+            // start the rollup job again
+            final Request startRollupJobRequest = new Request("POST", "_xpack/rollup/job/rollup-id-test/_start");
+            Map<String, Object> startRollupJobResponse = entityAsMap(client().performRequest(startRollupJobRequest));
+            assertThat(startRollupJobResponse.get("started"), equalTo(Boolean.TRUE));
+
+            waitForRollUpJob("rollup-id-test", anyOf(equalTo("indexing"), equalTo("started")));
 
             assertBusy(() -> {
                 client().performRequest(new Request("POST", "id-test-results-rollup/_refresh"));