Fix race condition in multi-shard rollup start/stop tests (#1529)

The rollup start/stop tests with multiple shards were experiencing race
conditions leading to version conflicts. The issue occurred when:

1. Test reads rollup document (seqNo = N)
2. Active rollup runner updates the document (seqNo = N+1)
3. Test tries to update with seqNo = N → 409 Version Conflict

The stop/start actions perform two sequential updates:
- First: Update rollup metadata status
- Second: Enable/disable the rollup job

The fix moves the _stop and _start API calls inside the waitFor block,
ensuring automatic retries on version conflicts. This is consistent with
the pattern already used in other rollup and transform tests.

Fixed tests:
- RestStopRollupActionIT: test stop rollup when multiple shards configured for IM config index
- RestStartRollupActionIT: test start rollup when multiple shards configured for IM config index

Signed-off-by: bowenlan-amzn <[email protected]>

Author: bowenlan-amzn

src/test/kotlin/org/opensearch/indexmanagement/rollup/resthandler/RestStartRollupActionIT.kt

@@ -245,12 +245,14 @@ class RestStartRollupActionIT : RollupRestAPITestCase() {
             val updatedRollup = getRollup(rollupId = rollup.id)
             val metadata = getRollupMetadataWithRoutingId(rollup.id, updatedRollup.metadataID!!)
             assertEquals("Status should be failed", RollupMetadata.Status.FAILED, metadata.status)
-        }
 
-        val response = client().makeRequest("POST", "${IndexManagementPlugin.ROLLUP_JOBS_BASE_URI}/${rollup.id}/_start")
-        assertEquals("Start rollup failed", RestStatus.OK, response.restStatus())
-        val expectedResponse = mapOf("acknowledged" to true)
-        assertEquals(expectedResponse, response.asMap())
+            // There are two calls to _start happening serially which is prone to version conflicts with multiple shards
+            // so including it in a waitFor to ensure it can retry a few times
+            val response = client().makeRequest("POST", "${IndexManagementPlugin.ROLLUP_JOBS_BASE_URI}/${rollup.id}/_start")
+            assertEquals("Start rollup failed", RestStatus.OK, response.restStatus())
+            val expectedResponse = mapOf("acknowledged" to true)
+            assertEquals(expectedResponse, response.asMap())
+        }
 
         val updatedRollup = getRollup(rollup.id)
         assertTrue("Rollup was not enabled", updatedRollup.enabled)

src/test/kotlin/org/opensearch/indexmanagement/rollup/resthandler/RestStopRollupActionIT.kt

@@ -305,11 +305,16 @@ class RestStopRollupActionIT : RollupRestAPITestCase() {
         waitFor {
             val rollupJob = getRollup(rollupId = rollup.id)
             assertNotNull("Rollup job doesn't have metadata set", rollupJob.metadataID)
+            val rollupMetadata = getRollupMetadataWithRoutingId(rollup.id, rollupJob.metadataID!!)
+            assertEquals("Rollup is not STARTED", RollupMetadata.Status.STARTED, rollupMetadata.status)
+
+            // There are two calls to _stop happening serially which is prone to version conflicts during an ongoing job
+            // so including it in a waitFor to ensure it can retry a few times
+            val response = client().makeRequest("POST", "$ROLLUP_JOBS_BASE_URI/${rollup.id}/_stop")
+            assertEquals("Stop rollup failed", RestStatus.OK, response.restStatus())
+            val expectedResponse = mapOf("acknowledged" to true)
+            assertEquals(expectedResponse, response.asMap())
         }
-        val response = client().makeRequest("POST", "$ROLLUP_JOBS_BASE_URI/${rollup.id}/_stop")
-        assertEquals("Stop rollup failed", RestStatus.OK, response.restStatus())
-        val expectedResponse = mapOf("acknowledged" to true)
-        assertEquals(expectedResponse, response.asMap())
 
         val updatedRollup = getRollup(rollup.id)
         assertFalse("Rollup was not disabled", updatedRollup.enabled)