java类org.apache.zookeeper.server.quorum.QuorumPeer的实例源码

/**
 * 启动伪集群模式
 * @param config
 * @throws IOException
 * @throws InterruptedException
 * @throws ConfigException
 */
public void startFakeCluster(QuorumPeerConfig config) throws IOException{

    ServerCnxnFactory cnxnFactory = new NIOServerCnxnFactory();
    cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());

    QuorumPeer quorumPeer = new QuorumPeer(config.getServers(), config.getDataDir(), config.getDataLogDir(), config.getElectionAlg(), config.getServerId(), config.getTickTime(), config.getInitLimit(), config.getSyncLimit(), config.getQuorumListenOnAllIPs(), cnxnFactory, config.getQuorumVerifier());
    quorumPeer.setClientAddress(config.getClientPortAddress());
    quorumPeer.setTxnFactory(new FileTxnSnapLog(config.getDataLogDir(), config.getDataDir()));
    quorumPeer.setElectionType(config.getElectionAlg());
    quorumPeer.setMyid(config.getServerId());
    quorumPeer.setTickTime(config.getTickTime());
    quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
    quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
    quorumPeer.setInitLimit(config.getInitLimit());
    quorumPeer.setSyncLimit(config.getSyncLimit());
    quorumPeer.setQuorumVerifier(config.getQuorumVerifier(), true);
    quorumPeer.setCnxnFactory(cnxnFactory);
    quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
    quorumPeer.setLearnerType(config.getPeerType());
    quorumPeer.setSyncEnabled(config.getSyncEnabled());
    quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());

    quorumPeer.start();
    LOGGER.info("ZkServerCluster Started! ClientPortAddress={}", config.getClientPortAddress());
}

private void fakeConfigureServers(int count) throws IOException {
    peers = new HashMap<Long,QuorumServer>();
    qps = new QuorumPeer[count];
    qpsDirs = new File[count];
    for(int i = 1; i <= count; i++) {
        peers.put(Long.valueOf(i), new QuorumServer(
            i, "127.0.0.1", fakeBasePort + i, serverCount + fakeBasePort + i, null));
    }
    StringBuilder sb = new StringBuilder();
    for(int i = 0; i < count; i++) {
        //make that testData exists otherwise it fails on windows
        testData.mkdirs();
        qpsDirs[i] = File.createTempFile("sysTest", ".tmp", testData);
        qpsDirs[i].delete();
        qpsDirs[i].mkdir();
        int port = fakeBasePort+10+i;
        if (sb.length() > 0) {
            sb.append(',');
        }
        sb.append("localhost:");
        sb.append(port);
    }
    serverHostPort = sb.toString();
}

@Test
public void testSocketTimeout() throws Exception {
    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 2000, 2, 2);
    QuorumCnxManager cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }
    int port = peers.get(peer.getId()).electionAddr.getPort();
    LOG.info("Election port: " + port);
    InetSocketAddress addr = new InetSocketAddress(port);
    Thread.sleep(1000);

    Socket sock = new Socket();
    sock.connect(peers.get(new Long(1)).electionAddr, 5000);
    long begin = System.currentTimeMillis();
    // Read without sending data. Verify timeout.
    cnxManager.receiveConnection(sock);
    long end = System.currentTimeMillis();
    if((end - begin) > ((peer.getSyncLimit() * peer.getTickTime()) + 500)) Assert.fail("Waited more than necessary");
}

/**
 * See ZOOKEEPER-1294. Confirms that an observer will not support the quorum
 * of a leader by forming a 5-node, 2-observer ensemble (so quorum size is 2).
 * When all but the leader and one observer are shut down, the leader should
 * enter the 'looking' state, not stay in the 'leading' state.
 */
