Dry up TransportMasterNodeAction Usage

1. It is confusing and unnecessary to handle response deserialization via inheritance
and request deserialization via composition. Consistently using composition saves
hundreds of lines of code and as a matter of fact also removes some indirection in
transport master node action since we pass a reader down to the response handling anyway.
2. Defining `executor` via inheritance but then assuming the return of the method is a constant
is confusing and again not in line with how we handle the `executor` definition for other transport
actions so this was simplified away as well.

Somewhat relates to the dry-up in elastic#63335

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/admin/cluster/allocation/TransportClusterAllocationExplainAction.java

@@ -37,17 +37,14 @@
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation.DebugMode;
 import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
-import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDeciders;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.snapshots.SnapshotsInfoService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.List;
 
 /**
@@ -62,35 +59,23 @@ public class TransportClusterAllocationExplainAction
     private final ClusterInfoService clusterInfoService;
     private final SnapshotsInfoService snapshotsInfoService;
     private final AllocationDeciders allocationDeciders;
-    private final ShardsAllocator shardAllocator;
     private final AllocationService allocationService;
 
     @Inject
     public TransportClusterAllocationExplainAction(TransportService transportService, ClusterService clusterService,
                                                    ThreadPool threadPool, ActionFilters actionFilters,
                                                    IndexNameExpressionResolver indexNameExpressionResolver,
                                                    ClusterInfoService clusterInfoService, SnapshotsInfoService snapshotsInfoService,
-                                                   AllocationDeciders allocationDeciders,
-                                                   ShardsAllocator shardAllocator, AllocationService allocationService) {
+                                                   AllocationDeciders allocationDeciders, AllocationService allocationService) {
         super(ClusterAllocationExplainAction.NAME, transportService, clusterService, threadPool, actionFilters,
-            ClusterAllocationExplainRequest::new, indexNameExpressionResolver);
+            ClusterAllocationExplainRequest::new, indexNameExpressionResolver, ClusterAllocationExplainResponse::new,
+                ThreadPool.Names.MANAGEMENT);
         this.clusterInfoService = clusterInfoService;
         this.snapshotsInfoService = snapshotsInfoService;
         this.allocationDeciders = allocationDeciders;
-        this.shardAllocator = shardAllocator;
         this.allocationService = allocationService;
     }
 
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.MANAGEMENT;
-    }
-
-    @Override
-    protected ClusterAllocationExplainResponse read(StreamInput in) throws IOException {
-        return new ClusterAllocationExplainResponse(in);
-    }
-
     @Override
     protected ClusterBlockException checkBlock(ClusterAllocationExplainRequest request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingConfigExclusionsAction.java

@@ -38,18 +38,15 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.Set;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -69,7 +66,8 @@ public TransportAddVotingConfigExclusionsAction(Settings settings, ClusterSettin
                                                     ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
                                                     IndexNameExpressionResolver indexNameExpressionResolver) {
         super(AddVotingConfigExclusionsAction.NAME, transportService, clusterService, threadPool, actionFilters,
-            AddVotingConfigExclusionsRequest::new, indexNameExpressionResolver);
+            AddVotingConfigExclusionsRequest::new, indexNameExpressionResolver, AddVotingConfigExclusionsResponse::new,
+                ThreadPool.Names.SAME);
 
         maxVotingConfigExclusions = MAXIMUM_VOTING_CONFIG_EXCLUSIONS_SETTING.get(settings);
         clusterSettings.addSettingsUpdateConsumer(MAXIMUM_VOTING_CONFIG_EXCLUSIONS_SETTING, this::setMaxVotingConfigExclusions);
@@ -79,16 +77,6 @@ private void setMaxVotingConfigExclusions(int maxVotingConfigExclusions) {
         this.maxVotingConfigExclusions = maxVotingConfigExclusions;
     }
 
-    @Override
-    protected String executor() {
-        return Names.SAME;
-    }
-
-    @Override
-    protected AddVotingConfigExclusionsResponse read(StreamInput in) throws IOException {
-        return new AddVotingConfigExclusionsResponse(in);
-    }
-
     @Override
     protected void masterOperation(Task task, AddVotingConfigExclusionsRequest request, ClusterState state,
                                    ActionListener<AddVotingConfigExclusionsResponse> listener) throws Exception {

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingConfigExclusionsAction.java

@@ -38,14 +38,11 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
 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.threadpool.ThreadPool;
-import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.function.Predicate;
 
 public class TransportClearVotingConfigExclusionsAction
@@ -58,17 +55,8 @@ public TransportClearVotingConfigExclusionsAction(TransportService transportServ
                                                       ThreadPool threadPool, ActionFilters actionFilters,
                                                       IndexNameExpressionResolver indexNameExpressionResolver) {
         super(ClearVotingConfigExclusionsAction.NAME, transportService, clusterService, threadPool, actionFilters,
-            ClearVotingConfigExclusionsRequest::new, indexNameExpressionResolver);
-    }
-
-    @Override
-    protected String executor() {
-        return Names.SAME;
-    }
-
-    @Override
-    protected ClearVotingConfigExclusionsResponse read(StreamInput in) throws IOException {
-        return new ClearVotingConfigExclusionsResponse(in);
+            ClearVotingConfigExclusionsRequest::new, indexNameExpressionResolver, ClearVotingConfigExclusionsResponse::new,
+                ThreadPool.Names.SAME);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.index.IndexNotFoundException;
@@ -50,7 +49,6 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 
@@ -65,21 +63,10 @@ public TransportClusterHealthAction(TransportService transportService, ClusterSe
                                         ThreadPool threadPool, ActionFilters actionFilters,
                                         IndexNameExpressionResolver indexNameExpressionResolver, AllocationService allocationService) {
         super(ClusterHealthAction.NAME, false, transportService, clusterService, threadPool, actionFilters,
-            ClusterHealthRequest::new, indexNameExpressionResolver);
+            ClusterHealthRequest::new, indexNameExpressionResolver,ClusterHealthResponse::new, ThreadPool.Names.SAME);
         this.allocationService = allocationService;
     }
 
-    @Override
-    protected String executor() {
-        // this should be executing quickly no need to fork off
-        return ThreadPool.Names.SAME;
-    }
-
-    @Override
-    protected ClusterHealthResponse read(StreamInput in) throws IOException {
-        return new ClusterHealthResponse(in);
-    }
-
     @Override
     protected ClusterBlockException checkBlock(ClusterHealthRequest request, ClusterState state) {
         // we want users to be able to call this even when there are global blocks, just to check the health (are there blocks?)

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java

@@ -38,7 +38,6 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.Repository;
 import org.elasticsearch.repositories.RepositoryCleanupResult;
@@ -49,7 +48,6 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.List;
 
 /**
@@ -78,18 +76,13 @@ public final class TransportCleanupRepositoryAction extends TransportMasterNodeA
 
     private final SnapshotsService snapshotsService;
 
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
     @Inject
     public TransportCleanupRepositoryAction(TransportService transportService, ClusterService clusterService,
                                             RepositoriesService repositoriesService, SnapshotsService snapshotsService,
                                             ThreadPool threadPool, ActionFilters actionFilters,
                                             IndexNameExpressionResolver indexNameExpressionResolver) {
         super(CleanupRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
-            CleanupRepositoryRequest::new, indexNameExpressionResolver);
+            CleanupRepositoryRequest::new, indexNameExpressionResolver, CleanupRepositoryResponse::new, ThreadPool.Names.SAME);
         this.repositoriesService = repositoriesService;
         this.snapshotsService = snapshotsService;
         // We add a state applier that will remove any dangling repository cleanup actions on master failover.
@@ -136,11 +129,6 @@ private static ClusterState removeInProgressCleanup(final ClusterState currentSt
             : currentState;
     }
 
-    @Override
-    protected CleanupRepositoryResponse read(StreamInput in) throws IOException {
-        return new CleanupRepositoryResponse(in);
-    }
-
     @Override
     protected void masterOperation(Task task, CleanupRepositoryRequest request, ClusterState state,
                                    ActionListener<CleanupRepositoryResponse> listener) {

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/delete/TransportDeleteRepositoryAction.java

@@ -46,15 +46,10 @@ public TransportDeleteRepositoryAction(TransportService transportService, Cluste
                                            RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
                                            IndexNameExpressionResolver indexNameExpressionResolver) {
         super(DeleteRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
-              DeleteRepositoryRequest::new, indexNameExpressionResolver);
+              DeleteRepositoryRequest::new, indexNameExpressionResolver, ThreadPool.Names.SAME);
         this.repositoriesService = repositoriesService;
     }
 
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
     @Override
     protected ClusterBlockException checkBlock(DeleteRepositoryRequest request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/get/TransportGetRepositoriesAction.java

@@ -31,14 +31,12 @@
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.repositories.RepositoryMissingException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashSet;
@@ -55,17 +53,7 @@ public TransportGetRepositoriesAction(TransportService transportService, Cluster
                                           ThreadPool threadPool, ActionFilters actionFilters,
                                           IndexNameExpressionResolver indexNameExpressionResolver) {
         super(GetRepositoriesAction.NAME, transportService, clusterService, threadPool, actionFilters,
-              GetRepositoriesRequest::new, indexNameExpressionResolver);
-    }
-
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
-    @Override
-    protected GetRepositoriesResponse read(StreamInput in) throws IOException {
-        return new GetRepositoriesResponse(in);
+              GetRepositoriesRequest::new, indexNameExpressionResolver, GetRepositoriesResponse::new, ThreadPool.Names.SAME);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/put/TransportPutRepositoryAction.java

@@ -46,15 +46,10 @@ public TransportPutRepositoryAction(TransportService transportService, ClusterSe
                                         RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
                                         IndexNameExpressionResolver indexNameExpressionResolver) {
         super(PutRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
-              PutRepositoryRequest::new, indexNameExpressionResolver);
+              PutRepositoryRequest::new, indexNameExpressionResolver, ThreadPool.Names.SAME);
         this.repositoriesService = repositoriesService;
     }
 
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
     @Override
     protected ClusterBlockException checkBlock(PutRepositoryRequest request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/verify/TransportVerifyRepositoryAction.java

@@ -29,14 +29,11 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
-
 /**
  * Transport action for verifying repository operation
  */
