Replace Streamable w/ Writeable in BaseTasksResponse and subclasses (elastic#36176)

This commit replaces usages of Streamable with Writeable for the
BaseTasksResponse / TransportTasksAction classes and subclasses of
these classes.

Note that where possible response fields were made final.

Relates to elastic#34389

Author: martijnvg

modules/reindex/src/main/java/org/elasticsearch/index/reindex/RethrottleAction.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
+import org.elasticsearch.common.io.stream.Writeable;
 
 public class RethrottleAction extends Action<ListTasksResponse> {
     public static final RethrottleAction INSTANCE = new RethrottleAction();
@@ -32,6 +33,11 @@ private RethrottleAction() {
 
     @Override
     public ListTasksResponse newResponse() {
-        return new ListTasksResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<ListTasksResponse> getResponseReader() {
+        return ListTasksResponse::new;
     }
 }

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportRethrottleAction.java

@@ -29,13 +29,11 @@
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.List;
 
 public class TransportRethrottleAction extends TransportTasksAction<BulkByScrollTask, RethrottleRequest, ListTasksResponse, TaskInfo> {
@@ -45,7 +43,7 @@ public class TransportRethrottleAction extends TransportTasksAction<BulkByScroll
     public TransportRethrottleAction(ClusterService clusterService, TransportService transportService,
                                      ActionFilters actionFilters, Client client) {
         super(RethrottleAction.NAME, clusterService, transportService, actionFilters,
-            RethrottleRequest::new, ListTasksResponse::new, ThreadPool.Names.MANAGEMENT);
+            RethrottleRequest::new, ListTasksResponse::new, TaskInfo::new, ThreadPool.Names.MANAGEMENT);
         this.client = client;
     }
 
@@ -101,11 +99,6 @@ private static void rethrottleChildTask(Logger logger, String localNodeId, BulkB
         listener.onResponse(task.taskInfo(localNodeId, true));
     }
 
-    @Override
-    protected TaskInfo readTaskResponse(StreamInput in) throws IOException {
-        return new TaskInfo(in);
-    }
-
     @Override
     protected ListTasksResponse newResponse(RethrottleRequest request, List<TaskInfo> tasks,
             List<TaskOperationFailure> taskOperationFailures, List<FailedNodeException> failedNodeExceptions) {

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksAction.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.cluster.node.tasks.cancel;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 /**
  * Action for cancelling running tasks
@@ -35,6 +36,11 @@ private CancelTasksAction() {
 
     @Override
     public CancelTasksResponse newResponse() {
-        return new CancelTasksResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<CancelTasksResponse> getResponseReader() {
+        return CancelTasksResponse::new;
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksResponse.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.TaskOperationFailure;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -40,7 +41,8 @@ public class CancelTasksResponse extends ListTasksResponse {
     private static final ConstructingObjectParser<CancelTasksResponse, Void> PARSER =
         setupParser("cancel_tasks_response", CancelTasksResponse::new);
 
-    public CancelTasksResponse() {
+    public CancelTasksResponse(StreamInput in) throws IOException {
+        super(in);
     }
 
     public CancelTasksResponse(List<TaskInfo> tasks, List<TaskOperationFailure> taskFailures, List<? extends ElasticsearchException>

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java

@@ -64,7 +64,7 @@ public class TransportCancelTasksAction extends TransportTasksAction<Cancellable
     @Inject
     public TransportCancelTasksAction(ClusterService clusterService, TransportService transportService, ActionFilters actionFilters) {
         super(CancelTasksAction.NAME, clusterService, transportService, actionFilters,
-            CancelTasksRequest::new, CancelTasksResponse::new, ThreadPool.Names.MANAGEMENT);
+            CancelTasksRequest::new, CancelTasksResponse::new, TaskInfo::new, ThreadPool.Names.MANAGEMENT);
         transportService.registerRequestHandler(BAN_PARENT_ACTION_NAME, ThreadPool.Names.SAME, BanParentTaskRequest::new,
             new BanParentRequestHandler());
     }
@@ -75,11 +75,6 @@ protected CancelTasksResponse newResponse(CancelTasksRequest request, List<TaskI
         return new CancelTasksResponse(tasks, taskOperationFailures, failedNodeExceptions);
     }
 
-    @Override
-    protected TaskInfo readTaskResponse(StreamInput in) throws IOException {
-        return new TaskInfo(in);
-    }
-
     protected void processTasks(CancelTasksRequest request, Consumer<CancellableTask> operation) {
         if (request.getTaskId().isSet()) {
             // we are only checking one task, we can optimize it

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksAction.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.cluster.node.tasks.list;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 /**
  * Action for retrieving a list of currently running tasks
@@ -35,6 +36,11 @@ private ListTasksAction() {
 
     @Override
     public ListTasksResponse newResponse() {
-        return new ListTasksResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<ListTasksResponse> getResponseReader() {
+        return ListTasksResponse::new;
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksResponse.java

@@ -52,22 +52,28 @@
 public class ListTasksResponse extends BaseTasksResponse implements ToXContentObject {
     private static final String TASKS = "tasks";
 
-    private List<TaskInfo> tasks;
+    private final List<TaskInfo> tasks;
 
     private Map<String, List<TaskInfo>> perNodeTasks;
 
     private List<TaskGroup> groups;
 
-    public ListTasksResponse() {
-        this(null, null, null);
-    }
-
     public ListTasksResponse(List<TaskInfo> tasks, List<TaskOperationFailure> taskFailures,
             List<? extends ElasticsearchException> nodeFailures) {
         super(taskFailures, nodeFailures);
         this.tasks = tasks == null ? Collections.emptyList() : Collections.unmodifiableList(new ArrayList<>(tasks));
     }
 
+    public ListTasksResponse(StreamInput in) throws IOException {
+        super(in);
+        tasks = Collections.unmodifiableList(in.readList(TaskInfo::new));
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeList(tasks);
+    }
 
     protected static <T> ConstructingObjectParser<T, Void> setupParser(String name,
                                                                        TriFunction<
@@ -96,18 +102,6 @@ protected static <T> ConstructingObjectParser<T, Void> setupParser(String name,
     private static final ConstructingObjectParser<ListTasksResponse, Void> PARSER =
         setupParser("list_tasks_response", ListTasksResponse::new);
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        tasks = Collections.unmodifiableList(in.readList(TaskInfo::new));
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeList(tasks);
-    }
-
     /**
      * Returns the list of tasks by node
      */

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java

@@ -26,14 +26,12 @@
 import org.elasticsearch.action.support.tasks.TransportTasksAction;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.List;
 import java.util.function.Consumer;
 
@@ -52,7 +50,7 @@ public static long waitForCompletionTimeout(TimeValue timeout) {
     @Inject
     public TransportListTasksAction(ClusterService clusterService, TransportService transportService, ActionFilters actionFilters) {
         super(ListTasksAction.NAME, clusterService, transportService, actionFilters,
-            ListTasksRequest::new, ListTasksResponse::new, ThreadPool.Names.MANAGEMENT);
+            ListTasksRequest::new, ListTasksResponse::new, TaskInfo::new, ThreadPool.Names.MANAGEMENT);
     }
 
     @Override
@@ -61,11 +59,6 @@ protected ListTasksResponse newResponse(ListTasksRequest request, List<TaskInfo>
         return new ListTasksResponse(tasks, taskOperationFailures, failedNodeExceptions);
     }
 
-    @Override
-    protected TaskInfo readTaskResponse(StreamInput in) throws IOException {
-        return new TaskInfo(in);
-    }
-
     @Override
     protected void taskOperation(ListTasksRequest request, Task task, ActionListener<TaskInfo> listener) {
         listener.onResponse(task.taskInfo(clusterService.localNode().getId(), request.getDetailed()));

server/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksResponse.java

@@ -55,34 +55,8 @@ public BaseTasksResponse(List<TaskOperationFailure> taskFailures, List<? extends
         this.nodeFailures = nodeFailures == null ? Collections.emptyList() : Collections.unmodifiableList(new ArrayList<>(nodeFailures));
     }
 
-    /**
-     * The list of task failures exception.
-     */
-    public List<TaskOperationFailure> getTaskFailures() {
-        return taskFailures;
-    }
-
-    /**
-     * The list of node failures exception.
-     */
-    public List<ElasticsearchException> getNodeFailures() {
-        return nodeFailures;
-    }
-
-    /**
-     * Rethrow task failures if there are any.
-     */
-    public void rethrowFailures(String operationName) {
-        rethrowAndSuppress(Stream.concat(
-                    getNodeFailures().stream(),
-                    getTaskFailures().stream().map(f -> new ElasticsearchException(
-                            "{} of [{}] failed", f.getCause(), operationName, new TaskId(f.getNodeId(), f.getTaskId()))))
-                .collect(toList()));
-    }
-
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+    public BaseTasksResponse(StreamInput in) throws IOException {
+        super(in);
         int size = in.readVInt();
         List<TaskOperationFailure> taskFailures = new ArrayList<>(size);
         for (int i = 0; i < size; i++) {
@@ -110,6 +84,31 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    /**
+     * The list of task failures exception.
+     */
+    public List<TaskOperationFailure> getTaskFailures() {
+        return taskFailures;
+    }
+
+    /**
+     * The list of node failures exception.
+     */
+    public List<ElasticsearchException> getNodeFailures() {
+        return nodeFailures;
+    }
+
+    /**
+     * Rethrow task failures if there are any.
+     */
+    public void rethrowFailures(String operationName) {
+        rethrowAndSuppress(Stream.concat(
+                    getNodeFailures().stream(),
+                    getTaskFailures().stream().map(f -> new ElasticsearchException(
+                            "{} of [{}] failed", f.getCause(), operationName, new TaskId(f.getNodeId(), f.getTaskId()))))
+                .collect(toList()));
+    }
+
     protected void toXContentCommon(XContentBuilder builder, ToXContent.Params params) throws IOException {
         if (getTaskFailures() != null && getTaskFailures().size() > 0) {
             builder.startArray(TASK_FAILURES);

server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java

@@ -55,7 +55,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.function.Consumer;
-import java.util.function.Supplier;
 
 import static java.util.Collections.emptyList;
 
@@ -71,20 +70,23 @@ public abstract class TransportTasksAction<
 
     protected final ClusterService clusterService;
     protected final TransportService transportService;
-    protected final Writeable.Reader<TasksRequest> requestSupplier;
-    protected final Supplier<TasksResponse> responseSupplier;
+    protected final Writeable.Reader<TasksRequest> requestReader;
+    protected final Writeable.Reader<TasksResponse> responsesReader;
+    protected final Writeable.Reader<TaskResponse> responseReader;
 
     protected final String transportNodeAction;
 
     protected TransportTasksAction(String actionName, ClusterService clusterService, TransportService transportService,
-                                   ActionFilters actionFilters, Writeable.Reader<TasksRequest> requestSupplier,
-                                   Supplier<TasksResponse> responseSupplier, String nodeExecutor) {
-        super(actionName, transportService, actionFilters, requestSupplier);
+                                   ActionFilters actionFilters, Writeable.Reader<TasksRequest> requestReader,
+                                   Writeable.Reader<TasksResponse> responsesReader, Writeable.Reader<TaskResponse> responseReader,
+                                   String nodeExecutor) {
+        super(actionName, transportService, actionFilters, requestReader);
         this.clusterService = clusterService;
         this.transportService = transportService;
         this.transportNodeAction = actionName + "[n]";
-        this.requestSupplier = requestSupplier;
-        this.responseSupplier = responseSupplier;
+        this.requestReader = requestReader;
+        this.responsesReader = responsesReader;
+        this.responseReader = responseReader;
 
         transportService.registerRequestHandler(transportNodeAction, nodeExecutor, NodeTaskRequest::new, new NodeTransportHandler());
     }
@@ -205,8 +207,6 @@ protected TasksResponse newResponse(TasksRequest request, AtomicReferenceArray r
         return newResponse(request, tasks, taskOperationFailures, failedNodeExceptions);
     }
 
-    protected abstract TaskResponse readTaskResponse(StreamInput in) throws IOException;
-
     /**
      * Perform the required operation on the task. It is OK start an asynchronous operation or to throw an exception but not both.
      */
@@ -364,7 +364,7 @@ private class NodeTaskRequest extends TransportRequest {
 
         protected NodeTaskRequest(StreamInput in) throws IOException {
             super(in);
-            this.tasksRequest = requestSupplier.read(in);
+            this.tasksRequest = requestReader.read(in);
         }
 
         @Override
@@ -411,7 +411,7 @@ public void readFrom(StreamInput in) throws IOException {
             int resultsSize = in.readVInt();
             results = new ArrayList<>(resultsSize);
             for (; resultsSize > 0; resultsSize--) {
-                final TaskResponse result = in.readBoolean() ? readTaskResponse(in) : null;
+                final TaskResponse result = in.readBoolean() ? responseReader.read(in) : null;
                 results.add(result);
             }
             if (in.readBoolean()) {