ElasticSearch 7.x之前选主流程

参数minimum_master_nodes防止脑裂、防止数据丢失的极其重要的参数：
discovery.zen.minimum_master_nodes = (master_eligible_ nodes)/2+1

PUT /_ cluster/settings
{
    persistent": {
         "discovery.zen.minimum_master_nodes": 3
    }
}

//ZenDiscovery.findMaster 函数
private DiscoveryNode findMaster() {
    
    // 1. PING 所有节点，获取各节点保存的集群信息
    List<ZenPing.PingResponse> fullPingResponses = pingAndWait(pingTimeout).toList();
    
    // 2. 由于上面是获取的其他节点的信息，这里需要将本节点加上
    final DiscoveryNode localNode = transportService.getLocalNode();    
    fullPingResponses.add(new ZenPing.PingResponse(localNode, null, this.clusterState()));    
    
    // 3. 若设置了 master_election_ignore_non_masters 则去掉没有 master 资格（node.master: false）的节点
    final List<ZenPing.PingResponse> pingResponses = filterPingResponses(fullPingResponses, masterElectionIgnoreNonMasters, logger);
    
    // 4. 将各节点认为的 master 加入 activeMasters 列表
    List<DiscoveryNode> activeMasters = new ArrayList<>();    
    for (ZenPing.PingResponse pingResponse : pingResponses) {
        // 避免未经其他节点检查就将本节点选为 master
        if (pingResponse.master() != null && !localNode.equals(pingResponse.master())) {           
            activeMasters.add(pingResponse.master());        
        }    
    }  
      
    // 5. 将 PING 到的具有 master 资格的节点加入 masterCandidates 列表作为候选节点
    List<ElectMasterService.MasterCandidate> masterCandidates = new ArrayList<>();    
    for (ZenPing.PingResponse pingResponse : pingResponses) {        
        if (pingResponse.node().isMasterNode()) { // 只选取具有称为Master资格的节点           
            masterCandidates.add(new ElectMasterService.MasterCandidate(pingResponse.node(), pingResponse.getClusterStateVersion()));        
        }    
    }
    
    if (activeMasters.isEmpty()) {
        // 6. 没有活跃的 master
        if (electMaster.hasEnoughCandidates(masterCandidates)) { // 判断候选节点是否符合法定节点数
            // 7. 拥有足够的候选节点，则进行选举
            final ElectMasterService.MasterCandidate winner = electMaster.electMaster(masterCandidates);                   
            return winner.getNode();        
        } else {
            // 8. 无法选举，无法得到 master，返回 null
            return null;        
        }    
    } else {
        assert !activeMasters.contains(localNode) : "local node should never be elected as master when other nodes indicate an active master";        
        // 9. 有活跃的 master，从 activeMasters 中选择
        return electMaster.tieBreakActiveMasters(activeMasters);    
    }
}

补充方法：
// 判断候选节点是否符合法定节点数
public boolean hasEnoughCandidates(Collection<MasterCandidate> candidates) {
   if (candidates.isEmpty()) {
       return false;
   }
   if (minimumMasterNodes < 1) {
       return true;
   }
   assert candidates.stream().map(MasterCandidate::getNode).collect(Collectors.toSet()).size() == candidates.size() :
            "duplicates ahead: " + candidates;
   return candidates.size() >= minimumMasterNodes;
}

1.在没有活跃的 master 时使用，上面第 7 步执行 master 的选举，通过 MasterCandidate::compare 对候选节点进行比较
// ElectMasterService.MasterCandidate::compare
public static int compare(MasterCandidate c1, MasterCandidate c2) {
    // 优先选择集群状态版本最新的节点
    int ret = Long.compare(c2.clusterStateVersion, c1.clusterStateVersion);
    if (ret == 0) {
        ret = compareNodes(c1.getNode(), c2.getNode());
    }
    return ret;
}

