Dry up TransportMasterNodeAction Usage (#63524)

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 #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) {