Skip to content

Commit 9a690cd

Browse files
original-brownbearoriginal-brownbear
authored
Dry up TransportMasterNodeAction Usage (elastic#63524) (elastic#64146)
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
1 parent d552efb commit 9a690cd

File tree

147 files changed

+233
-1615
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

147 files changed

+233
-1615
lines changed

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

+3-18
Original file line numberDiff line numberDiff line change
@@ -37,16 +37,13 @@
3737
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
3838
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation.DebugMode;
3939
import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
40-
import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
4140
import org.elasticsearch.cluster.routing.allocation.decider.AllocationDeciders;
4241
import org.elasticsearch.cluster.service.ClusterService;
4342
import org.elasticsearch.common.inject.Inject;
44-
import org.elasticsearch.common.io.stream.StreamInput;
4543
import org.elasticsearch.snapshots.SnapshotsInfoService;
4644
import org.elasticsearch.threadpool.ThreadPool;
4745
import org.elasticsearch.transport.TransportService;
4846

49-
import java.io.IOException;
5047
import java.util.List;
5148

5249
/**
@@ -61,35 +58,23 @@ public class TransportClusterAllocationExplainAction
6158
private final ClusterInfoService clusterInfoService;
6259
private final SnapshotsInfoService snapshotsInfoService;
6360
private final AllocationDeciders allocationDeciders;
64-
private final ShardsAllocator shardAllocator;
6561
private final AllocationService allocationService;
6662

6763
@Inject
6864
public TransportClusterAllocationExplainAction(TransportService transportService, ClusterService clusterService,
6965
ThreadPool threadPool, ActionFilters actionFilters,
7066
IndexNameExpressionResolver indexNameExpressionResolver,
7167
ClusterInfoService clusterInfoService, SnapshotsInfoService snapshotsInfoService,
72-
AllocationDeciders allocationDeciders,
73-
ShardsAllocator shardAllocator, AllocationService allocationService) {
68+
AllocationDeciders allocationDeciders, AllocationService allocationService) {
7469
super(ClusterAllocationExplainAction.NAME, transportService, clusterService, threadPool, actionFilters,
75-
ClusterAllocationExplainRequest::new, indexNameExpressionResolver);
70+
ClusterAllocationExplainRequest::new, indexNameExpressionResolver, ClusterAllocationExplainResponse::new,
71+
ThreadPool.Names.MANAGEMENT);
7672
this.clusterInfoService = clusterInfoService;
7773
this.snapshotsInfoService = snapshotsInfoService;
7874
this.allocationDeciders = allocationDeciders;
79-
this.shardAllocator = shardAllocator;
8075
this.allocationService = allocationService;
8176
}
8277

83-
@Override
84-
protected String executor() {
85-
return ThreadPool.Names.MANAGEMENT;
86-
}
87-
88-
@Override
89-
protected ClusterAllocationExplainResponse read(StreamInput in) throws IOException {
90-
return new ClusterAllocationExplainResponse(in);
91-
}
92-
9378
@Override
9479
protected ClusterBlockException checkBlock(ClusterAllocationExplainRequest request, ClusterState state) {
9580
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);

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

+2-14
Original file line numberDiff line numberDiff line change
@@ -38,17 +38,14 @@
3838
import org.elasticsearch.cluster.service.ClusterService;
3939
import org.elasticsearch.common.Priority;
4040
import org.elasticsearch.common.inject.Inject;
41-
import org.elasticsearch.common.io.stream.StreamInput;
4241
import org.elasticsearch.common.settings.ClusterSettings;
4342
import org.elasticsearch.common.settings.Setting;
4443
import org.elasticsearch.common.settings.Setting.Property;
4544
import org.elasticsearch.common.settings.Settings;
4645
import org.elasticsearch.common.unit.TimeValue;
4746
import org.elasticsearch.threadpool.ThreadPool;
48-
import org.elasticsearch.threadpool.ThreadPool.Names;
4947
import org.elasticsearch.transport.TransportService;
5048

51-
import java.io.IOException;
5249
import java.util.Set;
5350
import java.util.function.Predicate;
5451
import java.util.stream.Collectors;
@@ -68,7 +65,8 @@ public TransportAddVotingConfigExclusionsAction(Settings settings, ClusterSettin
6865
ClusterService clusterService, ThreadPool threadPool, ActionFilters actionFilters,
6966
IndexNameExpressionResolver indexNameExpressionResolver) {
7067
super(AddVotingConfigExclusionsAction.NAME, transportService, clusterService, threadPool, actionFilters,
71-
AddVotingConfigExclusionsRequest::new, indexNameExpressionResolver);
68+
AddVotingConfigExclusionsRequest::new, indexNameExpressionResolver, AddVotingConfigExclusionsResponse::new,
69+
ThreadPool.Names.SAME);
7270

7371
maxVotingConfigExclusions = MAXIMUM_VOTING_CONFIG_EXCLUSIONS_SETTING.get(settings);
7472
clusterSettings.addSettingsUpdateConsumer(MAXIMUM_VOTING_CONFIG_EXCLUSIONS_SETTING, this::setMaxVotingConfigExclusions);
@@ -78,16 +76,6 @@ private void setMaxVotingConfigExclusions(int maxVotingConfigExclusions) {
7876
this.maxVotingConfigExclusions = maxVotingConfigExclusions;
7977
}
8078

81-
@Override
82-
protected String executor() {
83-
return Names.SAME;
84-
}
85-
86-
@Override
87-
protected AddVotingConfigExclusionsResponse read(StreamInput in) throws IOException {
88-
return new AddVotingConfigExclusionsResponse(in);
89-
}
90-
9179
@Override
9280
protected void masterOperation(AddVotingConfigExclusionsRequest request, ClusterState state,
9381
ActionListener<AddVotingConfigExclusionsResponse> listener) throws Exception {

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

+2-14
Original file line numberDiff line numberDiff line change
@@ -38,13 +38,10 @@
3838
import org.elasticsearch.cluster.service.ClusterService;
3939
import org.elasticsearch.common.Priority;
4040
import org.elasticsearch.common.inject.Inject;
41-
import org.elasticsearch.common.io.stream.StreamInput;
4241
import org.elasticsearch.common.unit.TimeValue;
4342
import org.elasticsearch.threadpool.ThreadPool;
44-
import org.elasticsearch.threadpool.ThreadPool.Names;
4543
import org.elasticsearch.transport.TransportService;
4644

47-
import java.io.IOException;
4845
import java.util.function.Predicate;
4946

5047
public class TransportClearVotingConfigExclusionsAction
@@ -57,17 +54,8 @@ public TransportClearVotingConfigExclusionsAction(TransportService transportServ
5754
ThreadPool threadPool, ActionFilters actionFilters,
5855
IndexNameExpressionResolver indexNameExpressionResolver) {
5956
super(ClearVotingConfigExclusionsAction.NAME, transportService, clusterService, threadPool, actionFilters,
60-
ClearVotingConfigExclusionsRequest::new, indexNameExpressionResolver);
61-
}
62-
63-
@Override
64-
protected String executor() {
65-
return Names.SAME;
66-
}
67-
68-
@Override
69-
protected ClearVotingConfigExclusionsResponse read(StreamInput in) throws IOException {
70-
return new ClearVotingConfigExclusionsResponse(in);
57+
ClearVotingConfigExclusionsRequest::new, indexNameExpressionResolver, ClearVotingConfigExclusionsResponse::new,
58+
ThreadPool.Names.SAME);
7159
}
7260

7361
@Override

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

+1-14
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,6 @@
4141
import org.elasticsearch.cluster.service.ClusterService;
4242
import org.elasticsearch.common.Strings;
4343
import org.elasticsearch.common.inject.Inject;
44-
import org.elasticsearch.common.io.stream.StreamInput;
4544
import org.elasticsearch.common.unit.TimeValue;
4645
import org.elasticsearch.common.util.CollectionUtils;
4746
import org.elasticsearch.index.IndexNotFoundException;
@@ -50,7 +49,6 @@
5049
import org.elasticsearch.threadpool.ThreadPool;
5150
import org.elasticsearch.transport.TransportService;
5251

53-
import java.io.IOException;
5452
import java.util.function.Consumer;
5553
import java.util.function.Predicate;
5654

@@ -65,21 +63,10 @@ public TransportClusterHealthAction(TransportService transportService, ClusterSe
6563
ThreadPool threadPool, ActionFilters actionFilters,
6664
IndexNameExpressionResolver indexNameExpressionResolver, AllocationService allocationService) {
6765
super(ClusterHealthAction.NAME, false, transportService, clusterService, threadPool, actionFilters,
68-
ClusterHealthRequest::new, indexNameExpressionResolver);
66+
ClusterHealthRequest::new, indexNameExpressionResolver, ClusterHealthResponse::new, ThreadPool.Names.SAME);
6967
this.allocationService = allocationService;
7068
}
7169

72-
@Override
73-
protected String executor() {
74-
// this should be executing quickly no need to fork off
75-
return ThreadPool.Names.SAME;
76-
}
77-
78-
@Override
79-
protected ClusterHealthResponse read(StreamInput in) throws IOException {
80-
return new ClusterHealthResponse(in);
81-
}
82-
8370
@Override
8471
protected ClusterBlockException checkBlock(ClusterHealthRequest request, ClusterState state) {
8572
// 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

+1-13
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,6 @@
3939
import org.elasticsearch.cluster.service.ClusterService;
4040
import org.elasticsearch.common.Nullable;
4141
import org.elasticsearch.common.inject.Inject;
42-
import org.elasticsearch.common.io.stream.StreamInput;
4342
import org.elasticsearch.repositories.RepositoriesService;
4443
import org.elasticsearch.repositories.Repository;
4544
import org.elasticsearch.repositories.RepositoryCleanupResult;
@@ -49,7 +48,6 @@
4948
import org.elasticsearch.threadpool.ThreadPool;
5049
import org.elasticsearch.transport.TransportService;
5150

52-
import java.io.IOException;
5351
import java.util.Collections;
5452

5553
/**
@@ -80,18 +78,13 @@ public final class TransportCleanupRepositoryAction extends TransportMasterNodeA
8078

8179
private final SnapshotsService snapshotsService;
8280

83-
@Override
84-
protected String executor() {
85-
return ThreadPool.Names.SAME;
86-
}
87-
8881
@Inject
8982
public TransportCleanupRepositoryAction(TransportService transportService, ClusterService clusterService,
9083
RepositoriesService repositoriesService, SnapshotsService snapshotsService,
9184
ThreadPool threadPool, ActionFilters actionFilters,
9285
IndexNameExpressionResolver indexNameExpressionResolver) {
9386
super(CleanupRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
94-
CleanupRepositoryRequest::new, indexNameExpressionResolver);
87+
CleanupRepositoryRequest::new, indexNameExpressionResolver, CleanupRepositoryResponse::new, ThreadPool.Names.SAME);
9588
this.repositoriesService = repositoriesService;
9689
this.snapshotsService = snapshotsService;
9790
// We add a state applier that will remove any dangling repository cleanup actions on master failover.
@@ -138,11 +131,6 @@ private static ClusterState removeInProgressCleanup(final ClusterState currentSt
138131
: currentState;
139132
}
140133

141-
@Override
142-
protected CleanupRepositoryResponse read(StreamInput in) throws IOException {
143-
return new CleanupRepositoryResponse(in);
144-
}
145-
146134
@Override
147135
protected void masterOperation(CleanupRepositoryRequest request, ClusterState state,
148136
ActionListener<CleanupRepositoryResponse> listener) {

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

+1-6
Original file line numberDiff line numberDiff line change
@@ -45,15 +45,10 @@ public TransportDeleteRepositoryAction(TransportService transportService, Cluste
4545
RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
4646
IndexNameExpressionResolver indexNameExpressionResolver) {
4747
super(DeleteRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
48-
DeleteRepositoryRequest::new, indexNameExpressionResolver);
48+
DeleteRepositoryRequest::new, indexNameExpressionResolver, ThreadPool.Names.SAME);
4949
this.repositoriesService = repositoriesService;
5050
}
5151

52-
@Override
53-
protected String executor() {
54-
return ThreadPool.Names.SAME;
55-
}
56-
5752
@Override
5853
protected ClusterBlockException checkBlock(DeleteRepositoryRequest request, ClusterState state) {
5954
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);

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

+1-13
Original file line numberDiff line numberDiff line change
@@ -31,13 +31,11 @@
3131
import org.elasticsearch.cluster.metadata.RepositoryMetadata;
3232
import org.elasticsearch.cluster.service.ClusterService;
3333
import org.elasticsearch.common.inject.Inject;
34-
import org.elasticsearch.common.io.stream.StreamInput;
3534
import org.elasticsearch.common.regex.Regex;
3635
import org.elasticsearch.repositories.RepositoryMissingException;
3736
import org.elasticsearch.threadpool.ThreadPool;
3837
import org.elasticsearch.transport.TransportService;
3938

40-
import java.io.IOException;
4139
import java.util.ArrayList;
4240
import java.util.Collections;
4341
import java.util.LinkedHashSet;
@@ -55,17 +53,7 @@ public TransportGetRepositoriesAction(TransportService transportService, Cluster
5553
ThreadPool threadPool, ActionFilters actionFilters,
5654
IndexNameExpressionResolver indexNameExpressionResolver) {
5755
super(GetRepositoriesAction.NAME, transportService, clusterService, threadPool, actionFilters,
58-
GetRepositoriesRequest::new, indexNameExpressionResolver);
59-
}
60-
61-
@Override
62-
protected String executor() {
63-
return ThreadPool.Names.SAME;
64-
}
65-
66-
@Override
67-
protected GetRepositoriesResponse read(StreamInput in) throws IOException {
68-
return new GetRepositoriesResponse(in);
56+
GetRepositoriesRequest::new, indexNameExpressionResolver, GetRepositoriesResponse::new, ThreadPool.Names.SAME);
6957
}
7058

7159
@Override

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

+1-6
Original file line numberDiff line numberDiff line change
@@ -45,15 +45,10 @@ public TransportPutRepositoryAction(TransportService transportService, ClusterSe
4545
RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
4646
IndexNameExpressionResolver indexNameExpressionResolver) {
4747
super(PutRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
48-
PutRepositoryRequest::new, indexNameExpressionResolver);
48+
PutRepositoryRequest::new, indexNameExpressionResolver, ThreadPool.Names.SAME);
4949
this.repositoriesService = repositoriesService;
5050
}
5151

52-
@Override
53-
protected String executor() {
54-
return ThreadPool.Names.SAME;
55-
}
56-
5752
@Override
5853
protected ClusterBlockException checkBlock(PutRepositoryRequest request, ClusterState state) {
5954
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);

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

+1-14
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,10 @@
2929
import org.elasticsearch.cluster.node.DiscoveryNode;
3030
import org.elasticsearch.cluster.service.ClusterService;
3131
import org.elasticsearch.common.inject.Inject;
32-
import org.elasticsearch.common.io.stream.StreamInput;
3332
import org.elasticsearch.repositories.RepositoriesService;
3433
import org.elasticsearch.threadpool.ThreadPool;
3534
import org.elasticsearch.transport.TransportService;
3635

37-
import java.io.IOException;
38-
3936
/**
4037
* Transport action for verifying repository operation
4138
*/
@@ -50,20 +47,10 @@ public TransportVerifyRepositoryAction(TransportService transportService, Cluste
5047
RepositoriesService repositoriesService, ThreadPool threadPool, ActionFilters actionFilters,
5148
IndexNameExpressionResolver indexNameExpressionResolver) {
5249
super(VerifyRepositoryAction.NAME, transportService, clusterService, threadPool, actionFilters,
53-
VerifyRepositoryRequest::new, indexNameExpressionResolver);
50+
VerifyRepositoryRequest::new, indexNameExpressionResolver, VerifyRepositoryResponse::new, ThreadPool.Names.SAME);
5451
this.repositoriesService = repositoriesService;
5552
}
5653

