Skip to content

Commit d8956d2

Browse files
DaveCTurnerDaveCTurner
committed
Remove test-only customisation from TransReplAct (#40863)
The `getIndexShard()` and `sendReplicaRequest()` methods in TransportReplicationAction are effectively only used to customise some behaviour in tests. However there are other ways to do this that do not cause such an obstacle to separating the TransportReplicationAction into its two halves (see #40706). This commit removes these customisation points and injects the test-only behaviour using other techniques.
1 parent 669d72e commit d8956d2

File tree

5 files changed

+108
-114
lines changed

5 files changed

+108
-114
lines changed

‎server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java‎

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import org.elasticsearch.cluster.action.shard.ShardStateAction;
3030
import org.elasticsearch.cluster.block.ClusterBlockLevel;
3131
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
32-
import org.elasticsearch.cluster.node.DiscoveryNode;
3332
import org.elasticsearch.cluster.routing.ShardRouting;
3433
import org.elasticsearch.cluster.service.ClusterService;
3534
import org.elasticsearch.common.inject.Inject;
@@ -74,19 +73,6 @@ protected ReplicationOperation.Replicas newReplicasProxy(long primaryTerm) {
7473
return new ResyncActionReplicasProxy(primaryTerm);
7574
}
7675

77-
@Override
78-
protected void sendReplicaRequest(
79-
final ConcreteReplicaRequest<ResyncReplicationRequest> replicaRequest,
80-
final DiscoveryNode node,
81-
final ActionListener<ReplicationOperation.ReplicaResponse> listener) {
82-
if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
83-
super.sendReplicaRequest(replicaRequest, node, listener);
84-
} else {
85-
final long pre60NodeCheckpoint = SequenceNumbers.PRE_60_NODE_CHECKPOINT;
86-
listener.onResponse(new ReplicaResponse(pre60NodeCheckpoint, pre60NodeCheckpoint));
87-
}
88-
}
89-
9076
@Override
9177
protected ClusterBlockLevel globalBlockLevel() {
9278
// resync should never be blocked because it's an internal action

‎server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java‎

Lines changed: 7 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -619,7 +619,7 @@ public void onFailure(Exception e) {
619619
}
620620
}
621621

622-
protected IndexShard getIndexShard(final ShardId shardId) {
622+
private IndexShard getIndexShard(final ShardId shardId) {
623623
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
624624
return indexService.getShard(shardId.id());
625625
}
@@ -1058,7 +1058,12 @@ public void performOn(
10581058
}
10591059
final ConcreteReplicaRequest<ReplicaRequest> replicaRequest = new ConcreteReplicaRequest<>(
10601060
request, replica.allocationId().getId(), primaryTerm, globalCheckpoint, maxSeqNoOfUpdatesOrDeletes);
1061-
sendReplicaRequest(replicaRequest, node, listener);
1061+
final ActionListenerResponseHandler<ReplicaResponse> handler = new ActionListenerResponseHandler<>(listener, in -> {
1062+
ReplicaResponse replicaResponse = new ReplicaResponse();
1063+
replicaResponse.readFrom(in);
1064+
return replicaResponse;
1065+
});
1066+
transportService.sendRequest(node, transportReplicaAction, replicaRequest, transportOptions, handler);
10621067
}
10631068

10641069
@Override
@@ -1080,25 +1085,6 @@ public void markShardCopyAsStaleIfNeeded(ShardId shardId, String allocationId, A
10801085
}
10811086
}
10821087