@Test
public void testLEWithObserver() throws Exception {
    QuorumPeer leader = null;
    for (QuorumPeer server : Arrays.asList(qb.s1, qb.s2, qb.s3)) {
        if (server.getServerState().equals(
                QuorumStats.Provider.FOLLOWING_STATE)) {
            server.shutdown();
            assertTrue("Waiting for server down", ClientBase
                    .waitForServerDown("127.0.0.1:"
                            + server.getClientPort(),
                            ClientBase.CONNECTION_TIMEOUT));
        } else {
            assertNull("More than one leader found", leader);
            leader = server;
        }
    }
    assertTrue("Leader is not in Looking state", ClientBase
            .waitForServerState(leader, ClientBase.CONNECTION_TIMEOUT,
                    QuorumStats.Provider.LOOKING_STATE));
}

/**
 * See ZOOKEEPER-1294. Confirms that an observer will not support the quorum
 * of a leader by forming a 5-node, 2-observer ensemble (so quorum size is 2).
 * When all but the leader and one observer are shut down, the leader should
 * enter the 'looking' state, not stay in the 'leading' state.
 */
@Test
public void testLEWithObserver() throws Exception {
    QuorumPeer leader = null;
    for (QuorumPeer server : Arrays.asList(qb.s1, qb.s2, qb.s3)) {
        if (server.getServerState().equals(
                QuorumStats.Provider.FOLLOWING_STATE)) {
            server.shutdown();
            assertTrue("Waiting for server down", ClientBase
                    .waitForServerDown("127.0.0.1:"
                            + server.getClientPort(),
                            ClientBase.CONNECTION_TIMEOUT));
        } else {
            assertNull("More than one leader found", leader);
            leader = server;
        }
    }
    assertTrue("Leader is not in Looking state", ClientBase
            .waitForServerState(leader, ClientBase.CONNECTION_TIMEOUT,
                    QuorumStats.Provider.LOOKING_STATE));
}

public static boolean waitForServerState(QuorumPeer qp, int timeout,
        String serverState) {
    long start = System.currentTimeMillis();
    while (true) {
        try {
            Thread.sleep(250);
        } catch (InterruptedException e) {
            // ignore
        }
        if (qp.getServerState().equals(serverState))
            return true;
        if (System.currentTimeMillis() > start + timeout) {
            return false;
        }
    }
}

private void shutdownQP(QuorumPeer qp) throws InterruptedException {
    assertNotNull("QuorumPeer doesn't exist!", qp);
    qp.shutdown();

    int retryCnt = 30;
    while (retryCnt > 0) {
        if (qp.getPeerState() == ServerState.LOOKING) {
            LOG.info("Number of retries:{} to change the server state to {}",
                    retryCnt, ServerState.LOOKING);
            break;
        }
        Thread.sleep(500);
        retryCnt--;
    }
    Assert.assertEquals(
            "After shutdown, QuorumPeer should change its state to LOOKING",
            ServerState.LOOKING, qp.getPeerState());
}

/**
 * Walk through the target peer commmittedLog.
 * @param sessionId
 * @param peerId
 */
private void validateRequestLog(long sessionId, int peerId) {
    String session = Long.toHexString(sessionId);
    LOG.info("Searching for txn of session 0x " + session +
            " on peer " + peerId);
    String peerType = peerId == qb.getLeaderIndex() ? "leader" : "follower";
    QuorumPeer peer = qb.getPeerList().get(peerId);
    ZKDatabase db = peer.getActiveServer().getZKDatabase();
    for (Proposal p : db.getCommittedLog()) {
        Assert.assertFalse("Should not see " +
                           TraceFormatter.op2String(p.request.type) +
                           " request from local session 0x" + session +
                           " on the " + peerType,
                           p.request.sessionId == sessionId);
    }
}