// ElectMasterService::compareNodes
private static int compareNodes(DiscoveryNode o1, DiscoveryNode o2) {
    // 集群状态版本相同时优先选择具有 master 资格的节点
    if (o1.isMasterNode() && !o2.isMasterNode()) {
        return -1;
    }
    if (!o1.isMasterNode() && o2.isMasterNode()) {
        return 1;
    }
    // 集群存在多个 master 节点, 选择 id 较小的当做 master
    return o1.getId().compareTo(o2.getId());
}

2.在有活跃的 master 时，上面第 9 步通过 ElectMasterService::tieBreakAcitveMasters 使用 ElectMasterService::compareNodes 的规则从 activeMasters 中选择

// ZenDiscovery.innerJoinCluster函数
private void innerJoinCluster() {
    DiscoveryNode masterNode = null;
    final Thread currentThread = Thread.currentThread();
    nodeJoinController.startElectionContext();
    while (masterNode == null && joinThreadControl.joinThreadActive(currentThread)) {
        masterNode = findMaster();
    }

    if (!joinThreadControl.joinThreadActive(currentThread)) {
        logger.trace("thread is no longer in currentJoinThread. Stopping.");
        return;
    }

    // 选出的临时 master 是本节点，则等待被选举为真正的 master
    if (transportService.getLocalNode().equals(masterNode)) {
        final int requiredJoins = Math.max(0, electMaster.minimumMasterNodes() - 1); // we count as one
        logger.debug("elected as master, waiting for incoming joins ([{}] needed)", requiredJoins);
        nodeJoinController.waitToBeElectedAsMaster(requiredJoins, masterElectionWaitForJoinsTimeout,
                new NodeJoinController.ElectionCallback() {
                    @Override
                    public void onElectedAsMaster(ClusterState state) {
                        // 成功被选举为 master
                        synchronized (stateMutex) {
                            joinThreadControl.markThreadAsDone(currentThread);
                        }
                    }

                    @Override
                    public void onFailure(Throwable t) {
                        logger.trace("failed while waiting for nodes to join, rejoining", t);
                        // 等待超时，重新开始选举流程
                        synchronized (stateMutex) {
                            joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                        }
                    }
                }

        );
    } else {
        // process any incoming joins (they will fail because we are not the master)
        // 选出的临时 master 不是本节点，不再接收其他节点的 join 请求
        nodeJoinController.stopElectionContext(masterNode + " elected");

        // send join request
        // 向临时节点发送join请求（投票），被选举的临时master在确认成为master并发布新的集群状态后才会返回
        final boolean success = joinElectedMaster(masterNode);
        
        // 成功加入之前选择的临时 master 节点，则结束线程，否则重新选举
        synchronized (stateMutex) {
            if (success) {
                DiscoveryNode currentMasterNode = this.clusterState().getNodes().getMasterNode();
                if (currentMasterNode == null) {
                    // Post 1.3.0, the master should publish a new cluster state before acking our join request. we now should have
                    // a valid master.
                    logger.debug("no master node is set, despite of join request completing. retrying pings.");
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                } else if (currentMasterNode.equals(masterNode) == false) {
                    // update cluster state
                    joinThreadControl.stopRunningThreadAndRejoin("master_switched_while_finalizing_join");
                }

                joinThreadControl.markThreadAsDone(currentThread);
            } else {
                // failed to join. Try again...
                joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
            }
        }
    }
}

private synchronized void checkPendingJoinsAndElectIfNeeded() {
    // 计算节点得票数
    final int pendingMasterJoins = electionContext.getPendingMasterJoinsCount();
    if (electionContext.isEnoughPendingJoins(pendingMasterJoins) == false) {
        ...
    } else {
        // 得票数足够，成为 master
        electionContext.closeAndBecomeMaster();
    }
}

public synchronized int getPendingMasterJoinsCount() {
    int pendingMasterJoins = 0;
    // 统计节点得票数，只计算拥有 master 资格节点的投票
    for (DiscoveryNode node : joinRequestAccumulator.keySet()) {
        if (node.isMasterNode()) {
            pendingMasterJoins++;
        }
    }
    return pendingMasterJoins;
}