java类org.apache.zookeeper.server.persistence.FileTxnSnapLog的实例源码

protected QuorumZooKeeperServer(FileTxnSnapLog logFactory, int tickTime,
        int minSessionTimeout, int maxSessionTimeout,
        ZKDatabase zkDb, QuorumPeer self)
{
    super(logFactory, tickTime, minSessionTimeout, maxSessionTimeout, zkDb);
    this.self = self;
}

/**
 * Test verifies that the Leader should authenticate the connecting learner
 * quorumpeer. After the successful authentication it should add this
 * learner to the learnerHandler list.
 */
@Test(timeout = 30000)
public void testAuthLearnerConnectsToServerWithAuthRequired()
        throws Exception {
    File testDataLearner = ClientBase.createTmpDir();
    File tmpDir = File.createTempFile("test", ".dir", testDataLearner);
    tmpDir.delete();
    FileTxnSnapLog ftsl = new FileTxnSnapLog(tmpDir, tmpDir);
    QuorumPeer learnerPeer = createQuorumPeer(tmpDir, true, true, true,
            "QuorumLearner", "QuorumServer",
            QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL_DEFAULT_VALUE);
    SimpleLearner sl = new SimpleLearner(ftsl, learnerPeer);

    File testDataLeader = ClientBase.createTmpDir();
    tmpDir = File.createTempFile("test", ".dir", testDataLeader);
    tmpDir.delete();
    tmpDir.mkdir();
    Leader leader = null;
    QuorumPeer peer = createQuorumPeer(tmpDir, true, true, true, "QuorumLearner",
            "QuorumServer",
            QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL_DEFAULT_VALUE);
    CountDownLatch learnerLatch = new CountDownLatch(1);
    leader = createSimpleLeader(tmpDir, peer, learnerLatch);
    peer.leader = leader;

    startLearnerCnxAcceptorThread(leader);
    LOG.info("Start establishing a connection with the Leader");
    String hostname = getLeaderHostname(peer);
    sl.connectToLeader(peer.getQuorumAddress(), hostname);
    // wait till leader socket soTimeout period
    Assert.assertTrue("Leader should accept the auth learner connection",
            learnerLatch.await(leader.self.tickTime * leader.self.initLimit + 1000,
                    TimeUnit.MILLISECONDS));
    Assert.assertEquals("Failed to added the learner", 1,
            leader.getLearners().size());
    ClientBase.recursiveDelete(testDataLearner);
    ClientBase.recursiveDelete(testDataLeader);
}

/**
 * Tests that the ZooKeeper server will fail to start if the
 * transaction log directory is read only.
 *
 * This test will fail if it is executed as root user.
 */
