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;
 }
if (transportService.getLocalNode().equals(masterNode)) {
 final int requiredJoins = Math.max(, 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) {
 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)
 nodeJoinController.stopElectionContext(masterNode + " elected");
// send join request
 final boolean success = joinElectedMaster(masterNode);
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 DiscoveryNode findMaster() {
 logger.trace("starting to ping");
  ① List<ZenPing.PingResponse> fullPingResponses = pingAndWait(pingTimeout).toList();
 
final DiscoveryNode localNode = transportService.getLocalNode();
fullPingResponses.add(new ZenPing.PingResponse(localNode, null, this.clusterState()));
// filter responses
 final List<ZenPing.PingResponse> pingResponses = filterPingResponses(fullPingResponses, masterElectionIgnoreNonMasters, logger);
② List<DiscoveryNode> activeMasters = new ArrayList<>();
 for (ZenPing.PingResponse pingResponse : pingResponses) {
 // We can't include the local node in pingMasters list, otherwise we may up electing ourselves without
 // any check / verifications from other nodes in ZenDiscover#innerJoinCluster()
 if (pingResponse.master() != null && !localNode.equals(pingResponse.master())) {
 activeMasters.add(pingResponse.master());
 }
 }
// nodes discovered during pinging
 ③ List<ElectMasterService.MasterCandidate> masterCandidates = new ArrayList<>();
 for (ZenPing.PingResponse pingResponse : pingResponses) {
 if (pingResponse.node().isMasterNode()) {
 masterCandidates.add(new ElectMasterService.MasterCandidate(pingResponse.node(), pingResponse.getClusterStateVersion()));
 }
 }
if (activeMasters.isEmpty()) {
 if (electMaster.hasEnoughCandidates(masterCandidates)) {
 ④ final ElectMasterService.MasterCandidate winner = electMaster.electMaster(masterCandidates);
 logger.trace("candidate {} won election", winner);
 return winner.getNode();
 } else {
 // if we don't have enough master nodes, we bail, because there are not enough master to elect from
 logger.warn("not enough master nodes discovered during pinging (found [{}], but needed [{}]), pinging again",
 masterCandidates, electMaster.minimumMasterNodes());
 return null;
 }
 } else {
 assert !activeMasters.contains(localNode) : "local node should never be elected as master when other nodes indicate an active master";
 // lets tie break between discovered nodes
 ⑤ return electMaster.tieBreakActiveMasters(activeMasters);
 }
 }

 public MasterCandidate electMaster(Collection<MasterCandidate> candidates) {
 List<MasterCandidate> sortedCandidates = new ArrayList<>(candidates);
 sortedCandidates.sort(MasterCandidate::compare);
 return sortedCandidates.get();
 }

 public static int compare(MasterCandidate c1, MasterCandidate c2) {
 int ret = Long.compare(c2.clusterStateVersion, c1.clusterStateVersion);
 if (ret == ) {
 ret = compareNodes(c1.getNode(), c2.getNode());
 }
 return ret;
 }

protected void ping(final Consumer<PingCollection> resultsConsumer,
 final TimeValue scheduleDuration,
 final TimeValue requestDuration) {
 final List<DiscoveryNode> seedNodes;
 try {
 ① seedNodes = resolveHostsLists(
 unicastZenPingExecutorService,
 logger,
 configuredHosts,
 limitPortCounts,
 transportService,
 UNICAST_NODE_PREFIX,
 resolveTimeout);
 } catch (InterruptedException e) {
 throw new RuntimeException(e);
 }
 seedNodes.addAll(hostsProvider.buildDynamicNodes());
 final DiscoveryNodes nodes = contextProvider.clusterState().nodes();
 // add all possible master nodes that were active in the last known cluster configuration
 for (ObjectCursor<DiscoveryNode> masterNode : nodes.getMasterNodes().values()) {
 seedNodes.add(masterNode.value);
 }
final ConnectionProfile connectionProfile =
 ConnectionProfile.buildSingleChannelProfile(TransportRequestOptions.Type.REG, requestDuration, requestDuration);
 ② final PingingRound pingingRound = new PingingRound(pingingRoundIdGenerator.incrementAndGet(), seedNodes, resultsConsumer,
 nodes.getLocalNode(), connectionProfile);
 activePingingRounds.put(pingingRound.id(), pingingRound);
 final AbstractRunnable pingSender = new AbstractRunnable() {
 @Override
 public void onFailure(Exception e) {
 if (e instanceof AlreadyClosedException == false) {
 logger.warn("unexpected error while pinging", e);
 }
 }
@Override
 protected void doRun() throws Exception {
 sendPings(requestDuration, pingingRound);
 }
 };
 ③ threadPool.generic().execute(pingSender);
 threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3), ThreadPool.Names.GENERIC, pingSender);
 threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3 * 2), ThreadPool.Names.GENERIC, pingSender);
 threadPool.schedule(scheduleDuration, ThreadPool.Names.GENERIC, new AbstractRunnable() {
 @Override
 protected void doRun() throws Exception {
 finishPingingRound(pingingRound);
 }
@Override
 public void onFailure(Exception e) {
 logger.warn("unexpected error while finishing pinging round", e);
 }
 });
 }