RollupStart endpoint should return OK if job already started (#41502)

If a job is started or indexing, RollupStart should always return
a success (200 OK) response since the job is, in fact, started

docs/reference/migration/migrate_8_0.asciidoc

@@ -15,6 +15,7 @@ coming[8.0.0]
 * <<breaking_80_discovery_changes>>
 * <<breaking_80_mappings_changes>>
 * <<breaking_80_packaging_changes>>
+* <<breaking_80_rollup_changes>>
 * <<breaking_80_snapshots_changes>>
 * <<breaking_80_security_changes>>
 * <<breaking_80_ilm_changes>>
@@ -52,6 +53,7 @@ include::migrate_8_0/analysis.asciidoc[]
 include::migrate_8_0/discovery.asciidoc[]
 include::migrate_8_0/mappings.asciidoc[]
 include::migrate_8_0/packaging.asciidoc[]
+include::migrate_8_0/rollup.asciidoc[]
 include::migrate_8_0/snapshots.asciidoc[]
 include::migrate_8_0/security.asciidoc[]
 include::migrate_8_0/ilm.asciidoc[]

docs/reference/migration/migrate_8_0/rollup.asciidoc

@@ -0,0 +1,20 @@
+[float]
+[[breaking_80_rollup_changes]]
+=== Rollup changes
+
+//NOTE: The notable-breaking-changes tagged regions are re-used in the
+//Installation and Upgrade Guide
+
+//tag::notable-breaking-changes[]
+
+// end::notable-breaking-changes[]
+
+[float]
+==== StartRollupJob endpoint returns success if job already started
+
+Previously, attempting to start an already-started rollup job would
+result in a `500 InternalServerError: Cannot start task for Rollup Job
+[job] because state was [STARTED]` exception.
+
+Now, attempting to start a job that is already started will just
+return a successful `200 OK: started` response.

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/job/RollupJobTask.java

@@ -211,8 +211,10 @@ public class RollupJobTask extends AllocatedPersistentTask implements SchedulerE
     }
 
     /**
-     * Attempt to start the indexer.  If the state is anything other than STOPPED, this will fail.
-     * Otherwise, the persistent task's status will be updated to reflect the change.
+     * Attempt to start the indexer.
+     * - If the indexer is started/indexing, returns OK
+     * - If the indexer is stopped, starts task, updates persistent task's status, returns ok
+     * - Anything else returns error
      *
      * Note that while the job is started, the indexer will not necessarily run immediately.  That
      * will only occur when the scheduler triggers it based on the cron
@@ -221,8 +223,14 @@ public class RollupJobTask extends AllocatedPersistentTask implements SchedulerE
      */
     public synchronized void start(ActionListener<StartRollupJobAction.Response> listener) {
         final IndexerState prevState = indexer.getState();
-        if (prevState != IndexerState.STOPPED) {
-            // fails if the task is not STOPPED
+
+        if (prevState == IndexerState.STARTED || prevState == IndexerState.INDEXING) {
+            // We're already running so just return acknowledgement
+            logger.debug("Indexer already running (State: [" + prevState + "]), acknowledging start without change.");
+            listener.onResponse(new StartRollupJobAction.Response(true));
+            return;
+        } else if (prevState != IndexerState.STOPPED) {
+            // if we're not already started/indexing, we must be STOPPED to get started
             listener.onFailure(new ElasticsearchException("Cannot start task for Rollup Job [" + job.getConfig().getId() + "] because"
                     + " state was [" + prevState + "]"));
             return;
@@ -231,11 +239,10 @@ public class RollupJobTask extends AllocatedPersistentTask implements SchedulerE
         final IndexerState newState = indexer.start();
         if (newState != IndexerState.STARTED) {
             listener.onFailure(new ElasticsearchException("Cannot start task for Rollup Job [" + job.getConfig().getId() + "] because"
-                    + " state was [" + newState + "]"));
+                    + " new state was [" + newState + "]"));
             return;
         }
 
-
         final RollupJobStatus state = new RollupJobStatus(IndexerState.STARTED, indexer.getPosition());
         logger.debug("Updating state for rollup job [" + job.getConfig().getId() + "] to [" + state.getIndexerState() + "][" +
                 state.getPosition() + "]");

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/job/RollupJobTaskTests.java

@@ -163,17 +163,16 @@ public class RollupJobTaskTests extends ESTestCase {
         assertTrue(((RollupJobStatus)task.getStatus()).getPosition().containsKey("foo"));
 
         CountDownLatch latch = new CountDownLatch(1);
-        task.start(new ActionListener<StartRollupJobAction.Response>() {
+        task.start(new ActionListener<>() {
             @Override
             public void onResponse(StartRollupJobAction.Response response) {
-                fail("Should not have entered onResponse.");
+                assertTrue(response.isStarted());
+                latch.countDown();
             }
 
             @Override
             public void onFailure(Exception e) {
-                assertThat(e.getMessage(), equalTo("Cannot start task for Rollup Job ["
-                    + job.getConfig().getId() + "] because state was [STARTED]"));
-                latch.countDown();
+                fail("Should not have throw exception: " + e.getMessage());
             }
         });
         latch.await(3, TimeUnit.SECONDS);

x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/start_job.yml

@@ -59,9 +59,10 @@ setup:
   - is_true: started
 
   - do:
-      catch: /Cannot start task for Rollup Job \[foo\] because state was/
       headers:
         Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
       rollup.start_job:
         id: foo
+  - is_true: started
+