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

@Test
public void testLERestart() throws Exception {

    FastLeaderElection le[] = new FastLeaderElection[count];
    leaderDies = true;
    boolean allowOneBadLeader = leaderDies;

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

    for(int i = 0; i < count; i++) {
        QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 1000, 2, 2);
        peer.startLeaderElection();
        FLERestartThread thread = new FLERestartThread(peer, i);
        thread.start();
        restartThreads.add(thread);
    }
    LOG.info("Started threads " + getTestName());
    for(int i = 0; i < restartThreads.size(); i++) {
        restartThreads.get(i).join(10000);
        if (restartThreads.get(i).isAlive()) {
            Assert.fail("Threads didn't join");
        }

    }
}

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++) {
        InetSocketAddress peerAddress = new InetSocketAddress("127.0.0.1",
                fakeBasePort + i);
        InetSocketAddress electionAddr = new InetSocketAddress("127.0.0.1",
                serverCount + fakeBasePort + i);
        peers.put(Long.valueOf(i), new QuorumServer(i, peerAddress,
                electionAddr));
    }
    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();
}

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);
}

void mockServer() throws InterruptedException, IOException {
    QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
    cnxManager = new QuorumCnxManager(peer);
    cnxManager.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 QuorumZooKeeperServer getConnectedServer(long sessionId) {
    for (QuorumPeer peer : getPeerList()) {
        if (peer.getActiveServer().getSessionTracker().isTrackingSession(sessionId)) {
            return (QuorumZooKeeperServer)peer.getActiveServer();
        }
    }
    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;
}

/**
 * When we create ephemeral node, we need to check against global
 * session, so the leader never accept request from an expired session
 * (that we no longer track)
 *
 * This is not the same as SessionInvalidationTest since session
 * is not in closing state
 */
public void testCreateEphemeral(boolean localSessionEnabled) throws Exception {
    if (localSessionEnabled) {
        qu.enableLocalSession(true);
    }
    qu.startAll();
    QuorumPeer leader = qu.getLeaderQuorumPeer();

    ZooKeeper zk = ClientBase.createZKClient(qu.getConnectString(leader));

    CreateRequest createRequest = new CreateRequest("/impossible",
            new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL.toFlag());
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
    createRequest.serialize(boa, "request");
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());

    // Mimic sessionId generated by follower's local session tracker
    long sid = qu.getFollowerQuorumPeers().get(0).getActiveServer()
            .getServerId();
    long fakeSessionId = (sid << 56) + 1;

    LOG.info("Fake session Id: " + Long.toHexString(fakeSessionId));

    Request request = new Request(null, fakeSessionId, 0, OpCode.create,
            bb, new ArrayList<Id>());

    // Submit request directly to leader
    leader.getActiveServer().submitRequest(request);

    // Make sure that previous request is finished
    zk.create("/ok", new byte[0], Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    Stat stat = zk.exists("/impossible", null);
    Assert.assertEquals("Node from fake session get created", null, stat);

}

/**
 * When local session is enabled, leader will allow persistent node
 * to be create for unknown session
 */
@Test
public void testCreatePersistent() throws Exception {
    qu.enableLocalSession(true);
    qu.startAll();

    QuorumPeer leader = qu.getLeaderQuorumPeer();

    ZooKeeper zk = ClientBase.createZKClient(qu.getConnectString(leader));

    CreateRequest createRequest = new CreateRequest("/success",
            new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT.toFlag());
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
    createRequest.serialize(boa, "request");
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());

    // Mimic sessionId generated by follower's local session tracker
    long sid = qu.getFollowerQuorumPeers().get(0).getActiveServer()
            .getServerId();
    long locallSession = (sid << 56) + 1;

    LOG.info("Local session Id: " + Long.toHexString(locallSession));

    Request request = new Request(null, locallSession, 0, OpCode.create,
            bb, new ArrayList<Id>());

    // Submit request directly to leader
    leader.getActiveServer().submitRequest(request);

    // Make sure that previous request is finished
    zk.create("/ok", new byte[0], Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    Stat stat = zk.exists("/success", null);
    Assert.assertTrue("Request from local sesson failed", stat != null);

}

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

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

    for(int i = 0; i < le.length; i++) {
        QuorumHierarchical hq = new QuorumHierarchical(qp);
        QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 1000, 2, 2, hq);
        peer.startLeaderElection();
        LEThread thread = new LEThread(peer, i);
        thread.start();
        threads.add(thread);
    }
    LOG.info("Started threads " + getTestName());

    for(int i = 0; i < threads.size(); i++) {
        threads.get(i).join(15000);
        if (threads.get(i).isAlive()) {
            Assert.fail("Threads didn't join");
        } else {
            if(threads.get(i).fail)
                Assert.fail("Elected zero-weight server");
        }
    }
}

public void run(){
    try {
        QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[0], peerTmpdir[0], peerClientPort[0], 3, 0, 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 sid = 1;
        cnxManager.toSend(sid, createMsg(ServerState.LOOKING.ordinal(), 0, -1, 1));

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

        if(numRetries > THRESHOLD){
            failed = true;
            return;
        }

        cnxManager.testInitiateConnection(sid);

        m = cnxManager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
        if(m == null){
            failed = true;
            return;
        }
    } catch (Exception e) {
        LOG.error("Exception while running mock thread", e);
        Assert.fail("Unexpected exception");
    }
}