import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
protected AbstractSearchAsyncAction(String name, Logger logger, SearchTransportService searchTransportService,
                                    Function<String, Transport.Connection> nodeIdToConnection,
                                    Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                                    Executor executor, SearchRequest request,
                                    ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts, long startTime,
                                    long clusterStateVersion, SearchTask task, SearchPhaseResults<Result> resultConsumer) {
    super(name, request, shardsIts, logger);
    this.startTime = startTime;
    this.logger = logger;
    this.searchTransportService = searchTransportService;
    this.executor = executor;
    this.request = request;
    this.task = task;
    this.listener = listener;
    this.nodeIdToConnection = nodeIdToConnection;
    this.clusterStateVersion = clusterStateVersion;
    this.concreteIndexBoosts = concreteIndexBoosts;
    this.aliasFilter = aliasFilter;
    this.results = resultConsumer;
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的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.GroupShardsIterator; //导入依赖的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.GroupShardsIterator; //导入依赖的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.GroupShardsIterator; //导入依赖的package包/类
private GroupShardsIterator getShardIterators(WhereClause whereClause,
                                              @Nullable String preference,
                                              ClusterState clusterState) throws IndexNotFoundException {
    String[] routingIndices = concreteIndices;
    if (whereClause.partitions().size() > 0) {
        routingIndices = whereClause.partitions().toArray(new String[whereClause.partitions().size()]);
    }

    Map<String, Set<String>> routingMap = null;
    if (whereClause.clusteredBy().isPresent()) {
        routingMap = indexNameExpressionResolver.resolveSearchRouting(
                clusterState, whereClause.routingValues(), routingIndices);
    }
    return clusterService.operationRouting().searchShards(
            clusterState,
            routingIndices,
            routingMap,
            preference
    );
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
@Nullable
private Routing getRouting(ClusterState state, WhereClause whereClause, String preference, final List<ShardId> missingShards) {
    final Map<String, Map<String, List<Integer>>> locations = new TreeMap<>();
    GroupShardsIterator shardIterators;
    try {
        shardIterators = getShardIterators(whereClause, preference, state);
    } catch (IndexNotFoundException e) {
        return new Routing(locations);
    }

    fillLocationsFromShardIterators(locations, shardIterators, missingShards);

    if (missingShards.isEmpty()) {
        return new Routing(locations);
    } else {
        return null;
    }
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的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.GroupShardsIterator; //导入依赖的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.GroupShardsIterator; //导入依赖的package包/类
InitialSearchPhase(String name, SearchRequest request, GroupShardsIterator shardsIts, Logger logger) {
    super(name);
    this.request = request;
    this.shardsIts = shardsIts;
    this.logger = logger;
    // we need to add 1 for non active partition, since we count it in the total. This means for each shard in the iterator we sum up
    // it's number of active shards but use 1 as the default if no replica of a shard is active at this point.
    // on a per shards level we use shardIt.remaining() to increment the totalOps pointer but add 1 for the current shard result
    // we process hence we add one for the non active partition here.
    this.expectedTotalOps = shardsIts.totalSizeWith1ForEmpty();
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
SearchDfsQueryThenFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
                                   Function<String, Transport.Connection> nodeIdToConnection,
                                   Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                                   SearchPhaseController searchPhaseController, Executor executor, SearchRequest request,
                                   ActionListener<SearchResponse> listener, GroupShardsIterator shardsIts, long startTime,
                                   long clusterStateVersion, SearchTask task) {
    super("dfs", logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor,
            request, listener, shardsIts, startTime, clusterStateVersion, task, new SearchPhaseResults<>(shardsIts.size()));
    this.searchPhaseController = searchPhaseController;
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
SearchQueryThenFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
                                Function<String, Transport.Connection> nodeIdToConnection,
                                Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                                SearchPhaseController searchPhaseController, Executor executor,
                                SearchRequest request, ActionListener<SearchResponse> listener,
                                GroupShardsIterator shardsIts, long startTime, long clusterStateVersion,
                                SearchTask task) {
    super("query", logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor,
        request, listener, shardsIts, startTime, clusterStateVersion, task,
        searchPhaseController.newSearchPhaseResults(request, shardsIts.size()));
    this.searchPhaseController = searchPhaseController;
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
@Override
protected ShardIterator shards(ClusterState state, InternalRequest request) {
    if (request.request().doc() != null && request.request().routing() == null) {
        // artificial document without routing specified, ignore its "id" and use either random shard or according to preference
        GroupShardsIterator groupShardsIter = clusterService.operationRouting().searchShards(state,
                new String[] { request.concreteIndex() }, null, request.request().preference());
        return groupShardsIter.iterator().next();
    }

    return clusterService.operationRouting().getShards(state, request.concreteIndex(), request.request().id(),
            request.request().routing(), request.request().preference());
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
private Set<String> nodeIdsWithIndex(String... indices) {
    ClusterState state = client().admin().cluster().prepareState().execute().actionGet().getState();
    GroupShardsIterator allAssignedShardsGrouped = state.routingTable().allAssignedShardsGrouped(indices, true);
    Set<String> nodes = new HashSet<>();
    for (ShardIterator shardIterator : allAssignedShardsGrouped) {
        for (ShardRouting routing : shardIterator.asUnordered()) {
            if (routing.active()) {
                nodes.add(routing.currentNodeId());
            }

        }
    }
    return nodes;
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
private ShardRouting corruptRandomPrimaryFile(final boolean includePerCommitFiles) throws IOException {
    ClusterState state = client().admin().cluster().prepareState().get().getState();
    Index test = state.metaData().index("test").getIndex();
    GroupShardsIterator shardIterators = state.getRoutingTable().activePrimaryShardsGrouped(new String[]{"test"}, false);
    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> files = new TreeSet<>(); // treeset makes sure iteration order is deterministic
    for (FsInfo.Path info : nodeStatses.getNodes().get(0).getFs()) {
        String path = info.getPath();
        Path file = PathUtils.get(path).resolve("indices").resolve(test.getUUID()).resolve(Integer.toString(shardRouting.getId())).resolve("index");
        if (Files.exists(file)) { // multi data path might only have one path in use
            try (DirectoryStream<Path> stream = Files.newDirectoryStream(file)) {
                for (Path item : stream) {
                    if (Files.isRegularFile(item) && "write.lock".equals(item.getFileName().toString()) == false) {
                        if (includePerCommitFiles || isPerSegmentFile(item.getFileName().toString())) {
                            files.add(item);
                        }
                    }
                }
            }
        }
    }
    pruneOldDeleteGenerations(files);
    CorruptionUtils.corruptFile(random(), files.toArray(new Path[0]));
    return shardRouting;
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
private GroupShardsIterator getShardsIter(String index, int numShards, boolean doReplicas, DiscoveryNode primaryNode,
                                          DiscoveryNode replicaNode) {
    ArrayList<ShardIterator> list = new ArrayList<>();
    for (int i = 0; i < numShards; i++) {
        ArrayList<ShardRouting> started = new ArrayList<>();
        ArrayList<ShardRouting> initializing = new ArrayList<>();
        ArrayList<ShardRouting> unassigned = new ArrayList<>();

        ShardRouting routing = ShardRouting.newUnassigned(new ShardId(new Index(index, "_na_"), i), true,
            RecoverySource.StoreRecoverySource.EMPTY_STORE_INSTANCE, new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "foobar"));
        routing = routing.initialize(primaryNode.getId(), i + "p", 0);
        routing.started();
        started.add(routing);
        if (doReplicas) {
            routing = ShardRouting.newUnassigned(new ShardId(new Index(index, "_na_"), i), false,
                RecoverySource.PeerRecoverySource.INSTANCE, new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "foobar"));
            if (replicaNode != null) {
                routing = routing.initialize(replicaNode.getId(), i + "r", 0);
                if (randomBoolean()) {
                    routing.started();
                    started.add(routing);
                } else {
                    initializing.add(routing);
                }
            } else {
                unassigned.add(routing); // unused yet
            }
        }
        Collections.shuffle(started, random());
        started.addAll(initializing);
        list.add(new PlainShardIterator(new ShardId(new Index(index, "_na_"), i), started));
    }
    return new GroupShardsIterator(list);
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
/**
 * Retrieves the routing for sys.shards
 *
 * This routing contains ALL shards of ALL indices.
 * Any shards that are not yet assigned to a node will have a NEGATIVE shard id (see {@link UnassignedShard}
 */
@Override
public Routing getRouting(WhereClause whereClause, @Nullable String preference) {
    // TODO: filter on whereClause
    Map<String, Map<String, List<Integer>>> locations = new TreeMap<>();
    ClusterState state = service.state();
    String[] concreteIndices = state.metaData().concreteAllIndices();
    GroupShardsIterator groupShardsIterator = state.getRoutingTable().allAssignedShardsGrouped(concreteIndices, true, true);
    for (final ShardIterator shardIt : groupShardsIterator) {
        final ShardRouting shardRouting = shardIt.nextOrNull();
        processShardRouting(locations, shardRouting, shardIt.shardId());
    }
    return new Routing(locations);
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
@Override
protected GroupShardsIterator shards(ClusterState clusterState,
        SearchIntoRequest request, String[] concreteIndices) {
    Map<String, Set<String>> routingMap = clusterState.metaData()
            .resolveSearchRouting(request.routing(), request.indices());
    return clusterService.operationRouting().searchShards(clusterState,
            request.indices(), concreteIndices, routingMap,
            request.preference());
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
@Override
protected GroupShardsIterator shards(ClusterState clusterState, ValidateQueryRequest request, String[] concreteIndices) {
    // Hard-code routing to limit request to a single shard, but still, randomize it...
    Map<String, Set<String>> routingMap = indexNameExpressionResolver.resolveSearchRouting(clusterState, Integer.toString(Randomness.get().nextInt(1000)), request.indices());
    return clusterService.operationRouting().searchShards(clusterState, concreteIndices, routingMap, "_local");
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
@Override
protected GroupShardsIterator shards(ClusterState clusterState, FieldStatsRequest request,
                                     String[] concreteIndices) {
    return clusterService.operationRouting().searchShards(clusterState, concreteIndices, null, null);
}
 

import org.elasticsearch.cluster.routing.GroupShardsIterator; //导入依赖的package包/类
protected int numAssignedShards(String... indices) {
    ClusterState state = client().admin().cluster().prepareState().execute().actionGet().getState();
    GroupShardsIterator allAssignedShardsGrouped = state.routingTable().allAssignedShardsGrouped(indices, true);
    return allAssignedShardsGrouped.size();
}