import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private void performPhaseOnShard(final int shardIndex, final ShardIterator shardIt, final ShardRouting shard) {
    if (shard == null) {
        // TODO upgrade this to an assert...
        // no more active shards... (we should not really get here, but just for safety)
        onShardFailure(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
    } else {
        try {
            executePhaseOnShard(shardIt, shard, new ActionListener<FirstResult>() {
                @Override
                public void onResponse(FirstResult result) {
                    onShardResult(shardIndex, shard.currentNodeId(), result, shardIt);
                }

                @Override
                public void onFailure(Exception t) {
                    onShardFailure(shardIndex, shard, shard.currentNodeId(), shardIt, t);
                }
            });
        } catch (ConnectTransportException | IllegalArgumentException ex) {
            // we are getting the connection early here so we might run into nodes that are not connected. in that case we move on to
            // the next shard. previously when using discovery nodes here we had a special case for null when a node was not connected
            // at all which is not not needed anymore.
            onShardFailure(shardIndex, shard, shard.currentNodeId(), shardIt, ex);
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private void onShardResult(int shardIndex, String nodeId, FirstResult result, ShardIterator shardIt) {
    result.shardTarget(new SearchShardTarget(nodeId, shardIt.shardId()));
    onShardSuccess(shardIndex, result);
    // we need to increment successful ops first before we compare the exit condition otherwise if we
    // are fast we could concurrently update totalOps but then preempt one of the threads which can
    // cause the successor to read a wrong value from successfulOps if second phase is very fast ie. count etc.
    // increment all the "future" shards to update the total ops since we some may work and some may not...
    // and when that happens, we break on total ops, so we must maintain them
    final int xTotalOps = totalOps.addAndGet(shardIt.remaining() + 1);
    if (xTotalOps == expectedTotalOps) {
        onPhaseDone();
    } else if (xTotalOps > expectedTotalOps) {
        throw new AssertionError("unexpected higher total ops [" + xTotalOps + "] compared to expected ["
            + expectedTotalOps + "]");
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public void start() {
    if (shardsIts.size() == 0) {
        // no shards
        try {
            listener.onResponse(newResponse(request, new AtomicReferenceArray(0), clusterState));
        } catch (Exception e) {
            listener.onFailure(e);
        }
        return;
    }
    // count the local operations, and perform the non local ones
    int shardIndex = -1;
    for (final ShardIterator shardIt : shardsIts) {
        shardIndex++;
        final ShardRouting shard = shardIt.nextOrNull();
        if (shard != null) {
            performOperation(shardIt, shard, shardIndex);
        } else {
            // really, no shards active in this group
            onOperation(null, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private Set<Path> getTranslogDirs(String indexName) throws IOException {
    ClusterState state = client().admin().cluster().prepareState().get().getState();
    GroupShardsIterator shardIterators = state.getRoutingTable().activePrimaryShardsGrouped(new String[]{indexName}, false);
    final Index idx = state.metaData().index(indexName).getIndex();
    List<ShardIterator> iterators = iterableAsArrayList(shardIterators);
    ShardIterator shardIterator = RandomPicks.randomFrom(random(), iterators);
    ShardRouting shardRouting = shardIterator.nextOrNull();
    assertNotNull(shardRouting);
    assertTrue(shardRouting.primary());
    assertTrue(shardRouting.assignedToNode());
    String nodeId = shardRouting.currentNodeId();
    NodesStatsResponse nodeStatses = client().admin().cluster().prepareNodesStats(nodeId).setFs(true).get();
    Set<Path> translogDirs = new TreeSet<>(); // treeset makes sure iteration order is deterministic
    for (FsInfo.Path fsPath : nodeStatses.getNodes().get(0).getFs()) {
        String path = fsPath.getPath();
        final String relativeDataLocationPath =  "indices/"+ idx.getUUID() +"/" + Integer.toString(shardRouting.getId()) + "/translog";
        Path translogPath = PathUtils.get(path).resolve(relativeDataLocationPath);
        if (Files.isDirectory(translogPath)) {
            translogDirs.add(translogPath);
        }
    }
    return translogDirs;
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public void testIterator1() {
    MetaData metaData = MetaData.builder()
            .put(IndexMetaData.builder("test1").settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(2))
            .build();
    RoutingTable routingTable = RoutingTable.builder()
            .addAsNew(metaData.index("test1"))
            .build();

    ShardIterator shardIterator = routingTable.index("test1").shard(0).shardsIt(0);
    assertThat(shardIterator.size(), equalTo(3));
    ShardRouting shardRouting1 = shardIterator.nextOrNull();
    assertThat(shardRouting1, notNullValue());
    assertThat(shardIterator.remaining(), equalTo(2));
    ShardRouting shardRouting2 = shardIterator.nextOrNull();
    assertThat(shardRouting2, notNullValue());
    assertThat(shardIterator.remaining(), equalTo(1));
    assertThat(shardRouting2, not(sameInstance(shardRouting1)));
    ShardRouting shardRouting3 = shardIterator.nextOrNull();
    assertThat(shardRouting3, notNullValue());
    assertThat(shardRouting3, not(sameInstance(shardRouting1)));
    assertThat(shardRouting3, not(sameInstance(shardRouting2)));
    assertThat(shardIterator.nextOrNull(), nullValue());
    assertThat(shardIterator.remaining(), equalTo(0));
    assertThat(shardIterator.nextOrNull(), nullValue());
    assertThat(shardIterator.remaining(), equalTo(0));
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public void testRandomRouting() {
    MetaData metaData = MetaData.builder()
            .put(IndexMetaData.builder("test1").settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(1))
            .put(IndexMetaData.builder("test2").settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(1))
            .build();

    RoutingTable routingTable = RoutingTable.builder()
            .addAsNew(metaData.index("test1"))
            .addAsNew(metaData.index("test2"))
            .build();

    ShardIterator shardIterator = routingTable.index("test1").shard(0).shardsRandomIt();
    ShardRouting shardRouting1 = shardIterator.nextOrNull();
    assertThat(shardRouting1, notNullValue());
    assertThat(shardIterator.nextOrNull(), notNullValue());
    assertThat(shardIterator.nextOrNull(), nullValue());

    shardIterator = routingTable.index("test1").shard(0).shardsRandomIt();
    ShardRouting shardRouting2 = shardIterator.nextOrNull();
    assertThat(shardRouting2, notNullValue());
    ShardRouting shardRouting3 = shardIterator.nextOrNull();
    assertThat(shardRouting3, notNullValue());
    assertThat(shardIterator.nextOrNull(), nullValue());
    assertThat(shardRouting1, not(sameInstance(shardRouting2)));
    assertThat(shardRouting1, sameInstance(shardRouting3));
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private List<ShardStateAction.ShardEntry> createExistingShards(ClusterState currentState, String reason) {
    List<ShardRouting> shards = new ArrayList<>();
    GroupShardsIterator shardGroups =
        currentState.routingTable().allAssignedShardsGrouped(new String[] { INDEX }, true);
    for (ShardIterator shardIt : shardGroups) {
        for (ShardRouting shard : shardIt.asUnordered()) {
            shards.add(shard);
        }
    }
    List<ShardRouting> failures = randomSubsetOf(randomIntBetween(1, 1 + shards.size() / 4), shards.toArray(new ShardRouting[0]));
    String indexUUID = metaData.index(INDEX).getIndexUUID();
    int numberOfTasks = randomIntBetween(failures.size(), 2 * failures.size());
    List<ShardRouting> shardsToFail = new ArrayList<>(numberOfTasks);
    for (int i = 0; i < numberOfTasks; i++) {
        shardsToFail.add(randomFrom(failures));
    }
    return toTasks(currentState, shardsToFail, indexUUID, reason);
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
@Override
public Routing getRouting(WhereClause whereClause, @Nullable String preference) {
    Map<String, Map<String, List<Integer>>> locations = new TreeMap<>();
    GroupShardsIterator shardIterators = clusterService.operationRouting().searchShards(
            clusterService.state(),
            new String[]{index},
            null,
            preference
    );
    ShardRouting shardRouting;
    for (ShardIterator shardIterator : shardIterators) {
        shardRouting = shardIterator.nextOrNull();
        processShardRouting(locations, shardRouting, shardIterator.shardId());
    }

    return new Routing(locations);
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private void fillLocationsFromShardIterators(Map<String, Map<String, List<Integer>>> locations,
                                             GroupShardsIterator shardIterators,
                                             List<ShardId> missingShards) {
    ShardRouting shardRouting;
    for (ShardIterator shardIterator : shardIterators) {
        shardRouting = shardIterator.nextOrNull();
        if (shardRouting != null) {
            if (shardRouting.active()) {
                processShardRouting(locations, shardRouting);
            } else {
                missingShards.add(shardIterator.shardId());
            }
        } else {
            if (isPartitioned) {
                // if the table is partitioned maybe a index/shard just got newly created ...
                missingShards.add(shardIterator.shardId());
            } else {
                throw new UnavailableShardsException(shardIterator.shardId());
            }
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public void start() {
    if (expectedSuccessfulOps == 0) {
        // no search shards to search on, bail with empty response (it happens with search across _all with no indices around and consistent with broadcast operations)
        listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, buildTookInMillis(), ShardSearchFailure.EMPTY_ARRAY));
        return;
    }
    int shardIndex = -1;
    for (final ShardIterator shardIt : shardsIts) {
        shardIndex++;
        final ShardRouting shard = shardIt.nextOrNull();
        if (shard != null) {
            performFirstPhase(shardIndex, shardIt, shard);
        } else {
            // really, no shards active in this group
            onFirstPhaseResult(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
void performFirstPhase(final int shardIndex, final ShardIterator shardIt, final ShardRouting shard) {
    if (shard == null) {
        // no more active shards... (we should not really get here, but just for safety)
        onFirstPhaseResult(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
    } else {
        final DiscoveryNode node = nodes.get(shard.currentNodeId());
        if (node == null) {
            onFirstPhaseResult(shardIndex, shard, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
        } else {
            String[] filteringAliases = indexNameExpressionResolver.filteringAliases(clusterState, shard.index(), request.indices());
            sendExecuteFirstPhase(node, internalSearchRequest(shard, shardsIts.size(), request, filteringAliases, startTime()), new ActionListener<FirstResult>() {
                @Override
                public void onResponse(FirstResult result) {
                    onFirstPhaseResult(shardIndex, shard, result, shardIt);
                }

                @Override
                public void onFailure(Throwable t) {
                    onFirstPhaseResult(shardIndex, shard, node.id(), shardIt, t);
                }
            });
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
void onFirstPhaseResult(int shardIndex, ShardRouting shard, FirstResult result, ShardIterator shardIt) {
    result.shardTarget(new SearchShardTarget(shard.currentNodeId(), shard.index(), shard.id()));
    processFirstPhaseResult(shardIndex, result);
    // we need to increment successful ops first before we compare the exit condition otherwise if we
    // are fast we could concurrently update totalOps but then preempt one of the threads which can
    // cause the successor to read a wrong value from successfulOps if second phase is very fast ie. count etc.
    successfulOps.incrementAndGet();
    // increment all the "future" shards to update the total ops since we some may work and some may not...
    // and when that happens, we break on total ops, so we must maintain them
    final int xTotalOps = totalOps.addAndGet(shardIt.remaining() + 1);
    if (xTotalOps == expectedTotalOps) {
        try {
            innerMoveToSecondPhase();
        } catch (Throwable e) {
            if (logger.isDebugEnabled()) {
                logger.debug(shardIt.shardId() + ": Failed to execute [" + request + "] while moving to second phase", e);
            }
            raiseEarlyFailure(new ReduceSearchPhaseException(firstPhaseName(), "", e, buildShardFailures()));
        }
    } else if (xTotalOps > expectedTotalOps) {
        raiseEarlyFailure(new IllegalStateException("unexpected higher total ops [" + xTotalOps + "] compared to expected [" + expectedTotalOps + "]"));
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public void start() {
    if (shardsIts.size() == 0) {
        // no shards
        try {
            listener.onResponse(newResponse(request, new AtomicReferenceArray(0), clusterState));
        } catch (Throwable e) {
            listener.onFailure(e);
        }
        return;
    }
    // count the local operations, and perform the non local ones
    int shardIndex = -1;
    for (final ShardIterator shardIt : shardsIts) {
        shardIndex++;
        final ShardRouting shard = shardIt.nextOrNull();
        if (shard != null) {
            performOperation(shardIt, shard, shardIndex);
        } else {
            // really, no shards active in this group
            onOperation(null, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
private int findReplicaLevel(ShardIterator shardIt) {
    int replicaLevel = 0;
    shardIt.reset();
    ShardRouting shard;
    while ((shard = shardIt.nextOrNull()) != null) {
        if (shard.unassigned()) {
            continue;
        }
        boolean doOnlyOnRelocating = false;
        if (shard.primary()) {
            if (shard.relocating()) {
                doOnlyOnRelocating = true;
            } else {
                continue;
            }
        }
        String nodeId = !doOnlyOnRelocating ? shard.currentNodeId() : shard.relocating() ? shard.relocatingNodeId() : null;
        if (nodeId == null) {
            continue;
        }
        replicaLevel++;
    }
    return replicaLevel;
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
@Override
protected void masterOperation(final ClusterSearchShardsRequest request, final ClusterState state,
                               final ActionListener<ClusterSearchShardsResponse> listener) {
    ClusterState clusterState = clusterService.state();
    String[] concreteIndices = indexNameExpressionResolver.concreteIndexNames(clusterState, request);
    Map<String, Set<String>> routingMap = indexNameExpressionResolver.resolveSearchRouting(state, request.routing(), request.indices());
    Map<String, AliasFilter> indicesAndFilters = new HashMap<>();
    for (String index : concreteIndices) {
        AliasFilter aliasFilter = indicesService.buildAliasFilter(clusterState, index, request.indices());
        indicesAndFilters.put(index, aliasFilter);
    }

    Set<String> nodeIds = new HashSet<>();
    GroupShardsIterator groupShardsIterator = clusterService.operationRouting().searchShards(clusterState, concreteIndices,
            routingMap, request.preference());
    ShardRouting shard;
    ClusterSearchShardsGroup[] groupResponses = new ClusterSearchShardsGroup[groupShardsIterator.size()];
    int currentGroup = 0;
    for (ShardIterator shardIt : groupShardsIterator) {
        ShardId shardId = shardIt.shardId();
        ShardRouting[] shardRoutings = new ShardRouting[shardIt.size()];
        int currentShard = 0;
        shardIt.reset();
        while ((shard = shardIt.nextOrNull()) != null) {
            shardRoutings[currentShard++] = shard;
            nodeIds.add(shard.currentNodeId());
        }
        groupResponses[currentGroup++] = new ClusterSearchShardsGroup(shardId, shardRoutings);
    }
    DiscoveryNode[] nodes = new DiscoveryNode[nodeIds.size()];
    int currentNode = 0;
    for (String nodeId : nodeIds) {
        nodes[currentNode++] = clusterState.getNodes().get(nodeId);
    }
    listener.onResponse(new ClusterSearchShardsResponse(groupResponses, nodes, indicesAndFilters));
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
public final ShardSearchTransportRequest buildShardSearchRequest(ShardIterator shardIt, ShardRouting shard) {
    AliasFilter filter = aliasFilter.get(shard.index().getUUID());
    assert filter != null;
    float indexBoost = concreteIndexBoosts.getOrDefault(shard.index().getUUID(), DEFAULT_INDEX_BOOST);
    return new ShardSearchTransportRequest(request, shardIt.shardId(), getNumShards(),
        filter, indexBoost, startTime);
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
@Override
public final void run() throws IOException {
    int shardIndex = -1;
    for (final ShardIterator shardIt : shardsIts) {
        shardIndex++;
        final ShardRouting shard = shardIt.nextOrNull();
        if (shard != null) {
            performPhaseOnShard(shardIndex, shardIt, shard);
        } else {
            // really, no shards active in this group
            onShardFailure(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
Function<String, Transport.Connection> processRemoteShards(Map<String, ClusterSearchShardsResponse> searchShardsResponses,
                                                                   List<ShardIterator> remoteShardIterators,
                                                                   Map<String, AliasFilter> aliasFilterMap) {
    Map<String, Supplier<Transport.Connection>> nodeToCluster = new HashMap<>();
    for (Map.Entry<String, ClusterSearchShardsResponse> entry : searchShardsResponses.entrySet()) {
        String clusterName = entry.getKey();
        ClusterSearchShardsResponse searchShardsResponse = entry.getValue();
        for (DiscoveryNode remoteNode : searchShardsResponse.getNodes()) {
            nodeToCluster.put(remoteNode.getId(), () -> getConnection(remoteNode, clusterName));
        }
        Map<String, AliasFilter> indicesAndFilters = searchShardsResponse.getIndicesAndFilters();
        for (ClusterSearchShardsGroup clusterSearchShardsGroup : searchShardsResponse.getGroups()) {
            //add the cluster name to the remote index names for indices disambiguation
            //this ends up in the hits returned with the search response
            ShardId shardId = clusterSearchShardsGroup.getShardId();
            Index remoteIndex = shardId.getIndex();
            Index index = new Index(clusterName + REMOTE_CLUSTER_INDEX_SEPARATOR + remoteIndex.getName(), remoteIndex.getUUID());
            ShardIterator shardIterator = new PlainShardIterator(new ShardId(index, shardId.getId()),
                Arrays.asList(clusterSearchShardsGroup.getShards()));
            remoteShardIterators.add(shardIterator);
            AliasFilter aliasFilter;
            if (indicesAndFilters == null) {
                aliasFilter = new AliasFilter(null, Strings.EMPTY_ARRAY);
            } else {
                aliasFilter = indicesAndFilters.get(shardId.getIndexName());
                assert aliasFilter != null;
            }
            // here we have to map the filters to the UUID since from now on we use the uuid for the lookup
            aliasFilterMap.put(remoteIndex.getUUID(), aliasFilter);
        }
    }
    return (nodeId) -> {
        Supplier<Transport.Connection> supplier = nodeToCluster.get(nodeId);
        if (supplier == null) {
            throw new IllegalArgumentException("unknown remote node: " + nodeId);
        }
        return supplier.get();
    };
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
@Override
protected void doExecute(Task task, SearchRequest searchRequest, ActionListener<SearchResponse> listener) {
    // pure paranoia if time goes backwards we are at least positive
    final long startTimeInMillis = Math.max(0, System.currentTimeMillis());
    final String[] localIndices;
    final Map<String, List<String>> remoteClusterIndices;
    final ClusterState clusterState = clusterService.state();
    if (remoteClusterService.isCrossClusterSearchEnabled()) {
        remoteClusterIndices = remoteClusterService.groupClusterIndices( searchRequest.indices(), // empty string is not allowed
            idx -> indexNameExpressionResolver.hasIndexOrAlias(idx, clusterState));
        List<String> remove = remoteClusterIndices.remove(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY);
        localIndices = remove == null ? Strings.EMPTY_ARRAY : remove.toArray(new String[remove.size()]);
    } else {
        remoteClusterIndices = Collections.emptyMap();
        localIndices = searchRequest.indices();
    }

    if (remoteClusterIndices.isEmpty()) {
        executeSearch((SearchTask)task, startTimeInMillis, searchRequest, localIndices, Collections.emptyList(),
            (nodeId) -> null, clusterState, Collections.emptyMap(), listener);
    } else {
        remoteClusterService.collectSearchShards(searchRequest, remoteClusterIndices,
            ActionListener.wrap((searchShardsResponses) -> {
                List<ShardIterator> remoteShardIterators = new ArrayList<>();
                Map<String, AliasFilter> remoteAliasFilters = new HashMap<>();
                Function<String, Transport.Connection> connectionFunction = remoteClusterService.processRemoteShards(
                    searchShardsResponses, remoteShardIterators, remoteAliasFilters);
                executeSearch((SearchTask)task, startTimeInMillis, searchRequest, localIndices, remoteShardIterators,
                    connectionFunction, clusterState, remoteAliasFilters, listener);
            }, listener::onFailure));
    }
}
 

import org.elasticsearch.cluster.routing.ShardIterator; //导入依赖的package包/类
@SuppressWarnings({"unchecked"})
void onOperation(@Nullable ShardRouting shard, final ShardIterator shardIt, int shardIndex, Exception e) {
    // we set the shard failure always, even if its the first in the replication group, and the next one
    // will work (it will just override it...)
    setFailure(shardIt, shardIndex, e);
    ShardRouting nextShard = shardIt.nextOrNull();
    if (nextShard != null) {
        if (e != null) {
            if (logger.isTraceEnabled()) {
                if (!TransportActions.isShardNotAvailableException(e)) {
                    logger.trace(
                        (org.apache.logging.log4j.util.Supplier<?>)
                            () -> new ParameterizedMessage(
                                "{}: failed to execute [{}]",
                                shard != null ? shard.shortSummary() : shardIt.shardId(),
                                request),
                        e);
                }
            }
        }
        performOperation(shardIt, nextShard, shardIndex);
    } else {
        if (logger.isDebugEnabled()) {
            if (e != null) {
                if (!TransportActions.isShardNotAvailableException(e)) {
                    logger.debug(
                        (org.apache.logging.log4j.util.Supplier<?>)
                            () -> new ParameterizedMessage(
                                "{}: failed to execute [{}]",
                                shard != null ? shard.shortSummary() : shardIt.shardId(),
                                request),
                        e);
                }
            }
        }
        if (expectedOps == counterOps.incrementAndGet()) {
            finishHim();
        }
    }
}