elasticsearch的zenDiscovery和master选举机制

private void innerJoinCluster() {
        DiscoveryNode masterNode = null;
        final Thread currentThread = Thread.currentThread();
　　　　　//一直阻塞直到找到master节点，在集群刚刚启动，或者集群master丢失的情况，这种阻塞能够保证集群一致性
        while (masterNode == null && joinThreadControl.joinThreadActive(currentThread)) {
            masterNode = findMaster();
        }
　　　　　　//有可能自己会被选举为master（集群启动，或者加入时正在选举）
　　　　　　if (clusterService.localNode().equals(masterNode)) {
　　　　　　//如果本身是master，则需要向其它所有节点发送集群状态更新
            clusterService.submitStateUpdateTask("zen-disco-join (elected_as_master)", Priority.IMMEDIATE, new ProcessedClusterStateNonMasterUpdateTask() {
                @Override
                public ClusterState execute(ClusterState currentState) {
　　　　　　　　　　　　//选举时错误的，之前的master状态良好，则不更新状态，仍旧使用之前状态。
                    if (currentState.nodes().masterNode() != null) {
                       return currentState;
                    }
                    DiscoveryNodes.Builder builder = new DiscoveryNodes.Builder(currentState.nodes()).masterNodeId(currentState.nodes().localNode().id());
                    // update the fact that we are the master...
                    ClusterBlocks clusterBlocks = ClusterBlocks.builder().blocks(currentState.blocks()).removeGlobalBlock(discoverySettings.getNoMasterBlock()).build();
                    currentState = ClusterState.builder(currentState).nodes(builder).blocks(clusterBlocks).build();
                    // eagerly run reroute to remove dead nodes from routing table
                    RoutingAllocation.Result result = allocationService.reroute(currentState);
                    return ClusterState.builder(currentState).routingResult(result).build();
                }
                @Override
                public void onFailure(String source, Throwable t) {
                    logger.error("unexpected failure during [{}]", t, source);
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                }
                @Override
                public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
                    if (newState.nodes().localNodeMaster()) {
                        // we only starts nodesFD if we are master (it may be that we received a cluster state while pinging)
                        joinThreadControl.markThreadAsDone(currentThread);
                        nodesFD.updateNodesAndPing(newState); // start the nodes FD
                    } else {
                        // if we're not a master it means another node published a cluster state while we were pinging
                        // make sure we go through another pinging round and actively join it
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                    }
                    sendInitialStateEventIfNeeded();
                    long count = clusterJoinsCounter.incrementAndGet();
                    logger.trace("cluster joins counter set to [{}] (elected as master)", count);
                }
            });
        } else {
            // 找到的节点不是我，试图连接该master
            final boolean success = joinElectedMaster(masterNode);
            // finalize join through the cluster state update thread
            final DiscoveryNode finalMasterNode = masterNode;
            clusterService.submitStateUpdateTask("finalize_join (" + masterNode + ")", new ClusterStateNonMasterUpdateTask() {
                @Override
                public ClusterState execute(ClusterState currentState) throws Exception {
                    if (!success) {
                        // failed to join. Try again...
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                        return currentState;
                    }
                    if (currentState.getNodes().masterNode() == 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);
                        return currentState;
                    }
                    if (!currentState.getNodes().masterNode().equals(finalMasterNode)) {
                        return joinThreadControl.stopRunningThreadAndRejoin(currentState, "master_switched_while_finalizing_join");
                    }
                    // Note: we do not have to start master fault detection here because it's set at {@link #handleNewClusterStateFromMaster }
                    // when the first cluster state arrives.
                    joinThreadControl.markThreadAsDone(currentThread);
                    return currentState;
                }
                @Override
                public void onFailure(String source, @Nullable Throwable t) {
                    logger.error("unexpected error while trying to finalize cluster join", t);
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                }
            });
        }
    }