Fix cat tasks api params in spec and handler (#66294)

This commit fixes the cat tasks api parameter specification and the
handler so that the parameters are consumed during request preparation.

Closes #59493
Backport of #66272

rest-api-spec/src/main/resources/rest-api-spec/api/cat.tasks.json

@@ -20,7 +20,7 @@
         "type":"string",
         "description":"a short version of the Accept header, e.g. json, yaml"
       },
-      "node_id":{
+      "nodes":{
         "type":"list",
         "description":"A comma-separated list of node IDs or names to limit the returned information; use `_local` to return information from the node you're connecting to, leave empty to get information from all nodes"
       },
@@ -32,9 +32,9 @@
         "type":"boolean",
         "description":"Return detailed task information (default: false)"
       },
-      "parent_task":{
-        "type":"number",
-        "description":"Return tasks with specified parent task id. Set to -1 to return all."
+      "parent_task_id":{
+        "type":"string",
+        "description":"Return tasks with specified parent task id (node_id:task_number). Set to -1 to return all."
       },
       "h":{
         "type":"list",

server/src/main/java/org/elasticsearch/rest/action/cat/RestTasksAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.rest.action.cat;
 
+import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.TaskGroup;
 import org.elasticsearch.client.node.NodeClient;
@@ -72,8 +73,9 @@ protected void documentation(StringBuilder sb) {
 
     @Override
     public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
+        final ListTasksRequest listTasksRequest = generateListTasksRequest(request);
         return channel ->
-                client.admin().cluster().listTasks(generateListTasksRequest(request), new RestResponseListener<ListTasksResponse>(channel) {
+                client.admin().cluster().listTasks(listTasksRequest, new RestResponseListener<ListTasksResponse>(channel) {
             @Override
             public RestResponse buildResponse(ListTasksResponse listTasksResponse) throws Exception {
                 return RestTable.buildResponse(buildTable(request, listTasksResponse), channel);

server/src/test/java/org/elasticsearch/rest/action/cat/RestTasksActionTests.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.rest.action.cat;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.collect.MapBuilder;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.client.NoOpNodeClient;
+import org.elasticsearch.test.rest.FakeRestChannel;
+import org.elasticsearch.test.rest.FakeRestRequest;
+
+import static java.util.Collections.emptyList;
+import static org.hamcrest.Matchers.is;
+
+public class RestTasksActionTests extends ESTestCase {
+
+    public void testConsumesParameters() throws Exception {
+        RestTasksAction action = new RestTasksAction(() -> DiscoveryNodes.EMPTY_NODES);
+        FakeRestRequest fakeRestRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withParams(MapBuilder.<String, String>newMapBuilder()
+                .put("parent_task_id", "the node:3")
+                .put("nodes", "node1,node2")
+                .put("actions", "*")
+                .map()
+            ).build();
+        FakeRestChannel fakeRestChannel = new FakeRestChannel(fakeRestRequest, false, 1);
+        try (NoOpNodeClient nodeClient = buildNodeClient()) {
+            action.handleRequest(fakeRestRequest, fakeRestChannel, nodeClient);
+        }
+
+        assertThat(fakeRestChannel.errors().get(), is(0));
+        assertThat(fakeRestChannel.responses().get(), is(1));
+    }
+
+    private NoOpNodeClient buildNodeClient() {
+        return new NoOpNodeClient(getTestName()) {
+            @Override
+            @SuppressWarnings("unchecked")
+            public <Request extends ActionRequest, Response extends ActionResponse>
+            void doExecute(ActionType<Response> action, Request request, ActionListener<Response> listener) {
+                listener.onResponse((Response) new ListTasksResponse(emptyList(), emptyList(), emptyList()));
+            }
+        };
+    }
+}