@Test
public void testSocketTimeout() throws Exception {
    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 2000, 2, 2);
    QuorumCnxManager cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }
    int port = peers.get(peer.getId()).electionAddr.getPort();
    LOG.info("Election port: " + port);
    Thread.sleep(1000);

    Socket sock = new Socket();
    sock.connect(peers.get(1L).electionAddr, 5000);
    long begin = Time.currentElapsedTime();
    // Read without sending data. Verify timeout.
    cnxManager.receiveConnection(sock);
    long end = Time.currentElapsedTime();
    if((end - begin) > ((peer.getSyncLimit() * peer.getTickTime()) + 500)) Assert.fail("Waited more than necessary");
    cnxManager.halt();
    Assert.assertFalse(cnxManager.listener.isAlive());
}

@Test
public void testSocketTimeout() throws Exception {
    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 2000, 2, 2);
    QuorumCnxManager cnxManager = peer.createCnxnManager();
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }
    int port = peers.get(peer.getId()).electionAddr.getPort();
    LOG.info("Election port: " + port);
    InetSocketAddress addr = new InetSocketAddress(port);
    Thread.sleep(1000);

    Socket sock = new Socket();
    sock.connect(peers.get(new Long(1)).electionAddr, 5000);
    long begin = System.currentTimeMillis();
    // Read without sending data. Verify timeout.
    cnxManager.receiveConnection(sock);
    long end = System.currentTimeMillis();
    if((end - begin) > ((peer.getSyncLimit() * peer.getTickTime()) + 500)) Assert.fail("Waited more than necessary");
}

/**
 * See ZOOKEEPER-1294. Confirms that an observer will not support the quorum
 * of a leader by forming a 5-node, 2-observer ensemble (so quorum size is 2).
 * When all but the leader and one observer are shut down, the leader should
 * enter the 'looking' state, not stay in the 'leading' state.
 */
@Test
public void testLEWithObserver() throws Exception {
    QuorumPeer leader = null;
    for (QuorumPeer server : Arrays.asList(qb.s1, qb.s2, qb.s3)) {
        if (server.getServerState().equals(
                QuorumStats.Provider.FOLLOWING_STATE)) {
            server.shutdown();
            assertTrue("Waiting for server down", ClientBase
                    .waitForServerDown("127.0.0.1:"
                            + server.getClientPort(),
                            ClientBase.CONNECTION_TIMEOUT));
        } else {
            assertNull("More than one leader found", leader);
            leader = server;
        }
    }
    assertTrue("Leader is not in Looking state", ClientBase
            .waitForServerState(leader, ClientBase.CONNECTION_TIMEOUT,
                    QuorumStats.Provider.LOOKING_STATE));
}

private void assertLocalPeerMXBeanAttributes(QuorumPeer qp,
        String beanName, Boolean isPartOfEnsemble) throws Exception {
    Assert.assertEquals("Mismatches LearnerType!", qp.getLearnerType()
            .name(), JMXEnv.ensureBeanAttribute(beanName, "LearnerType"));
    Assert.assertEquals("Mismatches ClientAddress!",
            qp.getClientAddress().getHostString() + ":" + qp.getClientAddress().getPort(),
            JMXEnv.ensureBeanAttribute(beanName, "ClientAddress"));
    Assert.assertEquals("Mismatches LearnerType!",
            qp.getElectionAddress().getHostString() + ":" + qp.getElectionAddress().getPort(),
            JMXEnv.ensureBeanAttribute(beanName, "ElectionAddress"));
    Assert.assertEquals("Mismatches PartOfEnsemble!", isPartOfEnsemble,
            JMXEnv.ensureBeanAttribute(beanName, "PartOfEnsemble"));
    Assert.assertEquals("Mismatches ConfigVersion!", qp.getQuorumVerifier()
            .getVersion(), JMXEnv.ensureBeanAttribute(beanName,
            "ConfigVersion"));
    Assert.assertEquals("Mismatches QuorumSystemInfo!", qp
            .getQuorumVerifier().toString(), JMXEnv.ensureBeanAttribute(
            beanName, "QuorumSystemInfo"));
}

/**
 * Return true if any of the states is achieved
 */
