zookeeper源码分析之leader选举

痛定思痛。 2022-09-09 14:40 153阅读 0赞

# 设计步骤 #

*  传一个zoo.cfg配置文件
 *  快照数据加载
 *  zoo.cfg配置文件解析
 *  启动服务监听 --> IP:2181 (NIO / Netty)
 *  集群环境，涉及leader选举、数据同步
 *  监控（Prometheus，JMX）

# 核心源码 #

找到入口类，对于zookeeper的leader选举，并不是由客户端触发，而是在启动的时候会触发一次选举，因此可以通过启动脚本zkServer.sh中的运行命令来找到入口类：

可以看到ZOOMAIN对应的类就是QuorumPeerMain，这个就是入口类

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16][]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

# QuorumPeerMain.main() #

public static void main(String[] args) {
        QuorumPeerMain main = new QuorumPeerMain();
        main.initializeAndRun(args); //args是传入的配置文件路径
    }
    
    protected void initializeAndRun(String[] args) throws ConfigException, IOException, AdminServerException {
        QuorumPeerConfig config = new QuorumPeerConfig(); //定义一个config对象，用来存储解析后的配置文件数据
        if (args.length == 1) {
            config.parse(args[0]);
        }
        // 启动一个线程，定时进行日志清理
        DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config.getDataDir(), 
            config.getDataLogDir(), config.getSnapRetainCount(), config.getPurgeInterval());
        purgeMgr.start();
        // 判断是否是集群模式，如果是集群，则调用runFromConfig，否则单机运行
        if (args.length == 1 && config.isDistributed()) {
            runFromConfig(config);
        } else {
            LOG.warn("Either no config or no quorum defined in config, running in standalone mode");
            // there is only server in the quorum -- run as standalone
            ZooKeeperServerMain.main(args);
        }
    }

## runFromConfig ##

基于配置文件来启动zookeeper，整个方法的核心是基于zoo.cfg配置文件的内容去创建QuorumPeer并启动