1083-
/**
1084-
* Sends the specified replica request to the specified node.
1085-
*
1086-
* @param replicaRequest the replica request
1087-
* @param node the node to send the request to
1088-
* @param listener callback for handling the response or failure
1089-
*/
1090-
protected void sendReplicaRequest(
1091-
final ConcreteReplicaRequest<ReplicaRequest> replicaRequest,
1092-
final DiscoveryNode node,
1093-
final ActionListener<ReplicationOperation.ReplicaResponse> listener) {
1094-
final ActionListenerResponseHandler<ReplicaResponse> handler = new ActionListenerResponseHandler<>(listener, in -> {
1095-
ReplicaResponse replicaResponse = new ReplicaResponse();
1096-
replicaResponse.readFrom(in);
1097-
return replicaResponse;
1098-
});
1099-
transportService.sendRequest(node, transportReplicaAction, replicaRequest, transportOptions, handler);
1100-
}
1101-
11021088
/** a wrapper class to encapsulate a request when being sent to a specific allocation id **/
11031089
public static class ConcreteShardRequest<R extends TransportRequest> extends TransportRequest {
11041090

‎server/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java‎

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,13 @@
2222
import org.apache.logging.log4j.message.ParameterizedMessage;
2323
import org.apache.lucene.store.AlreadyClosedException;
2424
import org.elasticsearch.ExceptionsHelper;
25-
import org.elasticsearch.Version;
2625
import org.elasticsearch.action.ActionListener;
2726
import org.elasticsearch.action.support.ActionFilters;
28-
import org.elasticsearch.action.support.replication.ReplicationOperation;
2927
import org.elasticsearch.action.support.replication.ReplicationRequest;
3028
import org.elasticsearch.action.support.replication.ReplicationResponse;
3129
import org.elasticsearch.action.support.replication.TransportReplicationAction;
3230
import org.elasticsearch.cluster.action.shard.ShardStateAction;
3331
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
34-
import org.elasticsearch.cluster.node.DiscoveryNode;
3532
import org.elasticsearch.cluster.service.ClusterService;
3633
import org.elasticsearch.common.inject.Inject;
3734
import org.elasticsearch.common.settings.Settings;
@@ -103,19 +100,6 @@ protected ReplicationResponse newResponseInstance() {
103100
return new ReplicationResponse();
104101
}
105102

106-
@Override
107-
protected void sendReplicaRequest(
108-
final ConcreteReplicaRequest<Request> replicaRequest,
109-
final DiscoveryNode node,
110-
final ActionListener<ReplicationOperation.ReplicaResponse> listener) {
111-
if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
112-
super.sendReplicaRequest(replicaRequest, node, listener);
113-
} else {
114-
final long pre60NodeCheckpoint = SequenceNumbers.PRE_60_NODE_CHECKPOINT;
115-
listener.onResponse(new ReplicaResponse(pre60NodeCheckpoint, pre60NodeCheckpoint));
116-
}
117-
}
118-
119103
@Override
120104
protected PrimaryResult<Request, ReplicationResponse> shardOperationOnPrimary(
121105
final Request request, final IndexShard indexShard) throws Exception {

‎server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationActionTests.java‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -757,7 +757,7 @@ public void testReplicaProxy() throws InterruptedException, ExecutionException {
757757
assertEquals(0, shardFailedRequests.length);
758758
}
759759

760-
public void testSeqNoIsSetOnPrimary() throws Exception {
760+
public void testSeqNoIsSetOnPrimary() {
761761
final String index = "test";
762762
final ShardId shardId = new ShardId(index, "_na_", 0);
763763
// we use one replica to check the primary term was set on the operation and sent to the replica
@@ -788,14 +788,14 @@ public void testSeqNoIsSetOnPrimary() throws Exception {
788788
return null;
789789
}).when(shard).acquirePrimaryOperationPermit(any(), anyString(), anyObject());
790790

791-
TestAction action =
792-
new TestAction(Settings.EMPTY, "internal:testSeqNoIsSetOnPrimary", transportService, clusterService, shardStateAction,
793-
threadPool) {
794-
@Override
795-
protected IndexShard getIndexShard(ShardId shardId) {
796-
return shard;
797-
}
798-
};
791+
final IndexService indexService = mock(IndexService.class);
792+
when(indexService.getShard(shard.shardId().id())).thenReturn(shard);
793+
794+
final IndicesService indicesService = mock(IndicesService.class);
795+
when(indicesService.indexServiceSafe(shard.shardId().getIndex())).thenReturn(indexService);
796+
797+
TestAction action = new TestAction(Settings.EMPTY, "internal:testSeqNoIsSetOnPrimary", transportService, clusterService,
798+
shardStateAction, threadPool, indicesService);
799799

800800
action.handlePrimaryRequest(concreteShardRequest, createTransportChannel(listener), null);
801801
CapturingTransport.CapturedRequest[] requestsToReplicas = transport.capturedRequests();
@@ -1207,11 +1207,16 @@ static class TestResponse extends ReplicationResponse {
12071207

12081208
private class TestAction extends TransportReplicationAction<Request, Request, TestResponse> {
12091209

1210-
12111210
TestAction(Settings settings, String actionName, TransportService transportService,
12121211
ClusterService clusterService, ShardStateAction shardStateAction,
12131212
ThreadPool threadPool) {
1214-
super(settings, actionName, transportService, clusterService, mockIndicesService(clusterService), threadPool,
1213+
this(settings, actionName, transportService, clusterService, shardStateAction, threadPool, mockIndicesService(clusterService));
1214+
}
1215+
1216+
TestAction(Settings settings, String actionName, TransportService transportService,
1217+
ClusterService clusterService, ShardStateAction shardStateAction,
1218+
ThreadPool threadPool, IndicesService indicesService) {
1219+
super(settings, actionName, transportService, clusterService, indicesService, threadPool,
12151220
shardStateAction,
12161221
new ActionFilters(new HashSet<>()), new IndexNameExpressionResolver(),
12171222
Request::new, Request::new, ThreadPool.Names.SAME);
@@ -1241,7 +1246,7 @@ protected boolean resolveIndex() {
12411246
}
12421247
}
12431248

1244-
final IndicesService mockIndicesService(ClusterService clusterService) {
1249+
private IndicesService mockIndicesService(ClusterService clusterService) {
12451250
final IndicesService indicesService = mock(IndicesService.class);
12461251
when(indicesService.indexServiceSafe(any(Index.class))).then(invocation -> {
12471252
Index index = (Index)invocation.getArguments()[0];
@@ -1261,7 +1266,7 @@ final IndicesService mockIndicesService(ClusterService clusterService) {
12611266
return indicesService;
12621267
}
12631268

1264-
final IndexService mockIndexService(final IndexMetaData indexMetaData, ClusterService clusterService) {
1269+
private IndexService mockIndexService(final IndexMetaData indexMetaData, ClusterService clusterService) {
12651270
final IndexService indexService = mock(IndexService.class);
12661271
when(indexService.getShard(anyInt())).then(invocation -> {
12671272
int shard = (Integer) invocation.getArguments()[0];

0 commit comments

Comments
 (0)