@@ -51,20 +48,10 @@ public TransportVerifyRepositoryAction(TransportService transportService, Cluste
                                            RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
                                            IndexNameExpressionResolver indexNameExpressionResolver) {
         super(VerifyRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
-              VerifyRepositoryRequest::new, indexNameExpressionResolver);
+              VerifyRepositoryRequest::new, indexNameExpressionResolver, VerifyRepositoryResponse::new, ThreadPool.Names.SAME);
         this.repositoriesService = repositoriesService;
     }
 
-    @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
-    @Override
-    protected VerifyRepositoryResponse read(StreamInput in) throws IOException {
-        return new VerifyRepositoryResponse(in);
-    }
-
     @Override
     protected ClusterBlockException checkBlock(VerifyRepositoryRequest request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);

server/src/main/java/org/elasticsearch/action/admin/cluster/reroute/TransportClusterRerouteAction.java

@@ -47,12 +47,10 @@
 import org.elasticsearch.common.collect.ImmutableOpenIntMap;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -69,26 +67,15 @@ public TransportClusterRerouteAction(TransportService transportService, ClusterS
                                          ThreadPool threadPool, AllocationService allocationService, ActionFilters actionFilters,
                                          IndexNameExpressionResolver indexNameExpressionResolver) {
         super(ClusterRerouteAction.NAME, transportService, clusterService, threadPool, actionFilters,
-              ClusterRerouteRequest::new, indexNameExpressionResolver);
+              ClusterRerouteRequest::new, indexNameExpressionResolver, ClusterRerouteResponse::new, ThreadPool.Names.SAME);
         this.allocationService = allocationService;
     }
 
-    @Override
-    protected String executor() {
-        // we go async right away
-        return ThreadPool.Names.SAME;
-    }
-
     @Override
     protected ClusterBlockException checkBlock(ClusterRerouteRequest request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
     }
 
-    @Override
-    protected ClusterRerouteResponse read(StreamInput in) throws IOException {
-        return new ClusterRerouteResponse(in);
-    }
-
     @Override
     protected void masterOperation(Task task, final ClusterRerouteRequest request, final ClusterState state,
                                    final ActionListener<ClusterRerouteResponse> listener) {