QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider { public void shutdown() { running = false; x509Util.close(); if (leader != null) { leader.shutdown("quorum Peer shutdown"); } if (follower != null) { follower.shutdown(); } shutdownServerCnxnFactory(); if(udpSocket != null) { udpSocket.close(); } try { adminServer.shutdown(); } catch (AdminServerException e) { LOG.warn("Problem stopping AdminServer", e); } if(getElectionAlg() != null){ this.interrupt(); getElectionAlg().shutdown(); } try { zkDb.close(); } catch (IOException ie) { LOG.warn("Error closing logs ", ie); } } QuorumPeer();  QuorumPeer(Map<Long, QuorumServer> quorumPeers, File dataDir,
            File dataLogDir, int electionType,
            long myid, int tickTime, int initLimit, int syncLimit,
            ServerCnxnFactory cnxnFactory);  QuorumPeer(Map<Long, QuorumServer> quorumPeers, File dataDir,
            File dataLogDir, int electionType,
            long myid, int tickTime, int initLimit, int syncLimit,
            boolean quorumListenOnAllIPs,
            ServerCnxnFactory cnxnFactory,
            QuorumVerifier quorumConfig);  QuorumPeer(Map<Long,QuorumServer> quorumPeers, File snapDir,
            File logDir, int clientPort, int electionAlg,
            long myid, int tickTime, int initLimit, int syncLimit);  QuorumPeer(Map<Long,QuorumServer> quorumPeers, File snapDir,
            File logDir, int clientPort, int electionAlg,
            long myid, int tickTime, int initLimit, int syncLimit,
            QuorumVerifier quorumConfig); LearnerType getLearnerType(); void setLearnerType(LearnerType p); int getQuorumSize(); long getId(); boolean isSslQuorum(); boolean shouldUsePortUnification(); synchronized Vote getCurrentVote(); synchronized void setCurrentVote(Vote v); synchronized void setPeerState(ServerState newState); synchronized void reconfigFlagSet(); synchronized void reconfigFlagClear(); synchronized boolean isReconfigStateChange(); synchronized ServerState getPeerState(); void recreateSocketAddresses(long id); InetSocketAddress getQuorumAddress(); InetSocketAddress getElectionAddress(); InetSocketAddress getClientAddress(); static QuorumPeer testingQuorumPeer(); void initialize(); @Override synchronized void start(); synchronized void stopLeaderElection(); synchronized void startLeaderElection(); long getLastLoggedZxid(); synchronized ZooKeeperServer getActiveServer(); @Override void run(); void shutdown(); Map<Long,QuorumPeer.QuorumServer> getView(); Map<Long,QuorumPeer.QuorumServer> getVotingView(); Map<Long,QuorumPeer.QuorumServer> getObservingView(); synchronized Set<Long> getCurrentAndNextConfigVoters(); boolean viewContains(Long sid); String[] getQuorumPeers(); String getServerState(); void setMyid(long myid); int getTickTime(); void setTickTime(int tickTime); int getMaxClientCnxnsPerHost(); boolean areLocalSessionsEnabled(); void enableLocalSessions(boolean flag); boolean isLocalSessionsUpgradingEnabled(); void enableLocalSessionsUpgrading(boolean flag); int getMinSessionTimeout(); void setMinSessionTimeout(int min); int getMaxSessionTimeout(); void setMaxSessionTimeout(int max); int getInitLimit(); void setInitLimit(int initLimit); int getTick(); QuorumVerifier configFromString(String s); QuorumVerifier getQuorumVerifier(); QuorumVerifier getLastSeenQuorumVerifier(); synchronized void restartLeaderElection(QuorumVerifier qvOLD, QuorumVerifier qvNEW); String getNextDynamicConfigFilename(); void setLastSeenQuorumVerifier(QuorumVerifier qv, boolean writeToDisk); QuorumVerifier setQuorumVerifier(QuorumVerifier qv, boolean writeToDisk); Election getElectionAlg(); int getSyncLimit(); void setSyncLimit(int syncLimit); boolean getSyncEnabled(); void setSyncEnabled(boolean syncEnabled); int getElectionType(); void setElectionType(int electionType); boolean getQuorumListenOnAllIPs(); void setQuorumListenOnAllIPs(boolean quorumListenOnAllIPs); void setCnxnFactory(ServerCnxnFactory cnxnFactory); void setSecureCnxnFactory(ServerCnxnFactory secureCnxnFactory); void setSslQuorum(boolean sslQuorum); void setUsePortUnification(boolean shouldUsePortUnification); void setZooKeeperServer(ZooKeeperServer zks); void closeAllConnections(); int getClientPort(); void setTxnFactory(FileTxnSnapLog factory); FileTxnSnapLog getTxnFactory(); void setZKDatabase(ZKDatabase database); synchronized void initConfigInZKDatabase(); boolean isRunning(); QuorumCnxManager getQuorumCnxManager(); long getCurrentEpoch(); long getAcceptedEpoch(); void setCurrentEpoch(long e); void setAcceptedEpoch(long e); boolean processReconfig(QuorumVerifier qv, Long suggestedLeaderId, Long zxid, boolean restartLE); QuorumCnxManager createCnxnManager(); public long start_fle; static final String FLE_TIME_UNIT; public Follower follower; public Leader leader; public Observer observer; static final String SYNC_ENABLED; static final String CURRENT_EPOCH_FILENAME; static final String ACCEPTED_EPOCH_FILENAME; }
@Test public void testQuorumPeerListendOnSpecifiedClientIP() throws IOException { long myId = 1; File dataDir = ClientBase.createTmpDir(); int clientPort = PortAssignment.unique(); Map<Long, QuorumServer> peersView = new HashMap<Long, QuorumServer>(); InetAddress clientIP = InetAddress.getLoopbackAddress(); peersView.put(Long.valueOf(myId), new QuorumServer(myId, new InetSocketAddress(clientIP, PortAssignment.unique()), new InetSocketAddress(clientIP, PortAssignment.unique()), new InetSocketAddress(clientIP, clientPort), LearnerType.PARTICIPANT)); QuorumPeer peer1 = new QuorumPeer(peersView, dataDir, dataDir, clientPort, electionAlg, myId, tickTime, initLimit, syncLimit); String hostString1 = peer1.cnxnFactory.getLocalAddress().getHostString(); assertEquals(clientIP.getHostAddress(), hostString1); peer1.shutdown(); peersView.clear(); clientPort = PortAssignment.unique(); peersView.put(Long.valueOf(myId), new QuorumServer(myId, new InetSocketAddress(clientIP, PortAssignment.unique()), new InetSocketAddress(clientIP, PortAssignment.unique()), new InetSocketAddress(clientIP, clientPort), LearnerType.PARTICIPANT)); QuorumPeer peer2 = new QuorumPeer(peersView, dataDir, dataDir, clientPort, electionAlg, myId, tickTime, initLimit, syncLimit); String hostString2 = peer2.cnxnFactory.getLocalAddress().getHostString(); assertEquals(clientIP.getHostAddress(), hostString2); peer2.shutdown(); }