@Test(timeout = 30000)
public void testReadOnlyTxnLogDir() throws Exception {
    ClientBase.setupTestEnv();
    final int CLIENT_PORT = PortAssignment.unique();

    // Start up the ZK server to automatically create the necessary directories
    // and capture the directory where data is stored
    MainThread main = new MainThread(CLIENT_PORT, true);
    File tmpDir = main.tmpDir;
    main.start();
    Assert.assertTrue("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));
    main.shutdown();

    // Make the transaction log directory read only
    File logDir = new File(main.logDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
    logDir.setWritable(false);

    // Restart ZK and observe a failure
    main = new MainThread(CLIENT_PORT, false, tmpDir);
    main.start();

    Assert.assertFalse("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));

    main.shutdown();

    logDir.setWritable(true);

    main.deleteDirs();
}

@Override
protected Follower makeFollower(FileTxnSnapLog logFactory) throws IOException {

    return new Follower(this, new FollowerZooKeeperServer(logFactory, this, this.getZkDb())) {
        @Override
        protected void processPacket(QuorumPacket qp) throws Exception {
            if (stopPing && qp.getType() == Leader.PING) {
                LOG.info("Follower skipped ping");
                throw new SocketException("Socket time out while sending the ping response");
            } else {
                super.processPacket(qp);
            }
        }
    };
}

private LeaderZooKeeperServer prepareLeader(File tmpDir, QuorumPeer peer)
        throws IOException, NoSuchFieldException, IllegalAccessException {
    FileTxnSnapLog logFactory = new FileTxnSnapLog(tmpDir, tmpDir);
    peer.setTxnFactory(logFactory);
    ZKDatabase zkDb = new ZKDatabase(logFactory);
    LeaderZooKeeperServer zk = new LeaderZooKeeperServer(logFactory, peer, zkDb);
    return zk;
}

private ConversableFollower createFollower(File tmpDir, QuorumPeer peer)
throws IOException {
    FileTxnSnapLog logFactory = new FileTxnSnapLog(tmpDir, tmpDir);
    peer.setTxnFactory(logFactory);
    ZKDatabase zkDb = new ZKDatabase(logFactory);
    FollowerZooKeeperServer zk = new FollowerZooKeeperServer(logFactory, peer, zkDb);
    peer.setZKDatabase(zkDb);
    return new ConversableFollower(peer, zk);
}

/**
 * Tests that the ZooKeeper server will fail to start if the
 * snapshot directory is read only.
 *
 * This test will fail if it is executed as root user.
 */
@Test(timeout = 30000)
public void testReadOnlySnapshotDir() throws Exception {
    ClientBase.setupTestEnv();
    final int CLIENT_PORT = PortAssignment.unique();

    // Start up the ZK server to automatically create the necessary directories
    // and capture the directory where data is stored
    MainThread main = new MainThread(CLIENT_PORT, true, null);
    File tmpDir = main.tmpDir;
    main.start();
    Assert.assertTrue("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));
    main.shutdown();

    // Make the snapshot directory read only
    File snapDir = new File(main.dataDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
    snapDir.setWritable(false);

    // Restart ZK and observe a failure
    main = new MainThread(CLIENT_PORT, false, tmpDir, null);
    main.start();

    Assert.assertFalse("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));

    main.shutdown();

    snapDir.setWritable(true);

    main.deleteDirs();
}

/**
 * Tests that the ZooKeeper server will fail to start if the
 * transaction log directory is read only.
 *
 * This test will fail if it is executed as root user.
 */
@Test(timeout = 30000)
public void testReadOnlyTxnLogDir() throws Exception {
    ClientBase.setupTestEnv();
    final int CLIENT_PORT = PortAssignment.unique();

    // Start up the ZK server to automatically create the necessary directories
    // and capture the directory where data is stored
    MainThread main = new MainThread(CLIENT_PORT, true, null);
    File tmpDir = main.tmpDir;
    main.start();
    Assert.assertTrue("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));
    main.shutdown();

    // Make the transaction log directory read only
    File logDir = new File(main.logDir, FileTxnSnapLog.version + FileTxnSnapLog.VERSION);
    logDir.setWritable(false);

    // Restart ZK and observe a failure
    main = new MainThread(CLIENT_PORT, false, tmpDir, null);
    main.start();

    Assert.assertFalse("waiting for server being up", ClientBase
            .waitForServerUp("127.0.0.1:" + CLIENT_PORT,
                    CONNECTION_TIMEOUT / 2));

    main.shutdown();

    logDir.setWritable(true);

    main.deleteDirs();
}

@Test
public void testTxnLogElapsedSyncTime() throws IOException {

    File tmpDir = ClientBase.createEmptyTestDir();
    FileTxnSnapLog fileTxnSnapLog = new FileTxnSnapLog(new File(tmpDir, "data"),
            new File(tmpDir, "data_txnlog"));

    ZooKeeperServer zks = new ZooKeeperServer();
    zks.setTxnLogFactory(fileTxnSnapLog);

    ZooKeeperServerBean serverBean = new ZooKeeperServerBean(zks);
    long elapsedTime = serverBean.getTxnLogElapsedSyncTime();
    assertEquals(-1, elapsedTime);

    TxnHeader hdr = new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.setData);
    Record txn = new SetDataTxn("/foo", new byte[0], 1);
    Request req = new Request(0, 0, 0, hdr, txn, 0);

    try {

        zks.getTxnLogFactory().append(req);
        zks.getTxnLogFactory().commit();
        elapsedTime = serverBean.getTxnLogElapsedSyncTime();

        assertNotEquals(-1, elapsedTime);

        assertEquals(elapsedTime, serverBean.getTxnLogElapsedSyncTime());

    } finally {
        fileTxnSnapLog.close();
    }
}

/**
 * test the purge
 * @throws Exception an exception might be thrown here
 */
@Test
public void testPurge() throws Exception {
    tmpDir = ClientBase.createTmpDir();
    ClientBase.setupTestEnv();
    ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
    SyncRequestProcessor.setSnapCount(100);
    final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
    ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
    f.startup(zks);
    Assert.assertTrue("waiting for server being up ",
            ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));
    ZooKeeper zk = ClientBase.createZKClient(HOSTPORT);
    try {
        for (int i = 0; i< 2000; i++) {
            zk.create("/invalidsnap-" + i, new byte[0], Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT);
        }
    } finally {
        zk.close();
    }
    f.shutdown();
    zks.getTxnLogFactory().close();
    Assert.assertTrue("waiting for server to shutdown",
            ClientBase.waitForServerDown(HOSTPORT, CONNECTION_TIMEOUT));
    // now corrupt the snapshot
    PurgeTxnLog.purge(tmpDir, tmpDir, 3);
    FileTxnSnapLog snaplog = new FileTxnSnapLog(tmpDir, tmpDir);
    List<File> listLogs = snaplog.findNRecentSnapshots(4);
    int numSnaps = 0;
    for (File ff: listLogs) {
        if (ff.getName().startsWith("snapshot")) {
            numSnaps++;
        }
    }
    Assert.assertTrue("exactly 3 snapshots ", (numSnaps == 3));
    snaplog.close();
    zks.shutdown();
}