public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
        try {
            ManagedUtil.registerLog4jMBeans();
        } catch (JMException e) {
            LOG.warn("Unable to register log4j JMX control", e);
        }
    
        LOG.info("Starting quorum peer, myid=" + config.getServerId());
        MetricsProvider metricsProvider;
        try {
            metricsProvider = MetricsProviderBootstrap.startMetricsProvider(
                config.getMetricsProviderClassName(), config.getMetricsProviderConfiguration());
        } catch (MetricsProviderLifeCycleException error) {
            throw new IOException("Cannot boot MetricsProvider " + config.getMetricsProviderClassName(), error);
        }
        try {
            ServerMetrics.metricsProviderInitialized(metricsProvider);
            ServerCnxnFactory cnxnFactory = null;
            ServerCnxnFactory secureCnxnFactory = null;
            if (config.getClientPortAddress() != null) {
                cnxnFactory = ServerCnxnFactory.createFactory();
                cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(), config.getClientPortListenBacklog(), false);
            }
            if (config.getSecureClientPortAddress() != null) { //是否开启SSL
                secureCnxnFactory = ServerCnxnFactory.createFactory();
                secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(), config.getClientPortListenBacklog(), true);
            }
    
            quorumPeer = getQuorumPeer();
            quorumPeer.setTxnFactory(new FileTxnSnapLog(config.getDataLogDir(), config.getDataDir()));
            quorumPeer.enableLocalSessions(config.areLocalSessionsEnabled());
            quorumPeer.enableLocalSessionsUpgrading(config.isLocalSessionsUpgradingEnabled());
            //quorumPeer.setQuorumPeers(config.getAllMembers());
            quorumPeer.setElectionType(config.getElectionAlg()); //选举算法类型
            quorumPeer.setMyid(config.getServerId()); //服务id
            quorumPeer.setTickTime(config.getTickTime()); //心跳单位时间
            quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
            quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
            quorumPeer.setInitLimit(config.getInitLimit()); //默认初始化限制10个心跳单位时间
            quorumPeer.setSyncLimit(config.getSyncLimit()); //默认同步限制5个心跳单位时间
            quorumPeer.setConnectToLearnerMasterLimit(config.getConnectToLearnerMasterLimit());
            quorumPeer.setObserverMasterPort(config.getObserverMasterPort());
            quorumPeer.setConfigFileName(config.getConfigFilename());
            quorumPeer.setClientPortListenBacklog(config.getClientPortListenBacklog());
            quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
            quorumPeer.setQuorumVerifier(config.getQuorumVerifier(), false);
            if (config.getLastSeenQuorumVerifier() != null) {
                quorumPeer.setLastSeenQuorumVerifier(config.getLastSeenQuorumVerifier(), false);
            }
            quorumPeer.initConfigInZKDatabase();
            quorumPeer.setCnxnFactory(cnxnFactory);
            quorumPeer.setSecureCnxnFactory(secureCnxnFactory);
            quorumPeer.setSslQuorum(config.isSslQuorum());
            quorumPeer.setUsePortUnification(config.shouldUsePortUnification());
            quorumPeer.setLearnerType(config.getPeerType());
            quorumPeer.setSyncEnabled(config.getSyncEnabled());
            quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
            if (config.sslQuorumReloadCertFiles) {
                quorumPeer.getX509Util().enableCertFileReloading();
            }
            quorumPeer.setMultiAddressEnabled(config.isMultiAddressEnabled());
            quorumPeer.setMultiAddressReachabilityCheckEnabled(config.isMultiAddressReachabilityCheckEnabled());
            quorumPeer.setMultiAddressReachabilityCheckTimeoutMs(config.getMultiAddressReachabilityCheckTimeoutMs());
    
            // sets quorum sasl authentication configurations
            quorumPeer.setQuorumSaslEnabled(config.quorumEnableSasl);
            if (quorumPeer.isQuorumSaslAuthEnabled()) {
                quorumPeer.setQuorumServerSaslRequired(config.quorumServerRequireSasl);
                quorumPeer.setQuorumLearnerSaslRequired(config.quorumLearnerRequireSasl);
                quorumPeer.setQuorumServicePrincipal(config.quorumServicePrincipal);
                quorumPeer.setQuorumServerLoginContext(config.quorumServerLoginContext);
                quorumPeer.setQuorumLearnerLoginContext(config.quorumLearnerLoginContext);
            }
            quorumPeer.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);
            quorumPeer.initialize();
    
            if (config.jvmPauseMonitorToRun) {
                quorumPeer.setJvmPauseMonitor(new JvmPauseMonitor(config));
            }
    
            quorumPeer.start(); //启动主线程
            ZKAuditProvider.addZKStartStopAuditLog();
            quorumPeer.join(); //阻塞
        } catch (InterruptedException e) {
            // warn, but generally this is ok
            LOG.warn("Quorum Peer interrupted", e);
        } finally {
            if (metricsProvider != null) {
                try {
                    metricsProvider.stop();
                } catch (Throwable error) {
                    LOG.warn("Error while stopping metrics", error);
                }
            }
        }
    }

## QuorumPeer.start() ##

QuorumPeer是一个线程，重写了Thread的start方法，在线程启动前做两件事：恢复快照数据，启动zkServer

@Override
    public synchronized void start() {
        if (!getView().containsKey(myid)) {
            throw new RuntimeException("My id " + myid + " not in the peer list");
        }
        loadDataBase(); //恢复快照数据
        startServerCnxnFactory(); //启动zkServer，用户可以通过2181进行通信
        try {
            adminServer.start();
        } catch (AdminServerException e) {
            LOG.warn("Problem starting AdminServer", e);
            System.out.println(e);
        }
        startLeaderElection(); //开启leader选举
        startJvmPauseMonitor(); //监控JVM暂停的时间信息
        super.start();
    }

### startLeaderElection开启leader选举 ###

public synchronized void startLeaderElection() {
        try {
            if (getPeerState() == ServerState.LOOKING) {
                // 节点状态为LOOKING时发起投票，创建投票票据，票据包括serverId、zxid、currentEpoch当前选举轮数
                currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
            }
        } catch (IOException e) {
            RuntimeException re = new RuntimeException(e.getMessage());
            re.setStackTrace(e.getStackTrace());
            throw re;
        }
        this.electionAlg = createElectionAlgorithm(electionType);
    }

### createElectionAlgorithm ###

根据配置文件配置的选举算法类型创建选举算法，新版zookeeper默认FastLeaderElection