public static boolean waitForServerState(QuorumPeer qp, int timeout,
        String... serverStates) {
    long start = Time.currentElapsedTime();
    while (true) {
        try {
            Thread.sleep(250);
        } catch (InterruptedException e) {
            // ignore
        }
        for (String state : serverStates) {
            if (qp.getServerState().equals(state)) {
                return true;
            }
        }
        if (Time.currentElapsedTime() > start + timeout) {
            return false;
        }
    }
}

private void verifyElectionTimeTakenJMXAttribute(List<QuorumPeer> peers)
        throws Exception {
    LOG.info("Verify QuorumPeer#electionTimeTaken jmx bean attribute");

    for (int i = 1; i <= peers.size(); i++) {
        QuorumPeer qp = peers.get(i - 1);
        if (qp.getLearnerType() == LearnerType.OBSERVER) {
            continue; // Observer don't have electionTimeTaken attribute.
        }
        Long electionTimeTaken = -1L;
        String bean = "";
        if (qp.getPeerState() == ServerState.FOLLOWING) {
            bean = String.format(
                    "%s:name0=ReplicatedServer_id%d,name1=replica.%d,name2=Follower",
                    CommonNames.DOMAIN, i, i);
        } else if (qp.getPeerState() == ServerState.LEADING) {
            bean = String.format(
                    "%s:name0=ReplicatedServer_id%d,name1=replica.%d,name2=Leader",
                    CommonNames.DOMAIN, i, i);
        }
        electionTimeTaken = (Long) JMXEnv.ensureBeanAttribute(bean,
                "ElectionTimeTaken");
        Assert.assertTrue("Wrong electionTimeTaken value!",
                electionTimeTaken >= 0);
    }
}

private void fakeConfigureServers(int count) throws IOException {
    peers = new HashMap<Long,QuorumServer>();
    qps = new QuorumPeer[count];
    qpsDirs = new File[count];
    for(int i = 1; i <= count; i++) {
        peers.put(Long.valueOf(i), new QuorumServer(
            i, "127.0.0.1", fakeBasePort + i, serverCount + fakeBasePort + i, null));
    }
    StringBuilder sb = new StringBuilder();
    for(int i = 0; i < count; i++) {
        //make that testData exists otherwise it fails on windows
        testData.mkdirs();
        qpsDirs[i] = File.createTempFile("sysTest", ".tmp", testData);
        qpsDirs[i].delete();
        qpsDirs[i].mkdir();
        int port = fakeBasePort+10+i;
        if (sb.length() > 0) {
            sb.append(',');
        }
        sb.append("localhost:");
        sb.append(port);
    }
    serverHostPort = sb.toString();
}

void mockServer() throws InterruptedException, IOException {
    /*
     * Create an instance of the connection manager
     */
    QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
    cnxManager = peer.createCnxnManager();
    QuorumCnxManager.Listener listener = cnxManager.listener;
    listener.start();


    cnxManager.toSend(1l, FLETestUtils.createMsg(ServerState.LOOKING.ordinal(), 0, 0, 0));
    cnxManager.recvQueue.take();
    cnxManager.toSend(1L, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 1, 0, 0));
}

public void startServer(int index) throws IOException {
    int port = fakeBasePort+10+index;
    if (fakeMachines) {
        qps[index] = new QuorumPeer(peers, qpsDirs[index], qpsDirs[index], port, 0, index+1, tickTime, initLimit, syncLimit);
        qps[index].start();
    } else {
        try {
            QuorumPeerInstance.startInstance(im, quorumHostPort, index);
        } catch(Exception e) {
            IOException ioe = new IOException(e.getClass().getName() + ": " + e.getMessage());
            ioe.setStackTrace(e.getStackTrace());
            throw ioe;
        }
    }
}

@Before
public void setUp()
throws Exception {
    tmpdir = ClientBase.createTmpDir();
    peers = new HashMap<Long,QuorumServer>();
    for(int i = 0; i < 5; i++) {
        peers.put(Long.valueOf(i),
                  new QuorumServer(Long.valueOf(i), "127.0.0.1", PortAssignment.unique(), 0, null));
    }
    peer = new QuorumPeer(peers,
            tmpdir,
            tmpdir,
            PortAssignment.unique(),
            3, 3, 1000, 2, 2);
}

