waitForPersistentTaskCondition

elastic · May 29, 2018 · ab05475 · ab05475
1 parent b4a50b7
commit ab05475
Show file tree

Hide file tree

Showing 10 changed files with 19 additions and 19 deletions.
diff --git a/server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java b/server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java
@@ -112,7 +112,7 @@ public long getAllocationId() {
     public void waitForPersistentTask(final Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
                                       final @Nullable TimeValue timeout,
                                       final PersistentTasksService.WaitForPersistentTaskListener<?> listener) {
-        persistentTasksService.waitForPersistentTask(persistentTaskId, predicate, timeout, listener);
+        persistentTasksService.waitForPersistentTaskCondition(persistentTaskId, predicate, timeout, listener);
     }
 
     final boolean isCompleted() {

diff --git a/server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java b/server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java
@@ -161,10 +161,10 @@ void execute(final Req request, final Action<Req, Resp, Builder> action, final A
      * @param timeout a timeout for waiting
      * @param listener the callback listener
      */
-    public void waitForPersistentTask(final String taskId,
-                                      final Predicate<PersistentTask<?>> predicate,
-                                      final @Nullable TimeValue timeout,
-                                      final WaitForPersistentTaskListener<?> 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));
 
@@ -199,9 +199,9 @@ public void onTimeout(TimeValue timeout) {
      * @param timeout a timeout for waiting
      * @param listener the callback listener
      */
-    public void waitForPersistentTasks(final Predicate<PersistentTasksCustomMetaData> predicate,
-                                       final @Nullable TimeValue timeout,
-                                       final ActionListener<Boolean> 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));
 

diff --git a/server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java b/server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java
@@ -201,15 +201,15 @@ public void testPersistentActionStatusUpdate() throws Exception {
 
             int finalI = i;
             WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
-            persistentTasksService.waitForPersistentTask(taskId,
+            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));
         }
 
         WaitForPersistentTaskFuture<?> future1 = new WaitForPersistentTaskFuture<>();
-        persistentTasksService.waitForPersistentTask(taskId,
+        persistentTasksService.waitForPersistentTaskCondition(taskId,
                 task -> false, TimeValue.timeValueMillis(10), future1);
 
         assertThrows(future1, IllegalStateException.class, "timed out after 10ms");
@@ -221,7 +221,7 @@ public void testPersistentActionStatusUpdate() throws Exception {
 
         // Wait for the task to disappear
         WaitForPersistentTaskFuture<?> future2 = new WaitForPersistentTaskFuture<>();
-        persistentTasksService.waitForPersistentTask(taskId, Objects::isNull, TimeValue.timeValueSeconds(10), future2);
+        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

diff --git a/...ck/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportCloseJobAction.java b/...ck/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportCloseJobAction.java
@@ -400,7 +400,7 @@ public boolean hasJobsToWaitFor() {
     // so wait for that to happen here.
     void waitForJobClosed(CloseJobAction.Request request, WaitForCloseRequest waitForCloseRequest, CloseJobAction.Response response,
                           ActionListener<CloseJobAction.Response> listener) {
-        persistentTasksService.waitForPersistentTasks(persistentTasksCustomMetaData -> {
+        persistentTasksService.waitForPersistentTasksCondition(persistentTasksCustomMetaData -> {
             for (String persistentTaskId : waitForCloseRequest.persistentTaskIds) {
                 if (persistentTasksCustomMetaData.getTask(persistentTaskId) != null) {
                     return false;

diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java
@@ -518,7 +518,7 @@ public void onFailure(Exception e) {
 
     private void waitForJobStarted(String taskId, OpenJobAction.JobParams jobParams, ActionListener<OpenJobAction.Response> listener) {
         JobPredicate predicate = new JobPredicate();
-        persistentTasksService.waitForPersistentTask(taskId, predicate, jobParams.getTimeout(),
+        persistentTasksService.waitForPersistentTaskCondition(taskId, predicate, jobParams.getTimeout(),
                 new PersistentTasksService.WaitForPersistentTaskListener<OpenJobAction.JobParams>() {
             @Override
             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask) {

diff --git a/...ugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDatafeedAction.java b/...ugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDatafeedAction.java
@@ -156,7 +156,7 @@ protected ClusterBlockException checkBlock(StartDatafeedAction.Request request,
     private void waitForDatafeedStarted(String taskId, StartDatafeedAction.DatafeedParams params,
                                         ActionListener<StartDatafeedAction.Response> listener) {
         DatafeedPredicate predicate = new DatafeedPredicate();
-        persistentTasksService.waitForPersistentTask(taskId, predicate, params.getTimeout(),
+        persistentTasksService.waitForPersistentTaskCondition(taskId, predicate, params.getTimeout(),
                 new PersistentTasksService.WaitForPersistentTaskListener<StartDatafeedAction.DatafeedParams>() {
             @Override
             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<StartDatafeedAction.DatafeedParams> persistentTask) {

diff --git a/...lugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStopDatafeedAction.java b/...lugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStopDatafeedAction.java
@@ -275,7 +275,7 @@ private void sendResponseOrFailure(String datafeedId, ActionListener<StopDatafee
     void waitForDatafeedStopped(List<String> datafeedPersistentTaskIds, StopDatafeedAction.Request request,
                                 StopDatafeedAction.Response response,
                                 ActionListener<StopDatafeedAction.Response> listener) {
-        persistentTasksService.waitForPersistentTasks(persistentTasksCustomMetaData -> {
+        persistentTasksService.waitForPersistentTasksCondition(persistentTasksCustomMetaData -> {
             for (String persistentTaskId: datafeedPersistentTaskIds) {
                 if (persistentTasksCustomMetaData.getTask(persistentTaskId) != null) {
                     return false;

diff --git a/...p/src/main/java/org/elasticsearch/xpack/rollup/action/TransportDeleteRollupJobAction.java b/...p/src/main/java/org/elasticsearch/xpack/rollup/action/TransportDeleteRollupJobAction.java
@@ -65,7 +65,7 @@ public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persisten
                 logger.debug("Request to cancel Task for Rollup job [" + jobId + "] successful.");
 
                 // Step 2. Wait for the task to finish cancellation internally
-                persistentTasksService.waitForPersistentTask(jobId, Objects::isNull, timeout,
+                persistentTasksService.waitForPersistentTaskCondition(jobId, Objects::isNull, timeout,
                         new PersistentTasksService.WaitForPersistentTaskListener<RollupJob>() {
                             @Override
                             public void onResponse(PersistentTasksCustomMetaData.PersistentTask<RollupJob> task) {

diff --git a/...llup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportPutRollupJobAction.java b/...llup/src/main/java/org/elasticsearch/xpack/rollup/action/TransportPutRollupJobAction.java
@@ -220,7 +220,7 @@ static void startPersistentTask(RollupJob job, ActionListener<PutRollupJobAction
 
     private static void waitForRollupStarted(RollupJob job, ActionListener<PutRollupJobAction.Response> listener,
                                              PersistentTasksService persistentTasksService) {
-        persistentTasksService.waitForPersistentTask(job.getConfig().getId(), Objects::nonNull, job.getConfig().getTimeout(),
+        persistentTasksService.waitForPersistentTaskCondition(job.getConfig().getId(), Objects::nonNull, job.getConfig().getTimeout(),
                 new PersistentTasksService.WaitForPersistentTaskListener<RollupJob>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<RollupJob> task) {

diff --git a/...n/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/PutJobStateMachineTests.java b/...n/rollup/src/test/java/org/elasticsearch/xpack/rollup/action/PutJobStateMachineTests.java
@@ -334,10 +334,10 @@ public void testStartTask() {
             // Bail here with an error, further testing will happen through tests of #startPersistentTask
             requestCaptor2.getValue().onFailure(new RuntimeException("Ending"));
             return null;
-        }).when(tasksService).waitForPersistentTask(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).sendStartRequest(eq(job.getConfig().getId()), eq(RollupField.TASK_NAME), eq(job), any());
-        verify(tasksService).waitForPersistentTask(eq(job.getConfig().getId()), any(), any(), any());
+        verify(tasksService).waitForPersistentTaskCondition(eq(job.getConfig().getId()), any(), any(), any());
     }
 }