protected Election createElectionAlgorithm(int electionAlgorithm) {
        Election le = null;
        //TODO: use a factory rather than a switch
        switch (electionAlgorithm) {
        case 1:
            throw new UnsupportedOperationException("Election Algorithm 1 is not supported.");
        case 2:
            throw new UnsupportedOperationException("Election Algorithm 2 is not supported.");
        case 3:
            QuorumCnxManager qcm = createCnxnManager(); //负责管理leader选举中的网络连接
            QuorumCnxManager oldQcm = qcmRef.getAndSet(qcm);
            if (oldQcm != null) { //表示已经在选举了，停止选举
                LOG.warn("Clobbering already-set QuorumCnxManager (restarting leader election?)");
                oldQcm.halt(); //停止选举
            }
            QuorumCnxManager.Listener listener = qcm.listener;
            if (listener != null) {
                listener.start(); //设置监听，此监听和选举有关，监听投票信息
                FastLeaderElection fle = new FastLeaderElection(this, qcm);
                fle.start(); //启动FastLeaderElection
                le = fle;
            } else {
                LOG.error("Null listener when initializing cnx manager");
            }
            break;
        default:
            assert false;
        }
        return le;
    }

### FastLeaderElection ###

public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
        this.stop = false;
        this.manager = manager;
        starter(self, manager);
    }
    
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;
    
        sendqueue = new LinkedBlockingQueue<ToSend>(); //构建一个发送阻塞队列
        recvqueue = new LinkedBlockingQueue<Notification>(); //构建一个接收阻塞队列
        this.messenger = new Messenger(manager);
    }

### Messenger ###

构建两个工作线程，一个负责发送消息，一个接收消息，当FastLeaderElection启动时两个工作线程启动

Messenger(QuorumCnxManager manager) {
        this.ws = new WorkerSender(manager); //构建一个发送消息的工作线程
        this.wsThread = new Thread(this.ws, "WorkerSender[myid=" + self.getId() + "]");
        this.wsThread.setDaemon(true);
    
        this.wr = new WorkerReceiver(manager); //构建一个接收消息的工作线程
        this.wrThread = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
        this.wrThread.setDaemon(true);
    }

## ServerCnxnFactory ##

默认使用的是NIOServerCnxnFactory

### configure ###

@Override
    public void configure(InetSocketAddress addr, int maxcc, int backlog, boolean secure) throws IOException {
        if (secure) {
            throw new UnsupportedOperationException("SSL isn't supported in NIOServerCnxn");
        }
        configureSaslLogin();
    
        maxClientCnxns = maxcc;
        initMaxCnxns();
        sessionlessCnxnTimeout = Integer.getInteger(ZOOKEEPER_NIO_SESSIONLESS_CNXN_TIMEOUT, 10000);
        cnxnExpiryQueue = new ExpiryQueue<NIOServerCnxn>(sessionlessCnxnTimeout);
        expirerThread = new ConnectionExpirerThread();
        // 获取JVM核心数
        int numCores = Runtime.getRuntime().availableProcessors();
        // 32 cores sweet spot seems to be 4 selector threads
        numSelectorThreads = Integer.getInteger(ZOOKEEPER_NIO_NUM_SELECTOR_THREADS,
            // 32/2=16, 16=2的4次方，4与1之间最大值是4，所以返回4
            Math.max((int) Math.sqrt((float) numCores / 2), 1));
        if (numSelectorThreads < 1) {
            throw new IOException("numSelectorThreads must be at least 1");
        }
    
        numWorkerThreads = Integer.getInteger(ZOOKEEPER_NIO_NUM_WORKER_THREADS, 2 * numCores);
        workerShutdownTimeoutMS = Long.getLong(ZOOKEEPER_NIO_SHUTDOWN_TIMEOUT, 5000);
    
        for (int i = 0; i < numSelectorThreads; ++i) {
            selectorThreads.add(new SelectorThread(i));
        }
    
        listenBacklog = backlog;
        this.ss = ServerSocketChannel.open();
        ss.socket().setReuseAddress(true);
        LOG.info("binding to port {}", addr);
        if (listenBacklog == -1) {
            ss.socket().bind(addr);
        } else {
            ss.socket().bind(addr, listenBacklog);
        }
        ss.configureBlocking(false); //设置连接非阻塞
        // 构建Acceptor去处理客户端连接，并register到Selector中
        acceptThread = new AcceptThread(ss, addr, selectorThreads);
    }

### startServerCnxnFactory ###

public void start() {
        stopped = false;
        if (workerPool == null) {
            workerPool = new WorkerService("NIOWorker", numWorkerThreads, false);
        }
        for (SelectorThread thread : selectorThreads) {
            if (thread.getState() == Thread.State.NEW) {
                thread.start(); //所有的subSelector启动，负责I/O读写事件
            }
        }
        // ensure thread is started once and only once
        if (acceptThread.getState() == Thread.State.NEW) {
            acceptThread.start(); //Acceptor启动，处理客户端连接
        }
        if (expirerThread.getState() == Thread.State.NEW) {
            expirerThread.start();
        }
    }

