Skip to content

Commit a644bc0

Browse files
tlrxtlrx
authored
Add unit tests for ShardStateAction's ShardStartedClusterStateTaskExecutor (#37756)
1 parent dfecb25 commit a644bc0

File tree

5 files changed

+281
-79
lines changed

5 files changed

+281
-79
lines changed

‎server/src/test/java/org/elasticsearch/action/admin/cluster/allocation/ClusterAllocationExplainActionTests.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import org.elasticsearch.cluster.routing.RoutingNode;
2525
import org.elasticsearch.cluster.routing.ShardRouting;
2626
import org.elasticsearch.cluster.routing.ShardRoutingState;
27+
import org.elasticsearch.cluster.routing.UnassignedInfo;
28+
import org.elasticsearch.cluster.routing.allocation.AllocationDecision;
2729
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
2830
import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
2931
import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
@@ -35,6 +37,7 @@
3537
import org.elasticsearch.test.ESTestCase;
3638
import org.elasticsearch.test.gateway.TestGatewayAllocator;
3739

40+
import java.time.Instant;
3841
import java.util.Collections;
3942
import java.util.Locale;
4043

@@ -85,7 +88,16 @@ public ShardAllocationDecision decideShardAllocation(ShardRouting shard, Routing
8588
"wait until initialization has completed";
8689
}
8790
assertEquals("{\"index\":\"idx\",\"shard\":0,\"primary\":true,\"current_state\":\"" +
88-
shardRoutingState.toString().toLowerCase(Locale.ROOT) + "\",\"current_node\":" +
91+
shardRoutingState.toString().toLowerCase(Locale.ROOT) + "\"" +
92+
(shard.unassignedInfo() != null ?
93+
",\"unassigned_info\":{"
94+
+ "\"reason\":\"" + shard.unassignedInfo().getReason() + "\","
95+
+ "\"at\":\""+ UnassignedInfo.DATE_TIME_FORMATTER.format(
96+
Instant.ofEpochMilli(shard.unassignedInfo().getUnassignedTimeInMillis())) + "\","
97+
+ "\"last_allocation_status\":\"" + AllocationDecision.fromAllocationStatus(
98+
shard.unassignedInfo().getLastAllocationStatus()) + "\"}"
99+
: "")
100+
+ ",\"current_node\":" +
89101
"{\"id\":\"" + cae.getCurrentNode().getId() + "\",\"name\":\"" + cae.getCurrentNode().getName() +
90102
"\",\"transport_address\":\"" + cae.getCurrentNode().getAddress() +
91103
"\"},\"explanation\":\"" + explanation + "\"}", Strings.toString(builder));

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,8 @@ public static ClusterState state(String index, boolean activePrimaryLocal, Shard
118118
}
119119
if (primaryState == ShardRoutingState.RELOCATING) {
120120
relocatingNode = selectAndRemove(unassignedNodes);
121+
} else if (primaryState == ShardRoutingState.INITIALIZING) {
122+
unassignedInfo = new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null);
121123
}
122124
} else {
123125
unassignedInfo = new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, null);

‎server/src/test/java/org/elasticsearch/cluster/action/shard/ShardFailedClusterStateTaskExecutorTests.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,6 @@
4848
import org.elasticsearch.common.util.set.Sets;
4949
import org.elasticsearch.index.Index;
5050
import org.elasticsearch.index.shard.ShardId;
51-
import org.junit.Before;
5251

5352
import java.util.ArrayList;
5453
import java.util.Arrays;
@@ -73,7 +72,7 @@ public class ShardFailedClusterStateTaskExecutorTests extends ESAllocationTestCa
7372
private ClusterState clusterState;
7473
private ShardStateAction.ShardFailedClusterStateTaskExecutor executor;
7574

