Skip to content

Commit 1922aae

Browse files
committed
Remove reset and remaining from SearchShardIterator
remaining is used in tests only since the recent change to counting in AbstractSearchAsyncAction. reset was used in a couple of places where it does not seem like it's needed anymore.
1 parent eb6a49b commit 1922aae

File tree

5 files changed

+9
-40
lines changed

5 files changed

+9
-40
lines changed

server/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -434,7 +434,6 @@ private synchronized List<SearchShardIterator> getIterator(List<SearchShardItera
434434
}
435435
int i = 0;
436436
for (SearchShardIterator iter : shardsIts) {
437-
iter.reset();
438437
boolean match = possibleMatches.get(i++);
439438
if (match) {
440439
assert iter.skip() == false;

server/src/main/java/org/elasticsearch/action/search/SearchShardIterator.java

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111

1212
import org.elasticsearch.action.OriginalIndices;
1313
import org.elasticsearch.cluster.routing.ShardRouting;
14-
import org.elasticsearch.cluster.routing.ShardsIterator;
1514
import org.elasticsearch.common.util.Countable;
1615
import org.elasticsearch.common.util.PlainIterator;
1716
import org.elasticsearch.core.Nullable;
@@ -114,15 +113,6 @@ SearchShardTarget nextOrNull() {
114113
return null;
115114
}
116115

117-
/**
118-
* Return the number of shards remaining in this {@link ShardsIterator}
119-
*
120-
* @return number of shard remaining
121-
*/
122-
int remaining() {
123-
return targetNodesIterator.remaining();
124-
}
125-
126116
/**
127117
* Returns a non-null value if this request should use a specific search context instead of the latest one.
128118
*/
@@ -138,13 +128,6 @@ List<String> getTargetNodeIds() {
138128
return targetNodesIterator.asList();
139129
}
140130

141-
/**
142-
* Resets the iterator to its initial state.
143-
*/
144-
void reset() {
145-
targetNodesIterator.reset();
146-
}
147-
148131
/**
149132
* Returns <code>true</code> if the search execution should skip this shard since it can not match any documents given the query.
150133
*/

server/src/main/java/org/elasticsearch/action/search/TransportSearchShardsAction.java

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -172,15 +172,7 @@ public void searchShards(Task task, SearchShardsRequest searchShardsRequest, Act
172172
private static List<SearchShardsGroup> toGroups(List<SearchShardIterator> shardIts) {
173173
List<SearchShardsGroup> groups = new ArrayList<>(shardIts.size());
174174
for (SearchShardIterator shardIt : shardIts) {
175-
boolean skip = shardIt.skip();
176-
shardIt.reset();
177-
List<String> targetNodes = new ArrayList<>();
178-
SearchShardTarget target;
179-
while ((target = shardIt.nextOrNull()) != null) {
180-
targetNodes.add(target.getNodeId());
181-
}
182-
ShardId shardId = shardIt.shardId();
183-
groups.add(new SearchShardsGroup(shardId, targetNodes, skip));
175+
groups.add(new SearchShardsGroup(shardIt.shardId(), shardIt.getTargetNodeIds(), shardIt.skip()));
184176
}
185177
return groups;
186178
}

server/src/test/java/org/elasticsearch/action/search/AbstractSearchAsyncActionTests.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -225,7 +225,6 @@ public void testShardNotAvailableWithDisallowPartialFailures() {
225225
// skip one to avoid the "all shards failed" failure.
226226
SearchShardIterator skipIterator = new SearchShardIterator(null, null, Collections.emptyList(), null);
227227
skipIterator.skip(true);
228-
skipIterator.reset();
229228
action.skipShard(skipIterator);
230229
assertThat(exception.get(), instanceOf(SearchPhaseExecutionException.class));
231230
SearchPhaseExecutionException searchPhaseExecutionException = (SearchPhaseExecutionException) exception.get();

server/src/test/java/org/elasticsearch/action/search/SearchAsyncActionTests.java

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -545,7 +545,7 @@ public void testAllowPartialResults() throws InterruptedException {
545545

546546
SearchTransportService transportService = new SearchTransportService(null, null, null);
547547
Map<String, Transport.Connection> lookup = new HashMap<>();
548-
Map<ShardId, Boolean> seenShard = new ConcurrentHashMap<>();
548+
Map<ShardId, AtomicInteger> seenShard = new ConcurrentHashMap<>();
549549
lookup.put(primaryNode.getId(), new MockConnection(primaryNode));
550550
lookup.put(replicaNode.getId(), new MockConnection(replicaNode));
551551
Map<String, AliasFilter> aliasFilters = Collections.singletonMap("_na_", AliasFilter.EMPTY);
@@ -581,17 +581,18 @@ protected void executePhaseOnShard(
581581
Transport.Connection connection,
582582
SearchActionListener<TestSearchPhaseResult> listener
583583
) {
584-
seenShard.computeIfAbsent(shardIt.shardId(), (i) -> {
584+
AtomicInteger retries = seenShard.computeIfAbsent(shardIt.shardId(), (i) -> {
585585
numRequests.incrementAndGet(); // only count this once per shard copy
586-
return Boolean.TRUE;
586+
return new AtomicInteger(0);
587587
});
588+
int numRetries = retries.incrementAndGet();
588589
new Thread(() -> {
589590
TestSearchPhaseResult testSearchPhaseResult = new TestSearchPhaseResult(
590591
new ShardSearchContextId(UUIDs.randomBase64UUID(), contextIdGenerator.incrementAndGet()),
591592
connection.getNode()
592593
);
593594
try {
594-
if (shardIt.remaining() > 0) {
595+
if (numRetries < shardIt.size()) {
595596
numFailReplicas.incrementAndGet();
596597
listener.onFailure(new RuntimeException());
597598
} else {
@@ -643,10 +644,8 @@ public void testSkipUnavailableSearchShards() throws InterruptedException {
643644
);
644645
// Skip all the shards
645646
searchShardIterator.skip(true);
646-
searchShardIterator.reset();
647647
searchShardIterators.add(searchShardIterator);
648648
}
649-
List<SearchShardIterator> shardsIter = searchShardIterators;
650649
Map<String, Transport.Connection> lookup = Map.of(primaryNode.getId(), new MockConnection(primaryNode));
651650

652651
CountDownLatch latch = new CountDownLatch(1);
@@ -665,11 +664,11 @@ public void testSkipUnavailableSearchShards() throws InterruptedException {
665664
null,
666665
request,
667666
responseListener,
668-
shardsIter,
667+
searchShardIterators,
669668
new TransportSearchAction.SearchTimeProvider(0, 0, () -> 0),
670669
ClusterState.EMPTY_STATE,
671670
null,
672-
new ArraySearchPhaseResults<>(shardsIter.size()),
671+
new ArraySearchPhaseResults<>(searchShardIterators.size()),
673672
request.getMaxConcurrentShardRequests(),
674673
SearchResponse.Clusters.EMPTY
675674
) {
@@ -702,7 +701,7 @@ protected void run() {
702701
assertNotNull(searchResponse.get());
703702
assertThat(searchResponse.get().getSkippedShards(), equalTo(numUnavailableSkippedShards));
704703
assertThat(searchResponse.get().getFailedShards(), equalTo(0));
705-
assertThat(searchResponse.get().getSuccessfulShards(), equalTo(shardsIter.size()));
704+
assertThat(searchResponse.get().getSuccessfulShards(), equalTo(searchShardIterators.size()));
706705
}
707706

708707
static List<SearchShardIterator> getShardsIter(
@@ -728,7 +727,6 @@ static List<SearchShardIterator> getShardsIter(
728727
for (int i = 0; i < numShards; i++) {
729728
ArrayList<ShardRouting> started = new ArrayList<>();
730729
ArrayList<ShardRouting> initializing = new ArrayList<>();
731-
ArrayList<ShardRouting> unassigned = new ArrayList<>();
732730

733731
ShardRouting routing = ShardRouting.newUnassigned(
734732
new ShardId(index, i),
@@ -758,8 +756,6 @@ static List<SearchShardIterator> getShardsIter(
758756
} else {
759757
initializing.add(routing);
760758
}
761-
} else {
762-
unassigned.add(routing); // unused yet
763759
}
764760
}
765761
Collections.shuffle(started, random());

0 commit comments

Comments
 (0)