# Leader选举流程 #

### 流程图 ###

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

选举的触发是从QuorumPeer.start()方法的super.start()触发的

### QuorumPeer.run() ###

@Override
    public void run() {
        updateThreadName();
        // 省略JMX监控代码
        try {
            /*
             * Main loop
             */
            while (running) {
                switch (getPeerState()) { //根据选举状态进行不同的处理
                case LOOKING:
                    LOG.info("LOOKING");
                    ServerMetrics.getMetrics().LOOKING_COUNT.add(1);
    
                    if (Boolean.getBoolean("readonlymode.enabled")) {
                        // Create read-only server but don't start it immediately
                        // 省略只读模式server的处理逻辑
                    } else {
                        try {
                            reconfigFlagClear();
                            if (shuttingDownLE) {
                                shuttingDownLE = false;
                                startLeaderElection();
                            }
                            //设置当前投票，根据选举算法开启新一轮选举
                            setCurrentVote(makeLEStrategy().lookForLeader());
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception", e);
                            setPeerState(ServerState.LOOKING);
                        }
                    }
                    break;
                case OBSERVING: //如果是OBSERVING节点，则只监听leader节点数据同步即可
                    try {
                        LOG.info("OBSERVING");
                        setObserver(makeObserver(logFactory));
                        observer.observeLeader();
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception", e);
                    } finally {
                        observer.shutdown();
                        setObserver(null);
                        updateServerState();
    
                        // Add delay jitter before we switch to LOOKING
                        // state to reduce the load of ObserverMaster
                        if (isRunning()) {
                            Observer.waitForObserverElectionDelay();
                        }
                    }
                    break;
                case FOLLOWING: //如果是FOLLOWING节点，直接查找leader节点进行follow
                    try {
                        LOG.info("FOLLOWING");
                        setFollower(makeFollower(logFactory));
                        follower.followLeader();
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception", e);
                    } finally {
                        follower.shutdown();
                        setFollower(null);
                        updateServerState();
                    }
                    break;
                case LEADING:
                    LOG.info("LEADING");
                    try {
                        setLeader(makeLeader(logFactory));
                        leader.lead();
                        setLeader(null);
                    } catch (Exception e) {
                        LOG.warn("Unexpected exception", e);
                    } finally {
                        if (leader != null) {
                            leader.shutdown("Forcing shutdown");
                            setLeader(null);
                        }
                        updateServerState();
                    }
                    break;
                }
            }
        } finally {
            // 省略
        }
    }

### FastLeaderElection.lookForLeader() ###

