Required changes after merging in master.

Author: martijnvg

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

@@ -115,7 +115,7 @@ public void waitForPersistentTask(final Predicate<PersistentTasksCustomMetaData.
         persistentTasksService.waitForPersistentTaskCondition(persistentTaskId, predicate, timeout, listener);
     }
 
-    final boolean isCompleted() {
+    protected final boolean isCompleted() {
         return state.get() ==  State.COMPLETED;
     }
 

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowIndexAction.java

@@ -238,7 +238,7 @@ void start(Request request, String clusterNameAlias, IndexMetaData leaderIndexMe
                             new ShardId(followIndexMetadata.getIndex(), shardId),
                             new ShardId(leaderIndexMetadata.getIndex(), shardId),
                             request.batchSize, request.concurrentProcessors, request.processorMaxTranslogBytes, filteredHeaders);
-                    persistentTasksService.startPersistentTask(taskId, ShardFollowTask.NAME, shardFollowTask,
+                    persistentTasksService.sendStartRequest(taskId, ShardFollowTask.NAME, shardFollowTask,
                             new ActionListener<PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask>>() {
                                 @Override
                                 public void onResponse(PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask> task) {

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java

@@ -34,6 +34,10 @@ protected void onCancelled() {
         markAsCompleted();
     }
 
+    public boolean isRunning() {
+        return isCancelled() == false && isCompleted() == false;
+    }
+
     void updateProcessedGlobalCheckpoint(long processedGlobalCheckpoint) {
         this.processedGlobalCheckpoint.set(processedGlobalCheckpoint);
     }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java

@@ -130,7 +130,7 @@ protected void nodeOperation(AllocatedPersistentTask task, ShardFollowTask param
     void prepare(Client leaderClient, Client followerClient, ShardFollowNodeTask task, ShardFollowTask params,
                  long followGlobalCheckPoint,
                  IndexMetadataVersionChecker imdVersionChecker) {
-        if (task.getState() != AllocatedPersistentTask.State.STARTED) {
+        if (task.isRunning() == false) {
             // TODO: need better cancellation control
             return;
         }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/UnfollowIndexAction.java

@@ -141,7 +141,7 @@ protected void doExecute(Request request, ActionListener<Response> listener) {
                 for (int i = 0; i < numShards; i++) {
                     final int shardId = i;
                     String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
-                    persistentTasksService.cancelPersistentTask(taskId,
+                    persistentTasksService.sendRemoveRequest(taskId,
                             new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                         @Override
                         public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> task) {