76-
@Before
75+
@Override
7776
public void setUp() throws Exception {
7877
super.setUp();
7978
allocationService = createAllocationService(Settings.builder()

‎server/src/test/java/org/elasticsearch/cluster/action/shard/ShardStartedClusterStateTaskExecutorTests.java

Lines changed: 197 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.cluster.action.shard;
21+
22+
import org.elasticsearch.cluster.ClusterState;
23+
import org.elasticsearch.cluster.ClusterStateTaskExecutor;
24+
import org.elasticsearch.cluster.ESAllocationTestCase;
25+
import org.elasticsearch.cluster.action.shard.ShardStateAction.StartedShardEntry;
26+
import org.elasticsearch.cluster.metadata.IndexMetaData;
27+
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
28+
import org.elasticsearch.cluster.routing.ShardRouting;
29+
import org.elasticsearch.cluster.routing.ShardRoutingState;
30+
import org.elasticsearch.cluster.routing.allocation.AllocationService;
31+
import org.elasticsearch.common.settings.Settings;
32+
import org.elasticsearch.index.shard.ShardId;
33+
34+
import java.util.ArrayList;
35+
import java.util.Collections;
36+
import java.util.List;
37+
import java.util.function.Consumer;
38+
import java.util.stream.Collectors;
39+
import java.util.stream.IntStream;
40+
import java.util.stream.Stream;
41+
42+
import static java.util.Collections.singletonList;
43+
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.state;
44+
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.stateWithActivePrimary;
45+
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.stateWithAssignedPrimariesAndReplicas;
46+
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.stateWithNoShard;
47+
import static org.elasticsearch.cluster.routing.allocation.decider.ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES_SETTING;
48+
import static org.hamcrest.Matchers.equalTo;
49+
import static org.hamcrest.Matchers.is;
50+
import static org.hamcrest.Matchers.notNullValue;
51+
52+
public class ShardStartedClusterStateTaskExecutorTests extends ESAllocationTestCase {
53+
54+
private ShardStateAction.ShardStartedClusterStateTaskExecutor executor;
55+
56+
@Override
57+
public void setUp() throws Exception {
58+
super.setUp();
59+
AllocationService allocationService = createAllocationService(Settings.builder()
60+
.put(CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES_SETTING.getKey(), Integer.MAX_VALUE)
61+
.build());
62+
executor = new ShardStateAction.ShardStartedClusterStateTaskExecutor(allocationService, logger);
63+
}
64+
65+
public void testEmptyTaskListProducesSameClusterState() throws Exception {
66+
final ClusterState clusterState = stateWithNoShard();
67+
assertTasksExecution(clusterState, Collections.emptyList(), result -> assertSame(clusterState, result.resultingState));
68+
}
69+
70+
public void testNonExistentIndexMarkedAsSuccessful() throws Exception {
71+
final ClusterState clusterState = stateWithNoShard();
72+
final StartedShardEntry entry = new StartedShardEntry(new ShardId("test", "_na", 0), "aId", "test");
73+
assertTasksExecution(clusterState, singletonList(entry),
74+
result -> {
75+
assertSame(clusterState, result.resultingState);
76+
assertThat(result.executionResults.size(), equalTo(1));
77+
assertThat(result.executionResults.containsKey(entry), is(true));
78+
assertThat(((ClusterStateTaskExecutor.TaskResult) result.executionResults.get(entry)).isSuccess(), is(true));
79+
});
80+
}
81+
82+
public void testNonExistentShardsAreMarkedAsSuccessful() throws Exception {
83+
final String indexName = "test";
84+
final ClusterState clusterState = stateWithActivePrimary(indexName, true, randomInt(2), randomInt(2));
85+
86+
final IndexMetaData indexMetaData = clusterState.metaData().index(indexName);
87+
final List<StartedShardEntry> tasks = Stream.concat(
88+
// Existent shard id but different allocation id
89+
IntStream.range(0, randomIntBetween(1, 5))
90+
.mapToObj(i -> new StartedShardEntry(new ShardId(indexMetaData.getIndex(), 0), String.valueOf(i), "allocation id")),
91+
// Non existent shard id
92+
IntStream.range(1, randomIntBetween(2, 5))
93+
.mapToObj(i -> new StartedShardEntry(new ShardId(indexMetaData.getIndex(), i), String.valueOf(i), "shard id"))
94+
95+
).collect(Collectors.toList());
96+
97+
assertTasksExecution(clusterState, tasks, result -> {
98+
assertSame(clusterState, result.resultingState);
99+
assertThat(result.executionResults.size(), equalTo(tasks.size()));
100+
tasks.forEach(task -> {
101+
assertThat(result.executionResults.containsKey(task), is(true));
102+
assertThat(((ClusterStateTaskExecutor.TaskResult) result.executionResults.get(task)).isSuccess(), is(true));
103+
});
104+
});
105+
}
106+
107+
public void testNonInitializingShardAreMarkedAsSuccessful() throws Exception {
108+
final String indexName = "test";
109+
final ClusterState clusterState = stateWithAssignedPrimariesAndReplicas(new String[]{indexName}, randomIntBetween(2, 10), 1);
110+
111+
final IndexMetaData indexMetaData = clusterState.metaData().index(indexName);
112+
final List<StartedShardEntry> tasks = IntStream.range(0, randomIntBetween(1, indexMetaData.getNumberOfShards()))
113+
.mapToObj(i -> {
114+
final ShardId shardId = new ShardId(indexMetaData.getIndex(), i);
115+
final IndexShardRoutingTable shardRoutingTable = clusterState.routingTable().shardRoutingTable(shardId);
116+
final String allocationId;
117+
if (randomBoolean()) {
118+
allocationId = shardRoutingTable.primaryShard().allocationId().getId();
119+
} else {
120+
allocationId = shardRoutingTable.replicaShards().iterator().next().allocationId().getId();
121+
}
122+
return new StartedShardEntry(shardId, allocationId, "test");
123+
}).collect(Collectors.toList());
124+
125+
assertTasksExecution(clusterState, tasks, result -> {
126+
assertSame(clusterState, result.resultingState);
127+
assertThat(result.executionResults.size(), equalTo(tasks.size()));
128+
tasks.forEach(task -> {
129+
assertThat(result.executionResults.containsKey(task), is(true));
130+
assertThat(((ClusterStateTaskExecutor.TaskResult) result.executionResults.get(task)).isSuccess(), is(true));
131+
});
132+
});
133+
}
134+
135+
public void testStartedShards() throws Exception {
136+
final String indexName = "test";
137+
final ClusterState clusterState = state(indexName, randomBoolean(), ShardRoutingState.INITIALIZING, ShardRoutingState.INITIALIZING);
138+
139+
final IndexMetaData indexMetaData = clusterState.metaData().index(indexName);
140+
final ShardId shardId = new ShardId(indexMetaData.getIndex(), 0);
141+
final ShardRouting primaryShard = clusterState.routingTable().shardRoutingTable(shardId).primaryShard();
142+
final String primaryAllocationId = primaryShard.allocationId().getId();
143+
144+
final List<StartedShardEntry> tasks = new ArrayList<>();
145+
tasks.add(new StartedShardEntry(shardId, primaryAllocationId, "test"));
146+
if (randomBoolean()) {
147+
final ShardRouting replicaShard = clusterState.routingTable().shardRoutingTable(shardId).replicaShards().iterator().next();
148+
final String replicaAllocationId = replicaShard.allocationId().getId();
149+
tasks.add(new StartedShardEntry(shardId, replicaAllocationId, "test"));
150+
}
151+
assertTasksExecution(clusterState, tasks, result -> {
152+
assertNotSame(clusterState, result.resultingState);
153+
assertThat(result.executionResults.size(), equalTo(tasks.size()));
154+
tasks.forEach(task -> {
155+
assertThat(result.executionResults.containsKey(task), is(true));
156+
assertThat(((ClusterStateTaskExecutor.TaskResult) result.executionResults.get(task)).isSuccess(), is(true));
157+
158+
final IndexShardRoutingTable shardRoutingTable = result.resultingState.routingTable().shardRoutingTable(task.shardId);
159+
assertThat(shardRoutingTable.getByAllocationId(task.allocationId).state(), is(ShardRoutingState.STARTED));
160+
});
161+
});
162+
}
163+
164+
public void testDuplicateStartsAreOkay() throws Exception {
165+
final String indexName = "test";
166+
final ClusterState clusterState = state(indexName, randomBoolean(), ShardRoutingState.INITIALIZING);
167+
168+
final IndexMetaData indexMetaData = clusterState.metaData().index(indexName);
169+
final ShardId shardId = new ShardId(indexMetaData.getIndex(), 0);
170+
final ShardRouting shardRouting = clusterState.routingTable().shardRoutingTable(shardId).primaryShard();
171+
final String allocationId = shardRouting.allocationId().getId();
172+
173+
final List<StartedShardEntry> tasks = IntStream.range(0, randomIntBetween(2, 10))
174+
.mapToObj(i -> new StartedShardEntry(shardId, allocationId, "test"))
175+
.collect(Collectors.toList());
176+
177+
assertTasksExecution(clusterState, tasks, result -> {
178+
assertNotSame(clusterState, result.resultingState);
179+
assertThat(result.executionResults.size(), equalTo(tasks.size()));
180+
tasks.forEach(task -> {
181+
assertThat(result.executionResults.containsKey(task), is(true));
182+
assertThat(((ClusterStateTaskExecutor.TaskResult) result.executionResults.get(task)).isSuccess(), is(true));
183+
184+
final IndexShardRoutingTable shardRoutingTable = result.resultingState.routingTable().shardRoutingTable(task.shardId);
185+
assertThat(shardRoutingTable.getByAllocationId(task.allocationId).state(), is(ShardRoutingState.STARTED));
186+
});
187+
});
188+
}
189+
190+
private void assertTasksExecution(final ClusterState state,
191+
final List<StartedShardEntry> tasks,
192+
final Consumer<ClusterStateTaskExecutor.ClusterTasksResult> consumer) throws Exception {
193+
final ClusterStateTaskExecutor.ClusterTasksResult<StartedShardEntry> result = executor.execute(state, tasks);
194+
assertThat(result, notNullValue());
195+
consumer.accept(result);
196+
}
197+
}

0 commit comments

Comments
 (0)