/**
     * Starts a new round of leader election. Whenever our QuorumPeer
     * changes its state to LOOKING, this method is invoked, and it
     * sends notifications to all other peers.
     */
    public Vote lookForLeader() throws InterruptedException {
        self.start_fle = Time.currentElapsedTime();
        try {
            // 存储当前选举接收到的票据集合
            Map<Long, Vote> recvset = new HashMap<Long, Vote>();
            // 存储上一次leader选举的票据集合
            Map<Long, Vote> outofelection = new HashMap<Long, Vote>();
            int notTimeout = minNotificationInterval;
            synchronized (this) {
                logicalclock.incrementAndGet(); //逻辑时钟自增，用来判断票据是否在同一轮选举
                //更新提案：存储serverId, 最新的zxid， 选举轮数/周期
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }
    
            LOG.info("New election. My id = {}, proposed zxid=0x{}", self.getId(), Long.toHexString(proposedZxid));
            sendNotifications(); //当投票发生变化时异步发送通知给所有竞选节点
    
            SyncedLearnerTracker voteSet;
    
            // 不断循环，直到找到leader
            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
                //Remove next notification from queue, times out after 2 times the termination time
                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
    
                if (n == null) {
                    if (manager.haveDelivered()) {//判断如果所有的票据都已经发布，那么再次发送投票
                        sendNotifications();
                    } else {
                        manager.connectAll();//尝试和每个server建立连接
                    }
    
                    // Exponential backoff，2倍速衰减重试
                    int tmpTimeOut = notTimeout * 2;
                    notTimeout = Math.min(tmpTimeOut, maxNotificationInterval);
                    LOG.info("Notification time out: {}", notTimeout);
                } else if (validVoter(n.sid) && validVoter(n.leader)) { //验证节点serverId是否有效
                    //判断收到的投票者的状态，如果是LOOKING则代表在找leader
                    switch (n.state) {
                    case LOOKING:
                        if (getInitLastLoggedZxid() == -1) {//忽略zxid=-1的通知
                            LOG.debug("Ignoring notification as our zxid is -1");
                            break;
                        }
                        if (n.zxid == -1) {
                            LOG.debug("Ignoring notification from member with -1 zxid {}", n.sid);
                            break;
                        }
                        // If notification > current, replace and send messages out
                        if (n.electionEpoch > logicalclock.get()) {
                            logicalclock.set(n.electionEpoch);
                            recvset.clear();
                            //收到的票据和当前节点投票信息进行比较，比较epoch、zxid、serverId
                            if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                //更新当前节点的票据为收到的票据信息
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                            } else {//自己的票据信息为最新，更新为当前节点投票信息
                                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
                            }
                            sendNotifications();//再次发送通知给其他节点，说我已经赞同了提案中的节点为leader
                        } else if (n.electionEpoch < logicalclock.get()) {
                            //如果收到的投票轮数<当前节点逻辑时钟，代表收到的投票已过期，直接丢弃
                            break;
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                            //如果epoch相同，那么比较zxid和serverId，如果对方胜出，则更新收到的票据信息到当前节点提案中并发送通知
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            sendNotifications();
                        }
    
                        // 将收到的投票信息放入集合中，最终按照“过半原则”判断
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                        //将当前节点票据信息也放入，最终返回所有票据信息
                        voteSet = getVoteTracker(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch));
    
                        if (voteSet.hasAllQuorums()) {//判断是否收到了所有的投票ack确认
                            // 在超时时间（200ms）内等待，看是否还有投票，如果有再次比较票据信息
                            while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
                                if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                                    recvqueue.put(n);
                                    break;
                                }
                            }
    
                            //如果在超时时间（200ms）内没有收到新的投票，则投票结束
                            if (n == null) {
                                //设置当前节点状态：判断leader节点是否自己，如果是直接更新为LEADING；否则，更具节点特性更新为FOLLOWING或者OBSERVING
                                setPeerState(proposedLeader, voteSet);
                                //最终的投票结果
                                Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
                                leaveInstance(endVote);//清空收到的票据队列
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: {}", n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING: //如果收到的票据节点状态已经是LEADING
                        //判断leader的选举epoch和当前节点是否同一个周期
                        if (n.electionEpoch == logicalclock.get()) {
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
                            voteSet = getVoteTracker(recvset, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
                            if (voteSet.hasAllQuorums() && checkLeader(recvset, n.leader, n.electionEpoch)) {
                                setPeerState(n.leader, voteSet);
                                Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }
    
                        /*
                         * Before joining an established ensemble, verify that
                         * a majority are following the same leader.
                         *
                         * Note that the outofelection map also stores votes from the current leader election.
                         * See ZOOKEEPER-1732 for more information.
                         */
                        outofelection.put(n.sid, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
                        voteSet = getVoteTracker(outofelection, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
    
                        if (voteSet.hasAllQuorums() && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                            synchronized (this) {
                                logicalclock.set(n.electionEpoch);
                                setPeerState(n.leader, voteSet);
                            }
                            Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                            leaveInstance(endVote);
                            return endVote;
                        }
                        break;
                    default:
                        LOG.warn("Notification state unrecoginized: {} (n.state), {}(n.sid)", n.state, n.sid);
                        break;
                    }
                } else {
                    if (!validVoter(n.leader)) {
                        LOG.warn("Ignoring notification for non-cluster member sid {} from sid {}", n.leader, n.sid);
                    }
                    if (!validVoter(n.sid)) {
                        LOG.warn("Ignoring notification for sid {} from non-quorum member sid {}", n.leader, n.sid);
                    }
                }
            }
            return null;
        } finally {
            try {
                if (self.jmxLeaderElectionBean != null) {
                    MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
                }
            } catch (Exception e) {
                LOG.warn("Failed to unregister with JMX", e);
            }
            self.jmxLeaderElectionBean = null;
            LOG.debug("Number of connection processing threads: {}", manager.getConnectionThreadCount());
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/0e4f0f22301d47a7b5562728032cbaa4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/8ea8964c66d645fe9dd28b10138ff88f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/d8741362fa69483d98534d7ea8520144.png