From 8202bf212cddafcff79ca78d00efdee76e6d6ff4 Mon Sep 17 00:00:00 2001 From: Igor Motov Date: Mon, 21 Mar 2016 16:02:08 -0400 Subject: [PATCH] Don't wait for completion of list tasks tasks when wait_for_completion flag is set Waiting for completion of list tasks tasks can cause an infinite loop of a list tasks task waiting for its own completion or completion of its children. To reproduce run: ``` curl "localhost:9200/_tasks?wait_for_completion" ``` --- .../node/tasks/list/TransportListTasksAction.java | 8 +++++++- .../cluster/node/tasks/RestListTasksAction.java | 3 +++ .../action/admin/cluster/node/tasks/TasksIT.java | 13 +++++++++++++ 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java index d5175e263e7..e6ea002a794 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java @@ -84,7 +84,13 @@ public class TransportListTasksAction extends TransportTasksAction { while (System.nanoTime() - timeoutTime < 0) { - if (taskManager.getTask(t.getId()) == null) { + Task task = taskManager.getTask(t.getId()); + if (task == null) { + return; + } + if (task.getAction().startsWith(ListTasksAction.NAME)) { + // It doesn't make sense to wait for List Tasks and it can cause an infinite loop of the task waiting + // for itself of one of its child tasks return; } try { diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java index 9a9d1991298..168d091968e 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java @@ -24,6 +24,7 @@ import org.elasticsearch.client.Client; import org.elasticsearch.common.Strings; import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.rest.BaseRestHandler; import org.elasticsearch.rest.RestChannel; import org.elasticsearch.rest.RestController; @@ -51,6 +52,7 @@ public class RestListTasksAction extends BaseRestHandler { String[] actions = Strings.splitStringByCommaToArray(request.param("actions")); TaskId parentTaskId = new TaskId(request.param("parent_task_id")); boolean waitForCompletion = request.paramAsBoolean("wait_for_completion", false); + TimeValue timeout = request.paramAsTime("timeout", null); ListTasksRequest listTasksRequest = new ListTasksRequest(); listTasksRequest.setTaskId(taskId); @@ -59,6 +61,7 @@ public class RestListTasksAction extends BaseRestHandler { listTasksRequest.setActions(actions); listTasksRequest.setParentTaskId(parentTaskId); listTasksRequest.setWaitForCompletion(waitForCompletion); + listTasksRequest.setTimeout(timeout); client.admin().cluster().listTasks(listTasksRequest, new RestToXContentListener<>(channel)); } } diff --git a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java index 90dbfaa8c60..e9ade3ef5af 100644 --- a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java +++ b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java @@ -21,6 +21,7 @@ package org.elasticsearch.action.admin.cluster.node.tasks; import org.elasticsearch.ElasticsearchTimeoutException; import org.elasticsearch.action.FailedNodeException; import org.elasticsearch.action.ListenableActionFuture; +import org.elasticsearch.action.TaskOperationFailure; import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction; import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse; import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksAction; @@ -59,6 +60,7 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import static org.elasticsearch.common.unit.TimeValue.timeValueMillis; +import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds; import static org.hamcrest.Matchers.either; import static org.hamcrest.Matchers.emptyCollectionOf; import static org.hamcrest.Matchers.greaterThanOrEqualTo; @@ -406,6 +408,17 @@ public class TasksIT extends ESIntegTestCase { assertThat(waitResponseFuture.get().getTasks(), emptyCollectionOf(TaskInfo.class)); } + public void testTasksWaitForAllTask() throws Exception { + // Spin up a request to wait for all tasks in the cluster to make sure it doesn't cause an infinite loop + ListTasksResponse response = client().admin().cluster().prepareListTasks().setWaitForCompletion(true) + .setTimeout(timeValueSeconds(10)).get(); + + // It should finish quickly and without complaint and list the list tasks themselves + assertThat(response.getNodeFailures(), emptyCollectionOf(FailedNodeException.class)); + assertThat(response.getTaskFailures(), emptyCollectionOf(TaskOperationFailure.class)); + assertThat(response.getTasks().size(), greaterThanOrEqualTo(1)); + } + @Override public void tearDown() throws Exception { for (Map.Entry, RecordingTaskManagerListener> entry : listeners.entrySet()) {