Rename methods in PersistentTasksService (#30837)

This commit renames methods in the PersistentTasksService, to 
make obvious that the methods send requests in order to change 
the state of persistent tasks. 

Relates to #29608.

server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java

@@ -103,12 +103,16 @@ public class AllocatedPersistentTask extends CancellableTask {
     }
 
     /**
-     * Waits for this persistent task to have the desired state.
+     * Waits for a given persistent task to comply with a given predicate, then call back the listener accordingly.
+     *
+     * @param predicate the persistent task predicate to evaluate
+     * @param timeout a timeout for waiting
+     * @param listener the callback listener
      */
-    public void waitForPersistentTaskStatus(Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
-                                            @Nullable TimeValue timeout,
-                                            PersistentTasksService.WaitForPersistentTaskStatusListener<?> listener) {
-        persistentTasksService.waitForPersistentTaskStatus(persistentTaskId, predicate, timeout, listener);
+    public void waitForPersistentTask(final Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
+                                      final @Nullable TimeValue timeout,
+                                      final PersistentTasksService.WaitForPersistentTaskListener<?> listener) {
+        persistentTasksService.waitForPersistentTaskCondition(persistentTaskId, predicate, timeout, listener);
     }
 
     final boolean isCompleted() {
@@ -143,7 +147,7 @@ public class AllocatedPersistentTask extends CancellableTask {
                 this.failure = failure;
                 if (prevState == State.STARTED) {
                     logger.trace("sending notification for completed task [{}] with id [{}]", getAction(), getPersistentTaskId());
-                    persistentTasksService.sendCompletionNotification(getPersistentTaskId(), getAllocationId(), failure, new
+                    persistentTasksService.sendCompletionRequest(getPersistentTaskId(), getAllocationId(), failure, new
                             ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                                 @Override
                                 public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {

server/src/main/java/org/elasticsearch/persistent/PersistentTasksNodeService.java

@@ -196,7 +196,8 @@ public class PersistentTasksNodeService extends AbstractComponent implements Clu
         AllocatedPersistentTask task = runningTasks.remove(allocationId);
         if (task.markAsCancelled()) {
             // Cancel the local task using the task manager
-            persistentTasksService.sendTaskManagerCancellation(task.getId(), new ActionListener<CancelTasksResponse>() {
+            String reason = "task has been removed, cancelling locally";
+            persistentTasksService.sendCancelRequest(task.getId(), reason, new ActionListener<CancelTasksResponse>() {
                 @Override
                 public void onResponse(CancelTasksResponse cancelTasksResponse) {
                     logger.trace("Persistent task [{}] with id [{}] and allocation id [{}] was cancelled", task.getAction(),

server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java

@@ -22,14 +22,12 @@ import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
-import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractComponent;
@@ -37,20 +35,24 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.node.NodeClosedException;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 
-import java.util.function.BiConsumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /**
- * This service is used by persistent actions to propagate changes in the action state and notify about completion
+ * This service is used by persistent tasks and allocated persistent tasks to communicate changes
+ * to the master node so that the master can update the cluster state and can track of the states
+ * of the persistent tasks.
  */
 public class PersistentTasksService extends AbstractComponent {
 
+    private static final String ACTION_ORIGIN_TRANSIENT_NAME = "action.origin";
+    private static final String PERSISTENT_TASK_ORIGIN = "persistent_tasks";
+
     private final Client client;
     private final ClusterService clusterService;
     private final ThreadPool threadPool;
@@ -63,92 +65,115 @@ public class PersistentTasksService extends AbstractComponent {
     }
 
     /**
-     * Creates the specified persistent task and attempts to assign it to a node.
+     * Notifies the master node to create new persistent task and to assign it to a node.
      */
-    @SuppressWarnings("unchecked")
-    public <Params extends PersistentTaskParams> void startPersistentTask(String taskId, String taskName, @Nullable Params params,
-                                                                          ActionListener<PersistentTask<Params>> listener) {
-        StartPersistentTaskAction.Request createPersistentActionRequest =
-                new StartPersistentTaskAction.Request(taskId, taskName, params);
-        try {
-            executeAsyncWithOrigin(client, PERSISTENT_TASK_ORIGIN, StartPersistentTaskAction.INSTANCE, createPersistentActionRequest,
-                    ActionListener.wrap(o -> listener.onResponse((PersistentTask<Params>) o.getTask()), listener::onFailure));
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
+    public <Params extends PersistentTaskParams> void sendStartRequest(final String taskId,
+                                                                       final String taskName,
+                                                                       final @Nullable Params taskParams,
+                                                                       final ActionListener<PersistentTask<Params>> listener) {
+        @SuppressWarnings("unchecked")
+        final ActionListener<PersistentTask<?>> wrappedListener =
+            ActionListener.wrap(t -> listener.onResponse((PersistentTask<Params>) t), listener::onFailure);
+        StartPersistentTaskAction.Request request = new StartPersistentTaskAction.Request(taskId, taskName, taskParams);
+        execute(request, StartPersistentTaskAction.INSTANCE, wrappedListener);
     }
 
     /**
-     * Notifies the PersistentTasksClusterService about successful (failure == null) completion of a task or its failure
+     * Notifies the master node about the completion of a persistent task.
+     * <p>
+     * When {@code failure} is {@code null}, the persistent task is considered as successfully completed.
      */
-    public void sendCompletionNotification(String taskId, long allocationId, Exception failure,
-                                           ActionListener<PersistentTask<?>> listener) {
-        CompletionPersistentTaskAction.Request restartRequest = new CompletionPersistentTaskAction.Request(taskId, allocationId, failure);
-        try {
-            executeAsyncWithOrigin(client, PERSISTENT_TASK_ORIGIN, CompletionPersistentTaskAction.INSTANCE, restartRequest,
-                    ActionListener.wrap(o -> listener.onResponse(o.getTask()), listener::onFailure));
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
+    public void sendCompletionRequest(final String taskId,
+                                      final long taskAllocationId,
+                                      final @Nullable Exception taskFailure,
+                                      final ActionListener<PersistentTask<?>> listener) {
+        CompletionPersistentTaskAction.Request request = new CompletionPersistentTaskAction.Request(taskId, taskAllocationId, taskFailure);
+        execute(request, CompletionPersistentTaskAction.INSTANCE, listener);
     }
 
     /**
-     * Cancels a locally running task using the task manager
+     * Cancels a locally running task using the Task Manager API
      */
-    void sendTaskManagerCancellation(long taskId, ActionListener<CancelTasksResponse> listener) {
-        DiscoveryNode localNode = clusterService.localNode();
-        CancelTasksRequest cancelTasksRequest = new CancelTasksRequest();
-        cancelTasksRequest.setTaskId(new TaskId(localNode.getId(), taskId));
-        cancelTasksRequest.setReason("persistent action was removed");
+    void sendCancelRequest(final long taskId, final String reason, final ActionListener<CancelTasksResponse> listener) {
+        CancelTasksRequest request = new CancelTasksRequest();
+        request.setTaskId(new TaskId(clusterService.localNode().getId(), taskId));
+        request.setReason(reason);
         try {
-            executeAsyncWithOrigin(client.threadPool().getThreadContext(), PERSISTENT_TASK_ORIGIN, cancelTasksRequest, listener,
-                    client.admin().cluster()::cancelTasks);
+            final ThreadContext threadContext = client.threadPool().getThreadContext();
+            final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
+
+            try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, PERSISTENT_TASK_ORIGIN)) {
+                client.admin().cluster().cancelTasks(request, new ContextPreservingActionListener<>(supplier, listener));
+            }
         } catch (Exception e) {
             listener.onFailure(e);
         }
     }
 
     /**
-     * Updates status of the persistent task.
+     * Notifies the master node that the state of a persistent task has changed.
      * <p>
      * Persistent task implementers shouldn't call this method directly and use
      * {@link AllocatedPersistentTask#updatePersistentStatus} instead
      */
-    void updateStatus(String taskId, long allocationId, Task.Status status, ActionListener<PersistentTask<?>> listener) {
-        UpdatePersistentTaskStatusAction.Request updateStatusRequest =
-                new UpdatePersistentTaskStatusAction.Request(taskId, allocationId, status);
-        try {
-            executeAsyncWithOrigin(client, PERSISTENT_TASK_ORIGIN, UpdatePersistentTaskStatusAction.INSTANCE, updateStatusRequest,
-                    ActionListener.wrap(o -> listener.onResponse(o.getTask()), listener::onFailure));
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
+    void updateStatus(final String taskId,
+                      final long taskAllocationID,
+                      final Task.Status status,
+                      final ActionListener<PersistentTask<?>> listener) {
+        UpdatePersistentTaskStatusAction.Request request = new UpdatePersistentTaskStatusAction.Request(taskId, taskAllocationID, status);
+        execute(request, UpdatePersistentTaskStatusAction.INSTANCE, listener);
     }
 
     /**
-     * Cancels if needed and removes a persistent task
+     * Notifies the master node to remove a persistent task from the cluster state
      */
-    public void cancelPersistentTask(String taskId, ActionListener<PersistentTask<?>> listener) {
-        RemovePersistentTaskAction.Request removeRequest = new RemovePersistentTaskAction.Request(taskId);
-        try {
-            executeAsyncWithOrigin(client, PERSISTENT_TASK_ORIGIN, RemovePersistentTaskAction.INSTANCE, removeRequest,
-                ActionListener.wrap(o -> listener.onResponse(o.getTask()), listener::onFailure));
-        } catch (Exception e) {
-            listener.onFailure(e);
-        }
+    public void sendRemoveRequest(final String taskId, final ActionListener<PersistentTask<?>> listener) {
+        RemovePersistentTaskAction.Request request = new RemovePersistentTaskAction.Request(taskId);
+        execute(request, RemovePersistentTaskAction.INSTANCE, listener);
     }
 
     /**
-     * Checks if the persistent task with giving id (taskId) has the desired state and if it doesn't
-     * waits of it.
+     * Executes an asynchronous persistent task action using the client.
+     * <p>
+     * The origin is set in the context and the listener is wrapped to ensure the proper context is restored
      */
-    public void waitForPersistentTaskStatus(String taskId, Predicate<PersistentTask<?>> predicate, @Nullable TimeValue timeout,
-                                            WaitForPersistentTaskStatusListener<?> listener) {
-        ClusterStateObserver stateObserver = new ClusterStateObserver(clusterService, timeout, logger, threadPool.getThreadContext());
-        if (predicate.test(PersistentTasksCustomMetaData.getTaskWithId(stateObserver.setAndGetObservedState(), taskId))) {
-            listener.onResponse(PersistentTasksCustomMetaData.getTaskWithId(stateObserver.setAndGetObservedState(), taskId));
+    private <Req extends ActionRequest, Resp extends PersistentTaskResponse, Builder extends ActionRequestBuilder<Req, Resp, Builder>>
+        void execute(final Req request, final Action<Req, Resp, Builder> action, final ActionListener<PersistentTask<?>> listener) {
+            try {
+                final ThreadContext threadContext = client.threadPool().getThreadContext();
+                final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
+
+                try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, PERSISTENT_TASK_ORIGIN)) {
+                    client.execute(action, request,
+                        new ContextPreservingActionListener<>(supplier,
+                            ActionListener.wrap(r -> listener.onResponse(r.getTask()), listener::onFailure)));
+                }
+            } catch (Exception e) {
+                listener.onFailure(e);
+            }
+    }
+
+    /**
+     * Waits for a given persistent task to comply with a given predicate, then call back the listener accordingly.
+     *
+     * @param taskId the persistent task id
+     * @param predicate the persistent task predicate to evaluate
+     * @param timeout a timeout for waiting
+     * @param listener the callback listener
+     */
+    public void waitForPersistentTaskCondition(final String taskId,
+                                               final Predicate<PersistentTask<?>> predicate,
+                                               final @Nullable TimeValue timeout,
+                                               final WaitForPersistentTaskListener<?> listener) {
+        final Predicate<ClusterState> clusterStatePredicate = clusterState ->
+            predicate.test(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId));
+
+        final ClusterStateObserver observer = new ClusterStateObserver(clusterService, timeout, logger, threadPool.getThreadContext());
+        final ClusterState clusterState = observer.setAndGetObservedState();
+        if (clusterStatePredicate.test(clusterState)) {
+            listener.onResponse(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId));
         } else {
-            stateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
+            observer.waitForNextChange(new ClusterStateObserver.Listener() {
                 @Override
                 public void onNewClusterState(ClusterState state) {
                     listener.onResponse(PersistentTasksCustomMetaData.getTaskWithId(state, taskId));
@@ -163,18 +188,28 @@ public class PersistentTasksService extends AbstractComponent {
                 public void onTimeout(TimeValue timeout) {
                     listener.onTimeout(timeout);
                 }
-            }, clusterState -> predicate.test(PersistentTasksCustomMetaData.getTaskWithId(clusterState, taskId)));
+            }, clusterStatePredicate);
         }
     }
 
-    public void waitForPersistentTasksStatus(Predicate<PersistentTasksCustomMetaData> predicate,
-            @Nullable TimeValue timeout, ActionListener<Boolean> listener) {
-        ClusterStateObserver stateObserver = new ClusterStateObserver(clusterService, timeout,
-                logger, threadPool.getThreadContext());
-        if (predicate.test(stateObserver.setAndGetObservedState().metaData().custom(PersistentTasksCustomMetaData.TYPE))) {
+    /**
+     * Waits for persistent tasks to comply with a given predicate, then call back the listener accordingly.
+     *
+     * @param predicate the predicate to evaluate
+     * @param timeout a timeout for waiting
+     * @param listener the callback listener
+     */
+    public void waitForPersistentTasksCondition(final Predicate<PersistentTasksCustomMetaData> predicate,
+                                                final @Nullable TimeValue timeout,
+                                                final ActionListener<Boolean> listener) {
+        final Predicate<ClusterState> clusterStatePredicate = clusterState ->
+            predicate.test(clusterState.metaData().custom(PersistentTasksCustomMetaData.TYPE));
+
+        final ClusterStateObserver observer = new ClusterStateObserver(clusterService, timeout, logger, threadPool.getThreadContext());
+        if (clusterStatePredicate.test(observer.setAndGetObservedState())) {
             listener.onResponse(true);
         } else {
-            stateObserver.waitForNextChange(new ClusterStateObserver.Listener() {
+            observer.waitForNextChange(new ClusterStateObserver.Listener() {
                 @Override
                 public void onNewClusterState(ClusterState state) {
                     listener.onResponse(true);
@@ -187,45 +222,15 @@ public class PersistentTasksService extends AbstractComponent {
 
                 @Override
                 public void onTimeout(TimeValue timeout) {
-                    listener.onFailure(new IllegalStateException("timed out after " + timeout));
+                    listener.onFailure(new IllegalStateException("Timed out when waiting for persistent tasks after " + timeout));
                 }
-            }, clusterState -> predicate.test(clusterState.metaData().custom(PersistentTasksCustomMetaData.TYPE)), timeout);
+            }, clusterStatePredicate, timeout);
         }
     }
 
-    public interface WaitForPersistentTaskStatusListener<Params extends PersistentTaskParams>
-            extends ActionListener<PersistentTask<Params>> {
+    public interface WaitForPersistentTaskListener<P extends PersistentTaskParams> extends ActionListener<PersistentTask<P>> {
         default void onTimeout(TimeValue timeout) {
-            onFailure(new IllegalStateException("timed out after " + timeout));
-        }
-    }
-
-    private static final String ACTION_ORIGIN_TRANSIENT_NAME = "action.origin";
-    private static final String PERSISTENT_TASK_ORIGIN = "persistent_tasks";
-
-    /**
-     * Executes a consumer after setting the origin and wrapping the listener so that the proper context is restored
-     */
-    public static <Request extends ActionRequest, Response extends ActionResponse> void executeAsyncWithOrigin(
-            ThreadContext threadContext, String origin, Request request, ActionListener<Response> listener,
-            BiConsumer<Request, ActionListener<Response>> consumer) {
-        final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
-        try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, origin)) {
-            consumer.accept(request, new ContextPreservingActionListener<>(supplier, listener));
-        }
-    }
-    /**
-     * Executes an asynchronous action using the provided client. The origin is set in the context and the listener
-     * is wrapped to ensure the proper context is restored
-     */
-    public static <Request extends ActionRequest, Response extends ActionResponse,
-            RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void executeAsyncWithOrigin(
-            Client client, String origin, Action<Request, Response, RequestBuilder> action, Request request,
-            ActionListener<Response> listener) {
-        final ThreadContext threadContext = client.threadPool().getThreadContext();
-        final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
-        try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, origin)) {
-            client.execute(action, request, new ContextPreservingActionListener<>(supplier, listener));
+            onFailure(new IllegalStateException("Timed out when waiting for persistent task after " + timeout));
         }
     }
 
@@ -234,5 +239,4 @@ public class PersistentTasksService extends AbstractComponent {
         threadContext.putTransient(ACTION_ORIGIN_TRANSIENT_NAME, origin);
         return storedContext;
     }
-
 }

server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorFullRestartIT.java

@@ -65,8 +65,7 @@ public class PersistentTasksExecutorFullRestartIT extends ESIntegTestCase {
             PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
             futures.add(future);
             taskIds[i] = UUIDs.base64UUID();
-            service.startPersistentTask(taskIds[i], TestPersistentTasksExecutor.NAME, randomBoolean() ? null : new TestParams("Blah"),
-                    future);
+            service.sendStartRequest(taskIds[i], TestPersistentTasksExecutor.NAME, randomBoolean() ? null : new TestParams("Blah"), future);
         }
 
         for (int i = 0; i < numberOfTasks; i++) {

server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java

@@ -30,7 +30,7 @@ import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
-import org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskStatusListener;
+import org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskListener;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.Status;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
 import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
@@ -69,15 +69,15 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
         assertNoRunningTasks();
     }
 
-    public static class WaitForPersistentTaskStatusFuture<Params extends PersistentTaskParams>
+    public static class WaitForPersistentTaskFuture<Params extends PersistentTaskParams>
             extends PlainActionFuture<PersistentTask<Params>>
-            implements WaitForPersistentTaskStatusListener<Params> {
+            implements WaitForPersistentTaskListener<Params> {
     }
 
     public void testPersistentActionFailure() throws Exception {
         PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
         PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
-        persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
+        persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
         long allocationId = future.get().getAllocationId();
         assertBusy(() -> {
             // Wait for the task to start
@@ -108,7 +108,7 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
         PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
         PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
         String taskId = UUIDs.base64UUID();
-        persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
+        persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
         long allocationId = future.get().getAllocationId();
         assertBusy(() -> {
             // Wait for the task to start
@@ -127,7 +127,7 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
             logger.info("Simulating errant completion notification");
             //try sending completion request with incorrect allocation id
             PlainActionFuture<PersistentTask<?>> failedCompletionNotificationFuture = new PlainActionFuture<>();
-            persistentTasksService.sendCompletionNotification(taskId, Long.MAX_VALUE, null, failedCompletionNotificationFuture);
+            persistentTasksService.sendCompletionRequest(taskId, Long.MAX_VALUE, null, failedCompletionNotificationFuture);
             assertThrows(failedCompletionNotificationFuture, ResourceNotFoundException.class);
             // Make sure that the task is still running
             assertThat(client().admin().cluster().prepareListTasks().setActions(TestPersistentTasksExecutor.NAME + "[c]")
@@ -142,7 +142,7 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
         PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
         TestParams testParams = new TestParams("Blah");
         testParams.setExecutorNodeAttr("test");
-        persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, testParams, future);
+        persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, testParams, future);
         String taskId = future.get().getId();
 
         Settings nodeSettings = Settings.builder().put(nodeSettings(0)).put("node.attr.test_attr", "test").build();
@@ -169,14 +169,14 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
 
         // Remove the persistent task
         PlainActionFuture<PersistentTask<?>> removeFuture = new PlainActionFuture<>();
-        persistentTasksService.cancelPersistentTask(taskId, removeFuture);
+        persistentTasksService.sendRemoveRequest(taskId, removeFuture);
         assertEquals(removeFuture.get().getId(), taskId);
     }
 
     public void testPersistentActionStatusUpdate() throws Exception {
         PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
         PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
-        persistentTasksService.startPersistentTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
+        persistentTasksService.sendStartRequest(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
         String taskId = future.get().getId();
 
         assertBusy(() -> {
@@ -200,16 +200,16 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
                     .get().getTasks().size(), equalTo(1));
 
             int finalI = i;
-            WaitForPersistentTaskStatusFuture<?> future1 = new WaitForPersistentTaskStatusFuture<>();
-            persistentTasksService.waitForPersistentTaskStatus(taskId,
+            WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
+            persistentTasksService.waitForPersistentTaskCondition(taskId,
                     task -> task != null && task.getStatus() != null && task.getStatus().toString() != null &&
                             task.getStatus().toString().equals("{\"phase\":\"phase " + (finalI + 1) + "\"}"),
                     TimeValue.timeValueSeconds(10), future1);
             assertThat(future1.get().getId(), equalTo(taskId));
         }
 
-        WaitForPersistentTaskStatusFuture<?> future1 = new WaitForPersistentTaskStatusFuture<>();
-        persistentTasksService.waitForPersistentTaskStatus(taskId,
+        WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
+        persistentTasksService.waitForPersistentTaskCondition(taskId,
                 task -> false, TimeValue.timeValueMillis(10), future1);
 
         assertThrows(future1, IllegalStateException.class, "timed out after 10ms");
@@ -220,8 +220,8 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
                 " and allocation id -2 doesn't exist");
 
         // Wait for the task to disappear
-        WaitForPersistentTaskStatusFuture<?> future2 = new WaitForPersistentTaskStatusFuture<>();
-        persistentTasksService.waitForPersistentTaskStatus(taskId, Objects::isNull, TimeValue.timeValueSeconds(10), future2);
+        WaitForPersistentTaskFuture<?> future2 = new WaitForPersistentTaskFuture<>();
+        persistentTasksService.waitForPersistentTaskCondition(taskId, Objects::isNull, TimeValue.timeValueSeconds(10), future2);
 
         logger.info("Completing the running task");
         // Complete the running task and make sure it finishes properly
@@ -235,11 +235,11 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
         PersistentTasksService persistentTasksService = internalCluster().getInstance(PersistentTasksService.class);
         PlainActionFuture<PersistentTask<TestParams>> future = new PlainActionFuture<>();
         String taskId = UUIDs.base64UUID();
-        persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
+        persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future);
         future.get();
 
         PlainActionFuture<PersistentTask<TestParams>> future2 = new PlainActionFuture<>();
-        persistentTasksService.startPersistentTask(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future2);
+        persistentTasksService.sendStartRequest(taskId, TestPersistentTasksExecutor.NAME, new TestParams("Blah"), future2);
         assertThrows(future2, ResourceAlreadyExistsException.class);
 
         assertBusy(() -> {

server/src/test/java/org/elasticsearch/persistent/PersistentTasksNodeServiceTests.java

@@ -235,14 +235,14 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
         AtomicReference<ActionListener<CancelTasksResponse>> capturedListener = new AtomicReference<>();
         PersistentTasksService persistentTasksService = new PersistentTasksService(Settings.EMPTY, null, null, null) {
             @Override
-            public void sendTaskManagerCancellation(long taskId, ActionListener<CancelTasksResponse> listener) {
+            void sendCancelRequest(final long taskId, final String reason, final ActionListener<CancelTasksResponse> listener) {
                 capturedTaskId.set(taskId);
                 capturedListener.set(listener);
             }
 
             @Override
-            public void sendCompletionNotification(String taskId, long allocationId, Exception failure,
-                                                   ActionListener<PersistentTask<?>> listener) {
+            public void sendCompletionRequest(final String taskId, final long taskAllocationId,
+                                              final Exception taskFailure, final ActionListener<PersistentTask<?>> listener) {
                 fail("Shouldn't be called during Cluster State cancellation");
             }
         };

server/src/test/java/org/elasticsearch/persistent/decider/EnableAssignmentDeciderIT.java

@@ -71,7 +71,7 @@ public class EnableAssignmentDeciderIT extends ESIntegTestCase {
         final CountDownLatch latch = new CountDownLatch(numberOfTasks);
         for (int i = 0; i < numberOfTasks; i++) {
             PersistentTasksService service = internalCluster().getInstance(PersistentTasksService.class);
-            service.startPersistentTask("task_" + i, TestPersistentTasksExecutor.NAME, randomTaskParams(),
+            service.sendStartRequest("task_" + i, TestPersistentTasksExecutor.NAME, randomTaskParams(),
                 new ActionListener<PersistentTask<PersistentTaskParams>>() {
                     @Override
                     public void onResponse(PersistentTask<PersistentTaskParams> task) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportCloseJobAction.java

@@ -315,7 +315,7 @@ public class TransportCloseJobAction extends TransportTasksAction<TransportOpenJ
             PersistentTasksCustomMetaData.PersistentTask<?> jobTask = MlMetadata.getJobTask(jobId, tasks);
             if (jobTask != null) {
                 auditor.info(jobId, Messages.JOB_AUDIT_FORCE_CLOSING);
-                persistentTasksService.cancelPersistentTask(jobTask.getId(),
+                persistentTasksService.sendRemoveRequest(jobTask.getId(),
                         new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                             @Override
                             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {
@@ -400,7 +400,7 @@ public class TransportCloseJobAction extends TransportTasksAction<TransportOpenJ
     // so wait for that to happen here.
     void waitForJobClosed(CloseJobAction.Request request, WaitForCloseRequest waitForCloseRequest, CloseJobAction.Response response,
                           ActionListener<CloseJobAction.Response> listener) {
-        persistentTasksService.waitForPersistentTasksStatus(persistentTasksCustomMetaData -> {
+        persistentTasksService.waitForPersistentTasksCondition(persistentTasksCustomMetaData -> {
             for (String persistentTaskId : waitForCloseRequest.persistentTaskIds) {
                 if (persistentTasksCustomMetaData.getTask(persistentTaskId) != null) {
                     return false;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDatafeedAction.java

@@ -90,7 +90,7 @@ public class TransportDeleteDatafeedAction extends TransportMasterNodeAction<Del
         if (datafeedTask == null) {
             listener.onResponse(true);
         } else {
-            persistentTasksService.cancelPersistentTask(datafeedTask.getId(),
+            persistentTasksService.sendRemoveRequest(datafeedTask.getId(),
                     new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                         @Override
                         public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteJobAction.java

@@ -182,7 +182,7 @@ public class TransportDeleteJobAction extends TransportMasterNodeAction<DeleteJo
         if (jobTask == null) {
             listener.onResponse(null);
         } else {
-            persistentTasksService.cancelPersistentTask(jobTask.getId(),
+            persistentTasksService.sendRemoveRequest(jobTask.getId(),
                     new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                         @Override
                         public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java

@@ -465,7 +465,7 @@ public class TransportOpenJobAction extends TransportMasterNodeAction<OpenJobAct
 
             // Step 4. Start job task
             ActionListener<PutJobAction.Response> establishedMemoryUpdateListener = ActionListener.wrap(
-                    response -> persistentTasksService.startPersistentTask(MlMetadata.jobTaskId(jobParams.getJobId()),
+                    response -> persistentTasksService.sendStartRequest(MlMetadata.jobTaskId(jobParams.getJobId()),
                             OpenJobAction.TASK_NAME, jobParams, finalListener),
                     listener::onFailure
             );
@@ -518,8 +518,8 @@ public class TransportOpenJobAction extends TransportMasterNodeAction<OpenJobAct
 
     private void waitForJobStarted(String taskId, OpenJobAction.JobParams jobParams, ActionListener<OpenJobAction.Response> listener) {
         JobPredicate predicate = new JobPredicate();
-        persistentTasksService.waitForPersistentTaskStatus(taskId, predicate, jobParams.getTimeout(),
-                new PersistentTasksService.WaitForPersistentTaskStatusListener<OpenJobAction.JobParams>() {
+        persistentTasksService.waitForPersistentTaskCondition(taskId, predicate, jobParams.getTimeout(),
+                new PersistentTasksService.WaitForPersistentTaskListener<OpenJobAction.JobParams>() {
             @Override
             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask) {
                 if (predicate.exception != null) {
@@ -550,7 +550,7 @@ public class TransportOpenJobAction extends TransportMasterNodeAction<OpenJobAct
 
     private void cancelJobStart(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask, Exception exception,
                                 ActionListener<OpenJobAction.Response> listener) {
-        persistentTasksService.cancelPersistentTask(persistentTask.getId(),
+        persistentTasksService.sendRemoveRequest(persistentTask.getId(),
                 new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDatafeedAction.java

@@ -137,7 +137,7 @@ public class TransportStartDatafeedAction extends TransportMasterNodeAction<Star
             Job job = mlMetadata.getJobs().get(datafeed.getJobId());
             DataExtractorFactory.create(client, datafeed, job, ActionListener.wrap(
                     dataExtractorFactory ->
-                            persistentTasksService.startPersistentTask(MLMetadataField.datafeedTaskId(params.getDatafeedId()),
+                            persistentTasksService.sendStartRequest(MLMetadataField.datafeedTaskId(params.getDatafeedId()),
                             StartDatafeedAction.TASK_NAME, params, finalListener)
                     , listener::onFailure));
         } else {
@@ -156,8 +156,8 @@ public class TransportStartDatafeedAction extends TransportMasterNodeAction<Star
     private void waitForDatafeedStarted(String taskId, StartDatafeedAction.DatafeedParams params,
                                         ActionListener<StartDatafeedAction.Response> listener) {
         DatafeedPredicate predicate = new DatafeedPredicate();
-        persistentTasksService.waitForPersistentTaskStatus(taskId, predicate, params.getTimeout(),
-                new PersistentTasksService.WaitForPersistentTaskStatusListener<StartDatafeedAction.DatafeedParams>() {
+        persistentTasksService.waitForPersistentTaskCondition(taskId, predicate, params.getTimeout(),
+                new PersistentTasksService.WaitForPersistentTaskListener<StartDatafeedAction.DatafeedParams>() {
             @Override
             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<StartDatafeedAction.DatafeedParams> persistentTask) {
                 if (predicate.exception != null) {
@@ -184,7 +184,7 @@ public class TransportStartDatafeedAction extends TransportMasterNodeAction<Star
 
     private void cancelDatafeedStart(PersistentTasksCustomMetaData.PersistentTask<StartDatafeedAction.DatafeedParams> persistentTask,
                                      Exception exception, ActionListener<StartDatafeedAction.Response> listener) {
-        persistentTasksService.cancelPersistentTask(persistentTask.getId(),
+        persistentTasksService.sendRemoveRequest(persistentTask.getId(),
                 new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStopDatafeedAction.java

@@ -190,7 +190,7 @@ public class TransportStopDatafeedAction extends TransportTasksAction<TransportS
         for (String datafeedId : startedDatafeeds) {
             PersistentTasksCustomMetaData.PersistentTask<?> datafeedTask = MlMetadata.getDatafeedTask(datafeedId, tasks);
             if (datafeedTask != null) {
-                persistentTasksService.cancelPersistentTask(datafeedTask.getId(),
+                persistentTasksService.sendRemoveRequest(datafeedTask.getId(),
                         new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
@@ -275,7 +275,7 @@ public class TransportStopDatafeedAction extends TransportTasksAction<TransportS
     void waitForDatafeedStopped(List<String> datafeedPersistentTaskIds, StopDatafeedAction.Request request,
                                 StopDatafeedAction.Response response,
                                 ActionListener<StopDatafeedAction.Response> listener) {
-        persistentTasksService.waitForPersistentTasksStatus(persistentTasksCustomMetaData -> {
+        persistentTasksService.waitForPersistentTasksCondition(persistentTasksCustomMetaData -> {
             for (String persistentTaskId: datafeedPersistentTaskIds) {
                 if (persistentTasksCustomMetaData.getTask(persistentTaskId) != null) {
                     return false;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java

@@ -49,7 +49,7 @@ import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
-import static org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskStatusListener;
+import static org.elasticsearch.persistent.PersistentTasksService.WaitForPersistentTaskListener;
 
 public class DatafeedManager extends AbstractComponent {
 
@@ -391,8 +391,8 @@ public class DatafeedManager extends AbstractComponent {
                 return;
             }
 
-            task.waitForPersistentTaskStatus(Objects::isNull, TimeValue.timeValueSeconds(20),
-                            new WaitForPersistentTaskStatusListener<StartDatafeedAction.DatafeedParams>() {
+            task.waitForPersistentTask(Objects::isNull, TimeValue.timeValueSeconds(20),
+                            new WaitForPersistentTaskListener<StartDatafeedAction.DatafeedParams>() {
                 @Override
                 public void onResponse(PersistentTask<StartDatafeedAction.DatafeedParams> persistentTask) {
                     CloseJobAction.Request closeJobRequest = new CloseJobAction.Request(getJobId());

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportDeleteRollupJobAction.java

@@ -59,14 +59,14 @@ public class TransportDeleteRollupJobAction
         TimeValue timeout = new TimeValue(60, TimeUnit.SECONDS); // TODO make this a config option
 
         // Step 1. Cancel the persistent task
-        persistentTasksService.cancelPersistentTask(jobId, new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
+        persistentTasksService.sendRemoveRequest(jobId, new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
             @Override
             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
                 logger.debug("Request to cancel Task for Rollup job [" + jobId + "] successful.");
 
                 // Step 2. Wait for the task to finish cancellation internally
-                persistentTasksService.waitForPersistentTaskStatus(jobId, Objects::isNull, timeout,
-                        new PersistentTasksService.WaitForPersistentTaskStatusListener<RollupJob>() {
+                persistentTasksService.waitForPersistentTaskCondition(jobId, Objects::isNull, timeout,
+                        new PersistentTasksService.WaitForPersistentTaskListener<RollupJob>() {
                             @Override
                             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<RollupJob> task) {
                                 logger.debug("Task for Rollup job [" + jobId + "] successfully canceled.");

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportPutRollupJobAction.java

@@ -205,7 +205,7 @@ public class TransportPutRollupJobAction extends TransportMasterNodeAction<PutRo
     static void startPersistentTask(RollupJob job, ActionListener<PutRollupJobAction.Response> listener,
                                     PersistentTasksService persistentTasksService) {
 
-        persistentTasksService.startPersistentTask(job.getConfig().getId(), RollupField.TASK_NAME, job,
+        persistentTasksService.sendStartRequest(job.getConfig().getId(), RollupField.TASK_NAME, job,
                 ActionListener.wrap(
                         rollupConfigPersistentTask -> waitForRollupStarted(job, listener, persistentTasksService),
                         e -> {
@@ -220,8 +220,8 @@ public class TransportPutRollupJobAction extends TransportMasterNodeAction<PutRo
 
     private static void waitForRollupStarted(RollupJob job, ActionListener<PutRollupJobAction.Response> listener,
                                              PersistentTasksService persistentTasksService) {
-        persistentTasksService.waitForPersistentTaskStatus(job.getConfig().getId(), Objects::nonNull, job.getConfig().getTimeout(),
-                new PersistentTasksService.WaitForPersistentTaskStatusListener<RollupJob>() {
+        persistentTasksService.waitForPersistentTaskCondition(job.getConfig().getId(), Objects::nonNull, job.getConfig().getTimeout(),
+                new PersistentTasksService.WaitForPersistentTaskListener<RollupJob>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<RollupJob> task) {
                         listener.onResponse(new PutRollupJobAction.Response(true));

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/PutJobStateMachineTests.java

@@ -300,11 +300,11 @@ public class PutJobStateMachineTests extends ESTestCase {
         doAnswer(invocation -> {
             requestCaptor.getValue().onFailure(new ResourceAlreadyExistsException(job.getConfig().getRollupIndex()));
             return null;
-        }).when(tasksService).startPersistentTask(eq(job.getConfig().getId()),
+        }).when(tasksService).sendStartRequest(eq(job.getConfig().getId()),
                 eq(RollupField.TASK_NAME), eq(job), requestCaptor.capture());
 
         TransportPutRollupJobAction.startPersistentTask(job, testListener, tasksService);
-        verify(tasksService).startPersistentTask(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), any());
+        verify(tasksService).sendStartRequest(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), any());
     }
 
     @SuppressWarnings("unchecked")
@@ -326,18 +326,18 @@ public class PutJobStateMachineTests extends ESTestCase {
                     mock(PersistentTasksCustomMetaData.Assignment.class));
             requestCaptor.getValue().onResponse(response);
             return null;
-        }).when(tasksService).startPersistentTask(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), requestCaptor.capture());
+        }).when(tasksService).sendStartRequest(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), requestCaptor.capture());
 
-        ArgumentCaptor<PersistentTasksService.WaitForPersistentTaskStatusListener> requestCaptor2
-                = ArgumentCaptor.forClass(PersistentTasksService.WaitForPersistentTaskStatusListener.class);
+        ArgumentCaptor<PersistentTasksService.WaitForPersistentTaskListener> requestCaptor2
+                = ArgumentCaptor.forClass(PersistentTasksService.WaitForPersistentTaskListener.class);
         doAnswer(invocation -> {
             // Bail here with an error, further testing will happen through tests of #startPersistentTask
             requestCaptor2.getValue().onFailure(new RuntimeException("Ending"));
             return null;
-        }).when(tasksService).waitForPersistentTaskStatus(eq(job.getConfig().getId()), any(), any(), requestCaptor2.capture());
+        }).when(tasksService).waitForPersistentTaskCondition(eq(job.getConfig().getId()), any(), any(), requestCaptor2.capture());
 
         TransportPutRollupJobAction.startPersistentTask(job, testListener, tasksService);
-        verify(tasksService).startPersistentTask(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), any());
-        verify(tasksService).waitForPersistentTaskStatus(eq(job.getConfig().getId()), any(), any(), any());
+        verify(tasksService).sendStartRequest(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), any());
+        verify(tasksService).waitForPersistentTaskCondition(eq(job.getConfig().getId()), any(), any(), any());
     }
 }