private QuorumPeer waitForLeader() throws InterruptedException {
    int retryCnt = 0;
    QuorumPeer newLeaderQP = null;
    while (retryCnt < 30) {
        newLeaderQP = getLeaderQuorumPeer(mt);
        if (newLeaderQP != null) {
            LOG.info("Number of retries:{} to findout new Leader",
                    retryCnt);
            break;
        }
        retryCnt--;
        Thread.sleep(500);
    }
    return newLeaderQP;
}

@Test
public void testCnxManager() throws Exception {
    CnxManagerThread thread = new CnxManagerThread();

    thread.start();

    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
    QuorumCnxManager cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }

    cnxManager.toSend(new Long(0), createMsg(ServerState.LOOKING.ordinal(), 1, -1, 1));

    Message m = null;
    int numRetries = 1;
    while((m == null) && (numRetries++ <= THRESHOLD)){
        m = cnxManager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
        if(m == null) cnxManager.connectAll();
    }

    Assert.assertTrue("Exceeded number of retries", numRetries <= THRESHOLD);

    thread.join(5000);
    if (thread.isAlive()) {
        Assert.fail("Thread didn't join");
    } else {
        if(thread.failed)
            Assert.fail("Did not receive expected message");
    }

}

@Test
public void testCnxManagerTimeout() throws Exception {
    Random rand = new Random();
    byte b = (byte) rand.nextInt();
    int finalOctet = b & 0xFF;
    int deadPort = PortAssignment.unique();
    String deadAddress = new String("192.0.2." + finalOctet);

    LOG.info("This is the dead address I'm trying: " + deadAddress);

    peers.put(Long.valueOf(2),
              new QuorumServer(2, deadAddress, deadPort, PortAssignment.unique(), null));
    peerTmpdir[2] = ClientBase.createTmpDir();

    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
    QuorumCnxManager cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }

    long begin = System.currentTimeMillis();
    cnxManager.toSend(new Long(2), createMsg(ServerState.LOOKING.ordinal(), 1, -1, 1));
    long end = System.currentTimeMillis();

    if((end - begin) > 6000) Assert.fail("Waited more than necessary");

}

/**
 * Returns null on success, otw the message assoc with the failure 
 * @throws InterruptedException
 */
public String verifyThreadCount(ArrayList<QuorumPeer> peerList, long ecnt)
    throws InterruptedException
{
    String failure = null;
    for (int i = 0; i < 480; i++) {
        Thread.sleep(500);

        failure = _verifyThreadCount(peerList, ecnt);
        if (failure == null) {
            return null;
        }
    }
    return failure;
}

public String _verifyThreadCount(ArrayList<QuorumPeer> peerList, long ecnt) {
    for (int myid = 0; myid < peerList.size(); myid++) {
        QuorumPeer peer = peerList.get(myid);
        QuorumCnxManager cnxManager = peer.getQuorumCnxManager();
        long cnt = cnxManager.getThreadCount();
        if (cnt != ecnt) {
            return new String(new Date()
                + " Incorrect number of Worker threads for sid=" + myid
                + " expected " + ecnt + " found " + cnt);
        }
    }
    return null;
}

private QuorumPeer getLeaderQuorumPeer(List<MainThread> mtList) {
    for (MainThread mt : mtList) {
        QuorumPeer quorumPeer = mt.getQuorumPeer();
        if (null != quorumPeer
                && ServerState.LEADING == quorumPeer.getPeerState()) {
            return quorumPeer;
        }
    }
    return null;
}

private QuorumPeer getLeaderQuorumPeer(MainThread[] mt) {
    for (int i = mt.length - 1; i >= 0; i--) {
        QuorumPeer quorumPeer = mt[i].getQuorumPeer();
        if (null != quorumPeer
                && ServerState.LEADING == quorumPeer.getPeerState()) {
            return quorumPeer;
        }
    }
    return null;
}

