Replace Streamable w/ Writeable in BaseTasksRequest and subclasses

This commit replaces usages of Streamable with Writeable for the BaseTasksRequest / TransportTasksAction classes and subclasses of these classes. Relates to elastic#34389
martijnvg · Nov 23, 2018 · 5fe26b7 · 5fe26b7
1 parent d01436d
commit 5fe26b7
Show file tree

Hide file tree

Showing 28 changed files with 189 additions and 196 deletions.
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RethrottleRequest.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RethrottleRequest.java
@@ -39,6 +39,14 @@ public class RethrottleRequest extends BaseTasksRequest<RethrottleRequest> {
      */
     private Float requestsPerSecond;
 
+    public RethrottleRequest() {
+    }
+
+    public RethrottleRequest(StreamInput in) throws IOException {
+        super(in);
+        this.requestsPerSecond = in.readFloat();
+    }
+
     /**
      * The throttle to apply to all matching requests in sub-requests per second. 0 means set no throttle and that is the default.
      */
@@ -80,12 +88,6 @@ public ActionRequestValidationException validate() {
         return validationException;
     }
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        requestsPerSecond = in.readFloat();
-    }
-
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);

diff --git a/...ain/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequest.java b/...ain/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequest.java
@@ -36,10 +36,11 @@ public class CancelTasksRequest extends BaseTasksRequest<CancelTasksRequest> {
 
     private String reason = DEFAULT_REASON;
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        reason = in.readString();
+    public CancelTasksRequest() {}
+
+    public CancelTasksRequest(StreamInput in) throws IOException {
+        super(in);
+        this.reason = in.readString();
     }
 
     @Override

diff --git a/.../org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java b/.../org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java
@@ -65,8 +65,8 @@ public class TransportCancelTasksAction extends TransportTasksAction<Cancellable
     public TransportCancelTasksAction(ClusterService clusterService, TransportService transportService, ActionFilters actionFilters) {
         super(CancelTasksAction.NAME, clusterService, transportService, actionFilters,
             CancelTasksRequest::new, CancelTasksResponse::new, ThreadPool.Names.MANAGEMENT);
-        transportService.registerRequestHandler(BAN_PARENT_ACTION_NAME, BanParentTaskRequest::new, ThreadPool.Names.SAME, new
-            BanParentRequestHandler());
+        transportService.registerRequestHandler(BAN_PARENT_ACTION_NAME, ThreadPool.Names.SAME, BanParentTaskRequest::new,
+            new BanParentRequestHandler());
     }
 
     @Override
@@ -233,11 +233,9 @@ public void finish() {
 
     private static class BanParentTaskRequest extends TransportRequest {
 
-        private TaskId parentTaskId;
-
-        private boolean ban;
-
-        private String reason;
+        private final TaskId parentTaskId;
+        private final boolean ban;
+        private final String reason;
 
         static BanParentTaskRequest createSetBanParentTaskRequest(TaskId parentTaskId, String reason) {
             return new BanParentTaskRequest(parentTaskId, reason);
@@ -256,19 +254,14 @@ private BanParentTaskRequest(TaskId parentTaskId, String reason) {
         private BanParentTaskRequest(TaskId parentTaskId) {
             this.parentTaskId = parentTaskId;
             this.ban = false;
+            this.reason = null;
         }
 
-        BanParentTaskRequest() {
-        }
-
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+        private BanParentTaskRequest(StreamInput in) throws IOException {
+            super(in);
             parentTaskId = TaskId.readFromStream(in);
             ban = in.readBoolean();
-            if (ban) {
-                reason = in.readString();
-            }
+            reason = ban ? in.readString() : null;
         }
 
         @Override

diff --git a/...rc/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksRequest.java b/...rc/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksRequest.java
@@ -33,6 +33,15 @@ public class ListTasksRequest extends BaseTasksRequest<ListTasksRequest> {
     private boolean detailed = false;
     private boolean waitForCompletion = false;
 
+    public ListTasksRequest() {
+    }
+
+    public ListTasksRequest(StreamInput in) throws IOException {
+        super(in);
+        detailed = in.readBoolean();
+        waitForCompletion = in.readBoolean();
+    }
+
     /**
      * Should the detailed task information be returned.
      */
@@ -63,13 +72,6 @@ public ListTasksRequest setWaitForCompletion(boolean waitForCompletion) {
         return this;
     }
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        detailed = in.readBoolean();
-        waitForCompletion = in.readBoolean();
-    }
-
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);

diff --git a/server/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java b/server/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java
@@ -52,9 +52,20 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
 
     private TaskId taskId = TaskId.EMPTY_TASK_ID;
 
+    // NOTE: This constructor is only needed, because the setters in this class,
+    // otherwise it can be removed and above fields can be made final.
     public BaseTasksRequest() {
     }
 
+    protected BaseTasksRequest(StreamInput in) throws IOException {
+        super(in);
+        taskId = TaskId.readFromStream(in);
+        parentTaskId = TaskId.readFromStream(in);
+        nodes = in.readStringArray();
+        actions = in.readStringArray();
+        timeout = in.readOptionalTimeValue();
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = null;
@@ -137,16 +148,6 @@ public final Request setTimeout(String timeout) {
         return (Request) this;
     }
 
-    @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        taskId = TaskId.readFromStream(in);
-        parentTaskId = TaskId.readFromStream(in);
-        nodes = in.readStringArray();
-        actions = in.readStringArray();
-        timeout = in.readOptionalTimeValue();
-    }
-
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);

diff --git a/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java b/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java
@@ -71,13 +71,13 @@ public abstract class TransportTasksAction<
 
     protected final ClusterService clusterService;
     protected final TransportService transportService;
-    protected final Supplier<TasksRequest> requestSupplier;
+    protected final Writeable.Reader<TasksRequest> requestSupplier;
     protected final Supplier<TasksResponse> responseSupplier;
 
     protected final String transportNodeAction;
 
     protected TransportTasksAction(String actionName, ClusterService clusterService, TransportService transportService,
-                                   ActionFilters actionFilters, Supplier<TasksRequest> requestSupplier,
+                                   ActionFilters actionFilters, Writeable.Reader<TasksRequest> requestSupplier,
                                    Supplier<TasksResponse> responseSupplier, String nodeExecutor) {
         super(actionName, transportService, actionFilters, requestSupplier);
         this.clusterService = clusterService;
@@ -86,7 +86,7 @@ protected TransportTasksAction(String actionName, ClusterService clusterService,
         this.requestSupplier = requestSupplier;
         this.responseSupplier = responseSupplier;
 
-        transportService.registerRequestHandler(transportNodeAction, NodeTaskRequest::new, nodeExecutor, new NodeTransportHandler());
+        transportService.registerRequestHandler(transportNodeAction, nodeExecutor, NodeTaskRequest::new, new NodeTransportHandler());
     }
 
     @Override
@@ -362,22 +362,16 @@ public void onFailure(Exception e) {
     private class NodeTaskRequest extends TransportRequest {
         private TasksRequest tasksRequest;
 
-        protected NodeTaskRequest() {
-            super();
+        protected NodeTaskRequest(StreamInput in) throws IOException {
+            super(in);
+            this.tasksRequest = requestSupplier.read(in);
         }
 
         protected NodeTaskRequest(TasksRequest tasksRequest) {
             super();
             this.tasksRequest = tasksRequest;
         }
 
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            tasksRequest = requestSupplier.get();
-            tasksRequest.readFrom(in);
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);

diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java
@@ -373,6 +373,13 @@ public void writeTo(StreamOutput out) throws IOException {
 
 
     public static class UnblockTestTasksRequest extends BaseTasksRequest<UnblockTestTasksRequest> {
+
+        UnblockTestTasksRequest() {}
+
+        UnblockTestTasksRequest(StreamInput in) throws IOException {
+            super(in);
+        }
+
         @Override
         public boolean match(Task task) {
             return task instanceof TestTask && super.match(task);

diff --git a/...est/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java b/...est/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java
@@ -191,6 +191,11 @@ public String getStatus() {
 
     static class TestTasksRequest extends BaseTasksRequest<TestTasksRequest> {
 
+        TestTasksRequest(StreamInput in) throws IOException {
+            super(in);
+        }
+
+        TestTasksRequest() {}
     }
 
     static class TestTasksResponse extends BaseTasksResponse {

diff --git a/server/src/test/java/org/elasticsearch/persistent/TestPersistentTasksPlugin.java b/server/src/test/java/org/elasticsearch/persistent/TestPersistentTasksPlugin.java
@@ -447,9 +447,8 @@ public static class TestTasksRequest extends BaseTasksRequest<TestTasksRequest>
         public TestTasksRequest() {
         }
 
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+        public TestTasksRequest(StreamInput in) throws IOException {
+            super(in);
             operation = in.readOptionalString();
         }
 

diff --git a/...ugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportFollowStatsAction.java b/...ugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportFollowStatsAction.java
@@ -84,7 +84,6 @@ protected FollowStatsAction.StatsResponse readTaskResponse(final StreamInput in)
     protected void processTasks(final FollowStatsAction.StatsRequest request, final Consumer<ShardFollowNodeTask> operation) {
         final ClusterState state = clusterService.state();
         final PersistentTasksCustomMetaData persistentTasksMetaData = state.metaData().custom(PersistentTasksCustomMetaData.TYPE);
-
         if (persistentTasksMetaData == null) {
             return;
         }

diff --git a/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/action/FollowStatsAction.java b/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/action/FollowStatsAction.java
@@ -129,6 +129,13 @@ public static class StatsRequest extends BaseTasksRequest<StatsRequest> implemen
 
         private String[] indices;
 
+        public StatsRequest() {}
+
+        public StatsRequest(StreamInput in) throws IOException {
+            super(in);
+            indices = in.readOptionalStringArray();
+        }
+
         @Override
         public String[] indices() {
             return indices;
@@ -161,12 +168,6 @@ public ActionRequestValidationException validate() {
             return null;
         }
 
-        @Override
-        public void readFrom(final StreamInput in) throws IOException {
-            super.readFrom(in);
-            indices = in.readOptionalStringArray();
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/CloseJobAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/CloseJobAction.java
@@ -79,6 +79,18 @@ public Request() {
             openJobIds = new String[] {};
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            jobId = in.readString();
+            timeout = in.readTimeValue();
+            force = in.readBoolean();
+            openJobIds = in.readStringArray();
+            local = in.readBoolean();
+            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
+                allowNoJobs = in.readBoolean();
+            }
+        }
+
         public Request(String jobId) {
             this();
             this.jobId = jobId;
@@ -128,19 +140,6 @@ public void setOpenJobIds(String [] openJobIds) {
             this.openJobIds = openJobIds;
         }
 
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            jobId = in.readString();
-            timeout = in.readTimeValue();
-            force = in.readBoolean();
-            openJobIds = in.readStringArray();
-            local = in.readBoolean();
-            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
-                allowNoJobs = in.readBoolean();
-            }
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/FlushJobAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/FlushJobAction.java
@@ -75,6 +75,15 @@ public static Request parseRequest(String jobId, XContentParser parser) {
         public Request() {
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            calcInterim = in.readBoolean();
+            start = in.readOptionalString();
+            end = in.readOptionalString();
+            advanceTime = in.readOptionalString();
+            skipTime = in.readOptionalString();
+        }
+
         public Request(String jobId) {
             super(jobId);
         }
@@ -119,16 +128,6 @@ public void setSkipTime(String skipTime) {
             this.skipTime = skipTime;
         }
 
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            calcInterim = in.readBoolean();
-            start = in.readOptionalString();
-            end = in.readOptionalString();
-            advanceTime = in.readOptionalString();
-            skipTime = in.readOptionalString();
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);

diff --git a/...k/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/ForecastJobAction.java b/...k/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/ForecastJobAction.java
@@ -68,6 +68,12 @@ public static Request parseRequest(String jobId, XContentParser parser) {
         public Request() {
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            this.duration = in.readOptionalTimeValue();
+            this.expiresIn = in.readOptionalTimeValue();
+        }
+
         public Request(String jobId) {
             super(jobId);
         }
@@ -108,13 +114,6 @@ public void setExpiresIn(TimeValue expiresIn) {
             }
         }
 
-        @Override
-        public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            this.duration = in.readOptionalTimeValue();
-            this.expiresIn = in.readOptionalTimeValue();
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);