import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private void sendFailShard(ShardRouting shardRouting, String message, @Nullable Exception failure, ClusterState state) {
    try {
        logger.warn(
            (Supplier<?>) () -> new ParameterizedMessage(
                "[{}] marking and sending shard failed due to [{}]", shardRouting.shardId(), message), failure);
        failedShardsCache.put(shardRouting.shardId(), shardRouting);
        shardStateAction.localShardFailed(shardRouting, message, failure, SHARD_STATE_ACTION_LISTENER, state);
    } catch (Exception inner) {
        if (failure != null) inner.addSuppressed(failure);
        logger.warn(
            (Supplier<?>) () -> new ParameterizedMessage(
                "[{}][{}] failed to mark shard as failed (because of [{}])",
                shardRouting.getIndexName(),
                shardRouting.getId(),
                message),
            inner);
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
@Override
public void shardFailed(ShardRouting failedShard, UnassignedInfo unassignedInfo) {
    if (failedShard.primary() && failedShard.initializing()) {
        RecoverySource recoverySource = failedShard.recoverySource();
        if (recoverySource.getType() == RecoverySource.Type.SNAPSHOT) {
            Snapshot snapshot = ((SnapshotRecoverySource) recoverySource).snapshot();
            // mark restore entry for this shard as failed when it's due to a file corruption. There is no need wait on retries
            // to restore this shard on another node if the snapshot files are corrupt. In case where a node just left or crashed,
            // however, we only want to acknowledge the restore operation once it has been successfully restored on another node.
            if (unassignedInfo.getFailure() != null && Lucene.isCorruptionException(unassignedInfo.getFailure().getCause())) {
                changes(snapshot).failedShards.put(failedShard.shardId(), new ShardRestoreStatus(failedShard.currentNodeId(),
                    RestoreInProgress.State.FAILURE, unassignedInfo.getFailure().getCause().getMessage()));
            }
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
public List<Path> listShardFiles(ShardRouting routing) throws IOException {
    NodesStatsResponse nodeStatses = client().admin().cluster().prepareNodesStats(routing.currentNodeId()).setFs(true).get();
    ClusterState state = client().admin().cluster().prepareState().get().getState();
    final Index test = state.metaData().index("test").getIndex();
    assertThat(routing.toString(), nodeStatses.getNodes().size(), equalTo(1));
    List<Path> files = new ArrayList<>();
    for (FsInfo.Path info : nodeStatses.getNodes().get(0).getFs()) {
        String path = info.getPath();
        Path file = PathUtils.get(path).resolve("indices/" + test.getUUID() + "/" + Integer.toString(routing.getId()) + "/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) {
                    files.add(item);
                }
            }
        }
    }
    return files;
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private void assertFlushResponseEqualsShardStats(ShardStats[] shardsStats, List<ShardsSyncedFlushResult> syncedFlushResults) {

        for (final ShardStats shardStats : shardsStats) {
            for (final ShardsSyncedFlushResult shardResult : syncedFlushResults) {
                if (shardStats.getShardRouting().getId() == shardResult.shardId().getId()) {
                    for (Map.Entry<ShardRouting, SyncedFlushService.ShardSyncedFlushResponse> singleResponse : shardResult.shardResponses().entrySet()) {
                        if (singleResponse.getKey().currentNodeId().equals(shardStats.getShardRouting().currentNodeId())) {
                            if (singleResponse.getValue().success()) {
                                logger.info("{} sync flushed on node {}", singleResponse.getKey().shardId(), singleResponse.getKey().currentNodeId());
                                assertNotNull(shardStats.getCommitStats().getUserData().get(Engine.SYNC_COMMIT_ID));
                            } else {
                                logger.info("{} sync flush failed for on node {}", singleResponse.getKey().shardId(), singleResponse.getKey().currentNodeId());
                                assertNull(shardStats.getCommitStats().getUserData().get(Engine.SYNC_COMMIT_ID));
                            }
                        }
                    }
                }
            }
        }
    }
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
/**
 * Initializes an unassigned shard on a node and removes it from the unassigned
 *
 * @param allocation the allocation
 * @param routingNodes the routing nodes
 * @param routingNode the node to initialize it to
 * @param shardRouting the shard routing that is to be matched in unassigned shards
 * @param unassignedInfo unassigned info to override
 * @param recoverySource recovery source to override
 */
protected void initializeUnassignedShard(RoutingAllocation allocation, RoutingNodes routingNodes, RoutingNode routingNode,
                                         ShardRouting shardRouting, @Nullable UnassignedInfo unassignedInfo,
                                         @Nullable RecoverySource recoverySource) {
    for (RoutingNodes.UnassignedShards.UnassignedIterator it = routingNodes.unassigned().iterator(); it.hasNext(); ) {
        ShardRouting unassigned = it.next();
        if (!unassigned.equalsIgnoringMetaData(shardRouting)) {
            continue;
        }
        if (unassignedInfo != null || recoverySource != null) {
            unassigned = it.updateUnassigned(unassignedInfo != null ? unassignedInfo : unassigned.unassignedInfo(),
                recoverySource != null ? recoverySource : unassigned.recoverySource(), allocation.changes());
        }
        it.initialize(routingNode.nodeId(), null, allocation.clusterInfo().getShardSize(unassigned, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE), allocation.changes());
        return;
    }
    assert false : "shard to initialize not found in list of unassigned shards";
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的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.ShardRouting; //导入依赖的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.ShardRouting; //导入依赖的package包/类
public void start() {
    if (nodeIds.size() == 0) {
        try {
            onCompletion();
        } catch (Throwable e) {
            listener.onFailure(e);
        }
    } else {
        int nodeIndex = -1;
        for (Map.Entry<String, List<ShardRouting>> entry : nodeIds.entrySet()) {
            nodeIndex++;
            DiscoveryNode node = nodes.get(entry.getKey());
            sendNodeRequest(node, entry.getValue(), nodeIndex);
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
@Override
protected ShardStats shardOperation(IndexShardStatsRequest request, ShardRouting shardRouting) {
    IndexService indexService = indicesService.indexServiceSafe(shardRouting.shardId().getIndex());
    IndexShard indexShard = indexService.shardSafe(shardRouting.shardId().id());
    // if we don't have the routing entry yet, we need it stats wise, we treat it as if the shard is not ready yet
    if (indexShard.routingEntry() == null) {
        throw new ShardNotFoundException(indexShard.shardId());
    }

    if (!indexShard.state().equals(IndexShardState.STARTED)) {
        throw new ElasticsearchException(indexShard.shardId().toString() + " state is " + indexShard.state() + ", not started");
    }
    
    CommonStatsFlags flags = new CommonStatsFlags().clear();

    if (request.dl()) {
        flags.set(CommonStatsFlags.Flag.DL);
    }

    return new ShardStats(indexShard.routingEntry(), indexShard.shardPath(), new CommonStats(indexShard, flags), indexShard.commitStats());
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
public void testRecoveryFailsAfterMovingToRelocatedState() throws InterruptedException, IOException {
    final IndexShard shard = newStartedShard(true);
    ShardRouting origRouting = shard.routingEntry();
    assertThat(shard.state(), equalTo(IndexShardState.STARTED));
    ShardRouting inRecoveryRouting = ShardRoutingHelper.relocate(origRouting, "some_node");
    shard.updateRoutingEntry(inRecoveryRouting);
    shard.relocated("simulate mark as relocated");
    assertThat(shard.state(), equalTo(IndexShardState.RELOCATED));
    try {
        shard.updateRoutingEntry(origRouting);
        fail("Expected IndexShardRelocatedException");
    } catch (IndexShardRelocatedException expected) {
    }

    closeShards(shard);
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
public void testFindAnyUnassignedShardToExplain() {
    // find unassigned primary
    ClusterState clusterState = ClusterStateCreationUtils.state("idx", randomBoolean(), ShardRoutingState.UNASSIGNED);
    ClusterAllocationExplainRequest request = new ClusterAllocationExplainRequest();
    ShardRouting shard = findShardToExplain(request, routingAllocation(clusterState));
    assertEquals(clusterState.getRoutingTable().index("idx").shard(0).primaryShard(), shard);

    // find unassigned replica
    clusterState = ClusterStateCreationUtils.state("idx", randomBoolean(), ShardRoutingState.STARTED, ShardRoutingState.UNASSIGNED);
    request = new ClusterAllocationExplainRequest();
    shard = findShardToExplain(request, routingAllocation(clusterState));
    assertEquals(clusterState.getRoutingTable().index("idx").shard(0).replicaShards().get(0), shard);

    // no unassigned shard to explain
    final ClusterState allStartedClusterState = ClusterStateCreationUtils.state("idx", randomBoolean(),
        ShardRoutingState.STARTED, ShardRoutingState.STARTED);
    final ClusterAllocationExplainRequest anyUnassignedShardsRequest = new ClusterAllocationExplainRequest();
    expectThrows(IllegalStateException.class, () ->
        findShardToExplain(anyUnassignedShardsRequest, routingAllocation(allStartedClusterState)));
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
/**
 * Tests that when the nodes with prior copies of the given shard return a THROTTLE decision,
 * then we do not force allocate to that node but instead throttle.
 */
public void testDontForceAllocateOnThrottleDecision() {
    testAllocator.addData(node1, "allocId1", randomBoolean());
    AllocationDeciders deciders = new AllocationDeciders(Settings.EMPTY, Arrays.asList(
        // since we have a NO decision for allocating a shard (because the second decider returns a NO decision),
        // the allocator will see if it can force assign the primary, and in this case,
        // the TestAllocateDecision's decision for force allocating is to THROTTLE (using
        // the default behavior) so despite the other decider's decision to return YES for
        // force allocating the shard, we still THROTTLE due to the decision from TestAllocateDecision
        new TestAllocateDecision(Decision.THROTTLE), getNoDeciderThatAllowsForceAllocate()
    ));
    RoutingAllocation allocation = routingAllocationWithOnePrimaryNoReplicas(deciders, CLUSTER_RECOVERED, "allocId1");
    testAllocator.allocateUnassigned(allocation);
    assertThat(allocation.routingNodesChanged(), equalTo(true));
    List<ShardRouting> ignored = allocation.routingNodes().unassigned().ignored();
    assertEquals(ignored.size(), 1);
    assertEquals(ignored.get(0).unassignedInfo().getLastAllocationStatus(), AllocationStatus.DECIDERS_THROTTLED);
    assertTrue(allocation.routingNodes().shardsWithState(ShardRoutingState.INITIALIZING).isEmpty());
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private Set<ShardRouting> getExpectedReplicas(ShardId shardId, ClusterState state) {
    Set<ShardRouting> expectedReplicas = new HashSet<>();
    String localNodeId = state.nodes().getLocalNodeId();
    if (state.routingTable().hasIndex(shardId.getIndexName())) {
        for (ShardRouting shardRouting : state.routingTable().shardRoutingTable(shardId)) {
            if (shardRouting.unassigned()) {
                continue;
            }
            if (localNodeId.equals(shardRouting.currentNodeId()) == false) {
                expectedReplicas.add(shardRouting);
            }

            if (shardRouting.relocating() && localNodeId.equals(shardRouting.relocatingNodeId()) == false) {
                expectedReplicas.add(shardRouting.getTargetRelocatingShard());
            }
        }
    }
    return expectedReplicas;
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private void performOnReplicas(ReplicaRequest replicaRequest, List<ShardRouting> shards) {
    final String localNodeId = primary.routingEntry().currentNodeId();
    // If the index gets deleted after primary operation, we skip replication
    for (final ShardRouting shard : shards) {
        if (executeOnReplicas == false || shard.unassigned()) {
            if (shard.primary() == false) {
                totalShards.incrementAndGet();
            }
            continue;
        }

        if (shard.currentNodeId().equals(localNodeId) == false) {
            performOnReplica(shard, replicaRequest);
        }

        if (shard.relocating() && shard.relocatingNodeId().equals(localNodeId) == false) {
            performOnReplica(shard.getTargetRelocatingShard(), replicaRequest);
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private Decision decideSameNode(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation,
                                Iterable<ShardRouting> assignedShards) {
    for (ShardRouting assignedShard : assignedShards) {
        if (node.nodeId().equals(assignedShard.currentNodeId())) {
            if (assignedShard.isSameAllocation(shardRouting)) {
                return allocation.decision(Decision.NO, NAME,
                    "the shard cannot be allocated to the node on which it already exists [%s]",
                    shardRouting.toString());
            } else {
                return allocation.decision(Decision.NO, NAME,
                    "the shard cannot be allocated to the same node on which a copy of the shard already exists [%s]",
                    assignedShard.toString());
            }
        }
    }
    return allocation.decision(Decision.YES, NAME, "the shard does not exist on the same node");
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private ClusterState rebalance(ClusterState clusterState) {
    AllocationService strategy = createAllocationService(Settings.builder()
            .build());
    clusterState = strategy.reroute(clusterState, "reroute");
    int numRelocations = 0;
    while (true) {
        List<ShardRouting> initializing = clusterState.routingTable().shardsWithState(INITIALIZING);
        if (initializing.isEmpty()) {
            break;
        }
        logger.debug("Initializing shards: {}", initializing);
        numRelocations += initializing.size();
        clusterState = strategy.applyStartedShards(clusterState, initializing);
    }
    logger.debug("--> num relocations to get balance: {}", numRelocations);
    return clusterState;
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
/**
 * Allocate unassigned shards to nodes (if any) where valid copies of the shard already exist.
 * It is up to the individual implementations of {@link #makeAllocationDecision(ShardRouting, RoutingAllocation, Logger)}
 * to make decisions on assigning shards to nodes.
 *
 * @param allocation the allocation state container object
 */
public void allocateUnassigned(RoutingAllocation allocation) {
    final RoutingNodes routingNodes = allocation.routingNodes();
    final RoutingNodes.UnassignedShards.UnassignedIterator unassignedIterator = routingNodes.unassigned().iterator();
    while (unassignedIterator.hasNext()) {
        final ShardRouting shard = unassignedIterator.next();
        final AllocateUnassignedDecision allocateUnassignedDecision = makeAllocationDecision(shard, allocation, logger);

        if (allocateUnassignedDecision.isDecisionTaken() == false) {
            // no decision was taken by this allocator
            continue;
        }

        if (allocateUnassignedDecision.getAllocationDecision() == AllocationDecision.YES) {
            unassignedIterator.initialize(allocateUnassignedDecision.getTargetNode().getId(),
                allocateUnassignedDecision.getAllocationId(),
                shard.primary() ? ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE :
                                  allocation.clusterInfo().getShardSize(shard, ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE),
                allocation.changes());
        } else {
            unassignedIterator.removeAndIgnore(allocateUnassignedDecision.getAllocationStatus(), allocation.changes());
        }
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
@Override
public void failShardIfNeeded(ShardRouting replica, long primaryTerm, String message, Exception exception, Runnable onSuccess,
                              Consumer<Exception> onPrimaryDemoted, Consumer<Exception> onIgnoredFailure) {
    if (failedReplicas.add(replica) == false) {
        fail("replica [" + replica + "] was failed twice");
    }
    if (opFailures.containsKey(replica)) {
        if (randomBoolean()) {
            onSuccess.run();
        } else {
            onIgnoredFailure.accept(new ElasticsearchException("simulated"));
        }
    } else {
        fail("replica [" + replica + "] was failed");
    }
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的package包/类
private void processShardRouting(Map<String, Map<String, List<Integer>>> locations, ShardRouting shardRouting, ShardId shardId) {
    String node;
    if (shardRouting == null) {
        throw new NoShardAvailableActionException(shardId);
    }
    node = shardRouting.currentNodeId();
    Map<String, List<Integer>> nodeMap = locations.get(node);
    if (nodeMap == null) {
        nodeMap = new TreeMap<>();
        locations.put(shardRouting.currentNodeId(), nodeMap);
    }

    List<Integer> shards = nodeMap.get(shardRouting.getIndex());
    if (shards == null) {
        shards = new ArrayList<>();
        nodeMap.put(shardRouting.getIndex(), shards);
    }
    shards.add(shardRouting.id());
}
 

import org.elasticsearch.cluster.routing.ShardRouting; //导入依赖的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);
        }
    }
}