57-
@Override
58-
protected String executor() {
59-
return ThreadPool.Names.SAME;
60-
}
61-
62-
@Override
63-
protected VerifyRepositoryResponse read(StreamInput in) throws IOException {
64-
return new VerifyRepositoryResponse(in);
65-
}
66-
6754
@Override
6855
protected ClusterBlockException checkBlock(VerifyRepositoryRequest request, ClusterState state) {
6956
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);

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

+1-14
Original file line numberDiff line numberDiff line change
@@ -47,11 +47,9 @@
4747
import org.elasticsearch.common.collect.ImmutableOpenIntMap;
4848
import org.elasticsearch.common.collect.ImmutableOpenMap;
4949
import org.elasticsearch.common.inject.Inject;
50-
import org.elasticsearch.common.io.stream.StreamInput;
5150
import org.elasticsearch.threadpool.ThreadPool;
5251
import org.elasticsearch.transport.TransportService;
5352

54-
import java.io.IOException;
5553
import java.util.ArrayList;
5654
import java.util.HashMap;
5755
import java.util.List;
@@ -68,26 +66,15 @@ public TransportClusterRerouteAction(TransportService transportService, ClusterS
6866
ThreadPool threadPool, AllocationService allocationService, ActionFilters actionFilters,
6967
IndexNameExpressionResolver indexNameExpressionResolver) {
7068
super(ClusterRerouteAction.NAME, transportService, clusterService, threadPool, actionFilters,
71-
ClusterRerouteRequest::new, indexNameExpressionResolver);
69+
ClusterRerouteRequest::new, indexNameExpressionResolver, ClusterRerouteResponse::new, ThreadPool.Names.SAME);
7270
this.allocationService = allocationService;
7371
}
7472

75-
@Override
76-
protected String executor() {
77-
// we go async right away
78-
return ThreadPool.Names.SAME;
79-
}
80-
8173
@Override
8274
protected ClusterBlockException checkBlock(ClusterRerouteRequest request, ClusterState state) {
8375
return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
8476
}
8577

86-
@Override
87-
protected ClusterRerouteResponse read(StreamInput in) throws IOException {
88-
return new ClusterRerouteResponse(in);
89-
}
90-
9178
@Override
9279
protected void masterOperation(final ClusterRerouteRequest request, final ClusterState state,
9380
final ActionListener<ClusterRerouteResponse> listener) {

0 commit comments

Comments
 (0)