List<Class<? extends AllocationDecider>> expectedDeciders = Arrays.asList(
    MaxRetryAllocationDecider.class,
    ReplicaAfterPrimaryActiveAllocationDecider.class,
    RebalanceOnlyWhenActiveAllocationDecider.class,
    ClusterRebalanceAllocationDecider.class,
    ConcurrentRebalanceAllocationDecider.class,
    EnableAllocationDecider.class,
    NodeVersionAllocationDecider.class,
    SnapshotInProgressAllocationDecider.class,
    FilterAllocationDecider.class,
    SameShardAllocationDecider.class,
    DiskThresholdDecider.class,
    ThrottlingAllocationDecider.class,
    ShardsLimitAllocationDecider.class,
    AwarenessAllocationDecider.class);

* canRebalance
* canAllocate  指定 shard 是否可以分配到指定 node
* canRemain
* canForceAllocatePrimary

gatewayAllocator.allocateUnassigned(allocation);
shardsAllocator.allocate(allocation);

primaryShardAllocator.allocateUnassigned(allocation);
replicaShardAllocator.processExistingRecoveries(allocation);
replicaShardAllocator.allocateUnassigned(allocation);

public void allocateUnassigned(RoutingAllocation allocation) {
   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());
       }
   }
}

private void reroute(RoutingAllocation allocation) {
   assert hasDeadNodes(allocation) == false : "dead nodes should be explicitly cleaned up. See deassociateDeadNodes";



   // now allocate all the unassigned to available nodes
   if (allocation.routingNodes().unassigned().size() > 0) {
       removeDelayMarkers(allocation);
       gatewayAllocator.allocateUnassigned(allocation);
   }


   shardsAllocator.allocate(allocation);
   assert RoutingNodes.assertShardStats(allocation.routingNodes());
}

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);
    }
    //...
}

void asyncFetch(final DiscoveryNode[] nodes, long fetchingRound) {
    action.list(shardId, nodes, new ActionListener<BaseNodesResponse<T>>() {
        @Override
        public void onResponse(BaseNodesResponse<T> response) {
        processAsyncFetch(response.getNodes(), response.failures(), fetchingRound);
  }

o.e.action.s.n.TransportNodesAction.AsyncAction#start

reroute(shardId, "post_response");
接着调用到-->
routingService.reroute("async_shard_fetch");

protected void performReroute(final String reason) {
    clusterService.submitStateUpdateTask(CLUSTER_UPDATE_TASK_SOURCE + "(" + reason + ")", new ClusterStateUpdateTask(Priority.HIGH) {
        @Override
        public ClusterState execute(ClusterState currentState) {
            rerouting.set(false);
            RoutingAllocation.Result routingResult = allocationService.reroute(currentState, reason);
            return ClusterState.builder(currentState).routingResult(routingResult).build();
        }
    });
}

/**
* Split the list of node shard states into groups yes/no/throttle based on allocation deciders
*/
private NodesToAllocate buildNodesToAllocate(RoutingAllocation allocation,
                                            List<NodeGatewayStartedShards> nodeShardStates,
                                            ShardRouting shardRouting,
                                            boolean forceAllocate) {
   List<DecidedNode> yesNodeShards = new ArrayList<>();
   List<DecidedNode> throttledNodeShards = new ArrayList<>();
   List<DecidedNode> noNodeShards = new ArrayList<>();
   for (NodeGatewayStartedShards nodeShardState : nodeShardStates) {
       RoutingNode node = allocation.routingNodes().node(nodeShardState.getNode().getId());
       if (node == null) {
           continue;
       }
 
       Decision decision = forceAllocate ? allocation.deciders().canForceAllocatePrimary(shardRouting, node, allocation) :
                                           allocation.deciders().canAllocate(shardRouting, node, allocation);
       DecidedNode decidedNode = new DecidedNode(nodeShardState, decision);
       if (decision.type() == Type.THROTTLE) {
           throttledNodeShards.add(decidedNode);
       } else if (decision.type() == Type.NO) {
           noNodeShards.add(decidedNode);
       } else {
           yesNodeShards.add(decidedNode);
       }
   }
   return new NodesToAllocate(Collections.unmodifiableList(yesNodeShards), Collections.unmodifiableList(throttledNodeShards),
                                 Collections.unmodifiableList(noNodeShards));
}

public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
   if (allocation.shouldIgnoreShardForNode(shardRouting.shardId(), node.nodeId())) {
       return Decision.NO;
   }
   Decision.Multi ret = new Decision.Multi();
   for (AllocationDecider allocationDecider : allocations) {
       Decision decision = allocationDecider.canAllocate(shardRouting, node, allocation);
       // short track if a NO is returned.
       if (decision == Decision.NO) {
           if (logger.isTraceEnabled()) {
               logger.trace("Can not allocate [{}] on node [{}] due to [{}]", shardRouting, node.node(), allocationDecider.getClass().getSimpleName());
           }
           // short circuit only if debugging is not enabled
           if (!allocation.debugDecision()) {
               return decision;
           } else {
               ret.add(decision);
           }
       } else if (decision != Decision.ALWAYS
                   && (allocation.getDebugMode() != EXCLUDE_YES_DECISIONS || decision.type() != Decision.Type.YES)) {
           // the assumption is that a decider that returns the static instance Decision#ALWAYS
           // does not really implements canAllocate
           ret.add(decision);
       }
   }
   return ret;
}

public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
   Decision decision = canAllocate(shardRouting, node, allocation);
   if (decision.type() == Type.NO) {
       // On a NO decision, by default, we allow force allocating the primary.
       return allocation.decision(Decision.YES,
                                 decision.label(),
                                  "primary shard [{}] allowed to force allocate on node [{}]",
                                  shardRouting.shardId(), node.nodeId());
   } else {
       // On a THROTTLE/YES decision, we use the same decision instead of forcing allocation
       return decision;
   }
}

cluster.routing.allocation.balance.index
cluster.routing.allocation.balance.shard
cluster.routing.allocation.balance.threshold

public void allocate(RoutingAllocation allocation) {
   final Balancer balancer = new Balancer(logger, allocation, weightFunction, threshold);
   balancer.allocateUnassigned();
   balancer.moveShards();
   balancer.balance();
}

if (shardRouting.recoverySource().getType() == Type.PEER)  {
    sourceNode = findSourceNodeForPeerRecovery(logger, routingTable, nodes, shardRouting);
}
RecoveryState recoveryState = new RecoveryState(shardRouting, nodes.getLocalNode(), sourceNode);
indicesService.createShard(shardRouting, recoveryState, recoveryTargetService, new RecoveryListener(shardRouting),
                repositoriesService, failedShardHandler);

indexShard.startRecovery(recoveryState, recoveryTargetService, recoveryListener, repositoriesService,
    (type, mapping) -> {
     assert recoveryState.getRecoverySource().getType() == RecoverySource.Type.LOCAL_SHARDS:
         "mapping update consumer only required by local shards recovery";
     try {
         client.admin().indices().preparePutMapping()
             .setConcreteIndex(shardRouting.index()) // concrete index - no name clash, it uses uuid
             .setType(type)
             .setSource(mapping.source().string())
             .get();
     } catch (IOException ex) {
         throw new ElasticsearchException("failed to stringify mapping source", ex);
     }
    }, this);

cancellableThreads.execute(() -> responseHolder.set(
    transportService.submitRequest(request.sourceNode(), PeerRecoverySourceService.Actions.START_RECOVERY, request,
    new FutureTransportResponseHandler<RecoveryResponse>() {
      @Override
      public RecoveryResponse newInstance() {
          return new RecoveryResponse();
      }
    }).txGet()));