@Test
public void testLostMessage() throws Exception {
    FastLeaderElection le[] = new FastLeaderElection[count];

    LOG.info("TestLE: " + getTestName()+ ", " + count);
    for(int i = 0; i < count; i++) {
        int clientport = PortAssignment.unique();
        peers.put(Long.valueOf(i),
                  new QuorumServer(i, "0.0.0.0", clientport,
                                   PortAssignment.unique(), null));
        tmpdir[i] = ClientBase.createTmpDir();
        port[i] = clientport;
    }

    /*
     * Start server 0
     */

    QuorumPeer peer = new QuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 3, 1, 1000, 2, 2);
    peer.startLeaderElection();
    FLETestUtils.LEThread thread = new FLETestUtils.LEThread(peer, 1);
    thread.start();

    /*
     * Start mock server 1
     */
    mockServer();
    thread.join(5000);
    if (thread.isAlive()) {
        Assert.fail("Threads didn't join");
    }
}

void mockServer() throws InterruptedException, IOException {
    /*
     * Create an instance of the connection manager
     */
    QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
    cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    listener.start();


    cnxManager.toSend(1l, FLETestUtils.createMsg(ServerState.LOOKING.ordinal(), 0, 0, 0));
    cnxManager.recvQueue.take();
    cnxManager.toSend(1L, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 1, 0, 0));
}

private QuorumPeer getLeaderQuorumPeer(MainThread[] mt) {
    for (int i = mt.length - 1; i >= 0; i--) {
        QuorumPeer quorumPeer = mt[i].getQuorumPeer();
        if (null != quorumPeer
                && ServerState.LEADING == quorumPeer.getPeerState()) {
            return quorumPeer;
        }
    }
    return null;
}

/**
 * Test to verify that server is able to reform quorum if the Leader goes
 * down.
 */
@Test(timeout = 30000)
public void testRelectionWithValidCredentials() throws Exception {
    Map<String, String> authConfigs = new HashMap<String, String>();
    authConfigs.put(QuorumAuth.QUORUM_SASL_AUTH_ENABLED, "true");
    authConfigs.put(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED, "true");
    authConfigs.put(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED, "true");

    String connectStr = startQuorum(3, authConfigs, 3, false);
    CountdownWatcher watcher = new CountdownWatcher();
    zk = new ZooKeeper(connectStr, ClientBase.CONNECTION_TIMEOUT, watcher);
    watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
    zk.create("/myTestRoot", new byte[0], Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT_SEQUENTIAL);
    watcher.reset();

    // Shutdown Leader to trigger re-election
    QuorumPeer leaderQP = getLeaderQuorumPeer(mt);
    LOG.info("Shutdown Leader sid:{} to trigger quorum leader-election",
            leaderQP.getId());
    shutdownQP(leaderQP);

    // Wait for quorum formation
    QuorumPeer newLeaderQP = waitForLeader();
    assertNotNull("New leader must have been elected by now", newLeaderQP);
    watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
    zk.create("/myTestRoot", new byte[0], Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT_SEQUENTIAL);
}

public void startThenShutdown(int id) throws IOException {
    PeerStruct ps = getPeer(id);
    LOG.info("Creating QuorumPeer " + ps.id + "; public port " + ps.clientPort);
    ps.peer = new QuorumPeer(peersView, ps.dataDir, ps.dataDir, ps.clientPort, electionAlg,
            ps.id, tickTime, initLimit, syncLimit);
    Assert.assertEquals(ps.clientPort, ps.peer.getClientPort());

    ps.peer.start();
    Assert.assertTrue("Waiting for server up", ClientBase.waitForServerUp("127.0.0.1:"
            + getPeer(id).clientPort, ClientBase.CONNECTION_TIMEOUT));
    shutdown(id);
}