You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/10/10 10:30:22 UTC
[23/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch 3.4 move java server, client
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/PurgeTxnTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/PurgeTxnTest.java b/src/java/test/org/apache/zookeeper/server/PurgeTxnTest.java
deleted file mode 100644
index 1685a5b..0000000
--- a/src/java/test/org/apache/zookeeper/server/PurgeTxnTest.java
+++ /dev/null
@@ -1,604 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.zookeeper.data.Stat;
-
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.server.PurgeTxnLog;
-import org.apache.zookeeper.server.ServerCnxnFactory;
-import org.apache.zookeeper.server.SyncRequestProcessor;
-import org.apache.zookeeper.server.ZooKeeperServer;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-import org.apache.zookeeper.server.persistence.Util;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PurgeTxnTest extends ZKTestCase implements Watcher {
- private static final Logger LOG = LoggerFactory.getLogger(PurgeTxnTest.class);
- private static String HOSTPORT = "127.0.0.1:" + PortAssignment.unique();
- private static final int CONNECTION_TIMEOUT = 3000;
- private static final long OP_TIMEOUT_IN_MILLIS = 90000;
- private File tmpDir;
-
- @After
- public void teardown() {
- if (null != tmpDir) {
- ClientBase.recursiveDelete(tmpDir);
- }
- }
-
- /**
- * 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 = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this);
- 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();
- }
-
- /**
- * Tests purge when logs are rolling or a new snapshot is created, then
- * these newer files should alse be excluded in the current cycle.
- *
- * For frequent snapshotting, configured SnapCount to 30. There are three
- * threads which will create 1000 znodes each and simultaneously do purge
- * call
- */
- @Test
- public void testPurgeWhenLogRollingInProgress() throws Exception {
- tmpDir = ClientBase.createTmpDir();
- ClientBase.setupTestEnv();
- ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
- SyncRequestProcessor.setSnapCount(30);
- 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));
- final ZooKeeper zk = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this);
- final CountDownLatch doPurge = new CountDownLatch(1);
- final CountDownLatch purgeFinished = new CountDownLatch(1);
- final AtomicBoolean opFailed = new AtomicBoolean(false);
- new Thread() {
- public void run() {
- try {
- doPurge.await(OP_TIMEOUT_IN_MILLIS / 2,
- TimeUnit.MILLISECONDS);
- PurgeTxnLog.purge(tmpDir, tmpDir, 3);
- } catch (IOException ioe) {
- LOG.error("Exception when purge", ioe);
- opFailed.set(true);
- } catch (InterruptedException ie) {
- LOG.error("Exception when purge", ie);
- opFailed.set(true);
- } finally {
- purgeFinished.countDown();
- }
- };
- }.start();
- final int thCount = 3;
- List<String> znodes = manyClientOps(zk, doPurge, thCount,
- "/invalidsnap");
- Assert.assertTrue("Purging is not finished!", purgeFinished.await(
- OP_TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS));
- Assert.assertFalse("Purging failed!", opFailed.get());
- for (String znode : znodes) {
- try {
- zk.getData(znode, false, null);
- } catch (Exception ke) {
- LOG.error("Unexpected exception when visiting znode!", ke);
- Assert.fail("Unexpected exception when visiting znode!");
- }
- }
- zk.close();
- f.shutdown();
- zks.shutdown();
- zks.getTxnLogFactory().close();
- }
-
- /**
- * Tests finding n recent snapshots from set of snapshots and data logs
- */
- @Test
- public void testFindNRecentSnapshots() throws Exception {
- int nRecentSnap = 4; // n recent snap shots
- int nRecentCount = 30;
- int offset = 0;
-
- tmpDir = ClientBase.createTmpDir();
- File version2 = new File(tmpDir.toString(), "version-2");
- Assert.assertTrue("Failed to create version_2 dir:" + version2.toString(),
- version2.mkdir());
-
- // Test that with no snaps, findNRecentSnapshots returns empty list
- FileTxnSnapLog txnLog = new FileTxnSnapLog(tmpDir, tmpDir);
- List<File> foundSnaps = txnLog.findNRecentSnapshots(1);
- assertEquals(0, foundSnaps.size());
-
- List<File> expectedNRecentSnapFiles = new ArrayList<File>();
- int counter = offset + (2 * nRecentCount);
- for (int i = 0; i < nRecentCount; i++) {
- // simulate log file
- File logFile = new File(version2 + "/log." + Long.toHexString(--counter));
- Assert.assertTrue("Failed to create log File:" + logFile.toString(),
- logFile.createNewFile());
- // simulate snapshot file
- File snapFile = new File(version2 + "/snapshot."
- + Long.toHexString(--counter));
- Assert.assertTrue("Failed to create snap File:" + snapFile.toString(),
- snapFile.createNewFile());
- // add the n recent snap files for assertion
- if(i < nRecentSnap){
- expectedNRecentSnapFiles.add(snapFile);
- }
- }
-
- // Test that when we ask for recent snaps we get the number we asked for and
- // the files we expected
- List<File> nRecentSnapFiles = txnLog.findNRecentSnapshots(nRecentSnap);
- Assert.assertEquals("exactly 4 snapshots ", 4,
- nRecentSnapFiles.size());
- expectedNRecentSnapFiles.removeAll(nRecentSnapFiles);
- Assert.assertEquals("Didn't get the recent snap files", 0,
- expectedNRecentSnapFiles.size());
-
- // Test that when asking for more snaps than we created, we still only get snaps
- // not logs or anything else (per ZOOKEEPER-2420)
- nRecentSnapFiles = txnLog.findNRecentSnapshots(nRecentCount + 5);
- assertEquals(nRecentCount, nRecentSnapFiles.size());
- for (File f: nRecentSnapFiles) {
- Assert.assertTrue("findNRecentSnapshots() returned a non-snapshot: " + f.getPath(),
- (Util.getZxidFromName(f.getName(), "snapshot") != -1));
- }
-
- txnLog.close();
- }
-
- /**
- * Tests purge where the data directory contains old snapshots and data
- * logs, newest snapshots and data logs, (newest + n) snapshots and data
- * logs
- */
- @Test
- public void testSnapFilesGreaterThanToRetain() throws Exception {
- int nRecentCount = 4;
- int fileAboveRecentCount = 4;
- int fileToPurgeCount = 2;
- AtomicInteger offset = new AtomicInteger(0);
- tmpDir = ClientBase.createTmpDir();
- File version2 = new File(tmpDir.toString(), "version-2");
- Assert.assertTrue("Failed to create version_2 dir:" + version2.toString(),
- version2.mkdir());
- List<File> snapsToPurge = new ArrayList<File>();
- List<File> logsToPurge = new ArrayList<File>();
- List<File> snaps = new ArrayList<File>();
- List<File> logs = new ArrayList<File>();
- List<File> snapsAboveRecentFiles = new ArrayList<File>();
- List<File> logsAboveRecentFiles = new ArrayList<File>();
- createDataDirFiles(offset, fileToPurgeCount, false, version2, snapsToPurge,
- logsToPurge);
- createDataDirFiles(offset, nRecentCount, false, version2, snaps, logs);
- logs.add(logsToPurge.remove(0)); // log that precedes first retained snapshot is also retained
- createDataDirFiles(offset, fileAboveRecentCount, false, version2,
- snapsAboveRecentFiles, logsAboveRecentFiles);
-
- /**
- * The newest log file preceding the oldest retained snapshot is not removed as it may
- * contain transactions newer than the oldest snapshot.
- */
- logsToPurge.remove(0);
-
- FileTxnSnapLog txnLog = new FileTxnSnapLog(tmpDir, tmpDir);
- PurgeTxnLog.purgeOlderSnapshots(txnLog, snaps.get(snaps.size() - 1));
- txnLog.close();
- verifyFilesAfterPurge(snapsToPurge, false);
- verifyFilesAfterPurge(logsToPurge, false);
- verifyFilesAfterPurge(snaps, true);
- verifyFilesAfterPurge(logs, true);
- verifyFilesAfterPurge(snapsAboveRecentFiles, true);
- verifyFilesAfterPurge(logsAboveRecentFiles, true);
- }
-
- /**
- * Tests purge where the data directory contains snap files and log files equals to the
- * number of files to be retained
- */
- @Test
- public void testSnapFilesEqualsToRetain() throws Exception {
- internalTestSnapFilesEqualsToRetain(false);
- }
-
- /**
- * Tests purge where the data directory contains snap files equals to the
- * number of files to be retained, and a log file that precedes the earliest snapshot
- */
- @Test
- public void testSnapFilesEqualsToRetainWithPrecedingLog() throws Exception {
- internalTestSnapFilesEqualsToRetain(true);
- }
-
- public void internalTestSnapFilesEqualsToRetain(boolean testWithPrecedingLogFile) throws Exception {
- int nRecentCount = 3;
- AtomicInteger offset = new AtomicInteger(0);
- tmpDir = ClientBase.createTmpDir();
- File version2 = new File(tmpDir.toString(), "version-2");
- Assert.assertTrue("Failed to create version_2 dir:" + version2.toString(),
- version2.mkdir());
- List<File> snaps = new ArrayList<File>();
- List<File> logs = new ArrayList<File>();
- createDataDirFiles(offset, nRecentCount, testWithPrecedingLogFile, version2, snaps, logs);
-
- FileTxnSnapLog txnLog = new FileTxnSnapLog(tmpDir, tmpDir);
- PurgeTxnLog.purgeOlderSnapshots(txnLog, snaps.get(snaps.size() - 1));
- txnLog.close();
- verifyFilesAfterPurge(snaps, true);
- verifyFilesAfterPurge(logs, true);
- }
-
- /**
- * Tests purge where the data directory contains old snapshots and data
- * logs, newest snapshots and data logs
- */
- @Test
- public void testSnapFilesLessThanToRetain() throws Exception {
- int nRecentCount = 4;
- int fileToPurgeCount = 2;
- AtomicInteger offset = new AtomicInteger(0);
- tmpDir = ClientBase.createTmpDir();
- File version2 = new File(tmpDir.toString(), "version-2");
- Assert.assertTrue("Failed to create version_2 dir:" + version2.toString(),
- version2.mkdir());
- List<File> snapsToPurge = new ArrayList<File>();
- List<File> logsToPurge = new ArrayList<File>();
- List<File> snaps = new ArrayList<File>();
- List<File> logs = new ArrayList<File>();
- createDataDirFiles(offset, fileToPurgeCount, false, version2, snapsToPurge,
- logsToPurge);
- createDataDirFiles(offset, nRecentCount, false, version2, snaps, logs);
- logs.add(logsToPurge.remove(0)); // log that precedes first retained snapshot is also retained
-
- /**
- * The newest log file preceding the oldest retained snapshot is not removed as it may
- * contain transactions newer than the oldest snapshot.
- */
- logsToPurge.remove(0);
-
- FileTxnSnapLog txnLog = new FileTxnSnapLog(tmpDir, tmpDir);
- PurgeTxnLog.purgeOlderSnapshots(txnLog, snaps.get(snaps.size() - 1));
- txnLog.close();
- verifyFilesAfterPurge(snapsToPurge, false);
- verifyFilesAfterPurge(logsToPurge, false);
- verifyFilesAfterPurge(snaps, true);
- verifyFilesAfterPurge(logs, true);
- }
-
- /**
- * PurgeTxnLog is called with dataLogDir snapDir -n count This test case
- * verify these values are parsed properly and functionality works fine
- */
- @Test
- public void testPurgeTxnLogWithDataDir()
- throws Exception {
- tmpDir = ClientBase.createTmpDir();
- File dataDir = new File(tmpDir, "dataDir");
- File dataLogDir = new File(tmpDir, "dataLogDir");
-
- File dataDirVersion2 = new File(dataDir, "version-2");
- dataDirVersion2.mkdirs();
- File dataLogDirVersion2 = new File(dataLogDir, "version-2");
- dataLogDirVersion2.mkdirs();
-
- // create dummy log and transaction file
- int totalFiles = 20;
-
- // create transaction and snapshot files in different-different
- // directories
- for (int i = 0; i < totalFiles; i++) {
- // simulate log file
- File logFile = new File(dataLogDirVersion2, "log."
- + Long.toHexString(i));
- logFile.createNewFile();
- // simulate snapshot file
- File snapFile = new File(dataDirVersion2, "snapshot."
- + Long.toHexString(i));
- snapFile.createNewFile();
- }
-
- int numberOfSnapFilesToKeep = 10;
- // scenario where four parameter are passed
- String[] args = new String[] { dataLogDir.getAbsolutePath(),
- dataDir.getAbsolutePath(), "-n",
- Integer.toString(numberOfSnapFilesToKeep) };
- PurgeTxnLog.main(args);
-
- assertEquals(numberOfSnapFilesToKeep, dataDirVersion2.listFiles().length);
- // Since for each snapshot we have a log file with same zxid, expect same # logs as snaps to be kept
- assertEquals(numberOfSnapFilesToKeep, dataLogDirVersion2.listFiles().length);
- ClientBase.recursiveDelete(tmpDir);
-
- }
-
- /**
- * PurgeTxnLog is called with dataLogDir -n count This test case verify
- * these values are parsed properly and functionality works fine
- */
- @Test
- public void testPurgeTxnLogWithoutDataDir()
- throws Exception {
- tmpDir = ClientBase.createTmpDir();
- File dataDir = new File(tmpDir, "dataDir");
- File dataLogDir = new File(tmpDir, "dataLogDir");
-
- File dataDirVersion2 = new File(dataDir, "version-2");
- dataDirVersion2.mkdirs();
- File dataLogDirVersion2 = new File(dataLogDir, "version-2");
- dataLogDirVersion2.mkdirs();
-
- // create dummy log and transaction file
- int totalFiles = 20;
-
- // create transaction and snapshot files in data directory
- for (int i = 0; i < totalFiles; i++) {
- // simulate log file
- File logFile = new File(dataLogDirVersion2, "log."
- + Long.toHexString(i));
- logFile.createNewFile();
- // simulate snapshot file
- File snapFile = new File(dataLogDirVersion2, "snapshot."
- + Long.toHexString(i));
- snapFile.createNewFile();
- }
-
- int numberOfSnapFilesToKeep = 10;
- // scenario where only three parameter are passed
- String[] args = new String[] { dataLogDir.getAbsolutePath(), "-n",
- Integer.toString(numberOfSnapFilesToKeep) };
- PurgeTxnLog.main(args);
- assertEquals(numberOfSnapFilesToKeep * 2, // Since for each snapshot we have a log file with same zxid, expect same # logs as snaps to be kept
- dataLogDirVersion2.listFiles().length);
- ClientBase.recursiveDelete(tmpDir);
-
- }
-
- /**
- * Verifies that purge does not delete any log files which started before the oldest retained
- * snapshot but which might extend beyond it.
- * @throws Exception an exception might be thrown here
- */
- @Test
- public void testPurgeDoesNotDeleteOverlappingLogFile() throws Exception {
- // Setting used for snapRetainCount in this test.
- final int SNAP_RETAIN_COUNT = 3;
- // Number of znodes this test creates in each snapshot.
- final int NUM_ZNODES_PER_SNAPSHOT = 100;
- /**
- * Set a sufficiently high snapCount to ensure that we don't rollover the log. Normally,
- * the default value (100K at time of this writing) would ensure this, but we make that
- * dependence explicit here to make the test future-proof. Not rolling over the log is
- * important for this test since we are testing retention of the one and only log file which
- * predates each retained snapshot.
- */
- SyncRequestProcessor.setSnapCount(SNAP_RETAIN_COUNT * NUM_ZNODES_PER_SNAPSHOT * 10);
-
- // Create Zookeeper and connect to it.
- tmpDir = ClientBase.createTmpDir();
- ClientBase.setupTestEnv();
- ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
- 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);
-
- // Unique identifier for each znode that we create.
- int unique = 0;
- try {
- /**
- * Create some znodes and take a snapshot. Repeat this until we have SNAP_RETAIN_COUNT
- * snapshots. Do not rollover the log.
- */
- for (int snapshotCount = 0; snapshotCount < SNAP_RETAIN_COUNT; snapshotCount++) {
- for (int i = 0; i< 100; i++, unique++) {
- zk.create("/snap-" + unique, new byte[0], Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- }
- zks.takeSnapshot();
- }
- // Create some additional znodes without taking a snapshot afterwards.
- for (int i = 0; i< 100; i++, unique++) {
- zk.create("/snap-" + unique, new byte[0], Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- }
- } finally {
- zk.close();
- }
-
- // Shutdown Zookeeper.
- f.shutdown();
- zks.getTxnLogFactory().close();
- zks.shutdown();
- Assert.assertTrue("waiting for server to shutdown",
- ClientBase.waitForServerDown(HOSTPORT, CONNECTION_TIMEOUT));
-
- // Purge snapshot and log files.
- PurgeTxnLog.purge(tmpDir, tmpDir, SNAP_RETAIN_COUNT);
-
- // Initialize Zookeeper again from the same dataDir.
- zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
- f = ServerCnxnFactory.createFactory(PORT, -1);
- f.startup(zks);
- zk = ClientBase.createZKClient(HOSTPORT);
-
- /**
- * Verify that the last znode that was created above exists. This znode's creation was
- * captured by the transaction log which was created before any of the above
- * SNAP_RETAIN_COUNT snapshots were created, but it's not captured in any of these
- * snapshots. So for it it exist, the (only) existing log file should not have been purged.
- */
- final String lastZnode = "/snap-" + (unique - 1);
- final Stat stat = zk.exists(lastZnode, false);
- Assert.assertNotNull("Last znode does not exist: " + lastZnode, stat);
-
- // Shutdown for the last time.
- f.shutdown();
- zks.getTxnLogFactory().close();
- zks.shutdown();
- }
-
- private File createDataDirLogFile(File version_2, int Zxid) throws IOException {
- File logFile = new File(version_2 + "/log." + Long.toHexString(Zxid));
- Assert.assertTrue("Failed to create log File:" + logFile.toString(),
- logFile.createNewFile());
- return logFile;
- }
-
- private void createDataDirFiles(AtomicInteger offset, int limit, boolean createPrecedingLogFile,
- File version_2, List<File> snaps, List<File> logs)
- throws IOException {
- int counter = offset.get() + (2 * limit);
- if (createPrecedingLogFile) {
- counter++;
- }
- offset.set(counter);
- for (int i = 0; i < limit; i++) {
- // simulate log file
- logs.add(createDataDirLogFile(version_2, --counter));
- // simulate snapshot file
- File snapFile = new File(version_2 + "/snapshot."
- + Long.toHexString(--counter));
- Assert.assertTrue("Failed to create snap File:" + snapFile.toString(),
- snapFile.createNewFile());
- snaps.add(snapFile);
- }
- if (createPrecedingLogFile) {
- logs.add(createDataDirLogFile(version_2, --counter));
- }
- }
-
- private void verifyFilesAfterPurge(List<File> logs, boolean exists) {
- for (File file : logs) {
- Assert.assertEquals("After purging, file " + file, exists,
- file.exists());
- }
- }
-
- private List<String> manyClientOps(final ZooKeeper zk,
- final CountDownLatch doPurge, int thCount, final String prefix) {
- Thread[] ths = new Thread[thCount];
- final List<String> znodes = Collections
- .synchronizedList(new ArrayList<String>());
- final CountDownLatch finished = new CountDownLatch(thCount);
- for (int indx = 0; indx < thCount; indx++) {
- final String myprefix = prefix + "-" + indx;
- Thread th = new Thread() {
- public void run() {
- for (int i = 0; i < 1000; i++) {
- try {
- String mynode = myprefix + "-" + i;
- znodes.add(mynode);
- zk.create(mynode, new byte[0], Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- } catch (Exception e) {
- LOG.error("Unexpected exception occurred!", e);
- }
- if (i == 200) {
- doPurge.countDown();
- }
- }
- finished.countDown();
- };
- };
- ths[indx] = th;
- }
-
- for (Thread thread : ths) {
- thread.start();
- }
- try {
- Assert.assertTrue("ZkClient ops is not finished!",
- finished.await(OP_TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS));
- } catch (InterruptedException ie) {
- LOG.error("Unexpected exception occurred!", ie);
- Assert.fail("Unexpected exception occurred!");
- }
- return znodes;
- }
-
- public void process(WatchedEvent event) {
- // do nothing
- }
-
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/ReferenceCountedACLCacheTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/ReferenceCountedACLCacheTest.java b/src/java/test/org/apache/zookeeper/server/ReferenceCountedACLCacheTest.java
deleted file mode 100644
index 795472f..0000000
--- a/src/java/test/org/apache/zookeeper/server/ReferenceCountedACLCacheTest.java
+++ /dev/null
@@ -1,287 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import org.apache.jute.BinaryInputArchive;
-import org.apache.jute.BinaryOutputArchive;
-import org.apache.jute.InputArchive;
-import org.apache.jute.OutputArchive;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
-import org.junit.Test;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.junit.Assert.*;
-
-public class ReferenceCountedACLCacheTest {
- @Test
- public void testSameACLGivesSameID() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
-
- List<ACL> testACL2 = createACL("myid");
-
- assertEquals(aclId, cache.convertAcls(testACL2));
- }
-
- @Test
- public void testWhetherOrderingMatters() {
- List<ACL> testACL = new ArrayList<ACL>();
- testACL.add(new ACL(ZooDefs.Perms.READ, new Id("scheme", "ro")));
- testACL.add(new ACL(ZooDefs.Perms.WRITE, new Id("scheme", "rw")));
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
-
- List<ACL> testACL2 = new ArrayList<ACL>();
- testACL2.add(new ACL(ZooDefs.Perms.WRITE, new Id("scheme", "rw")));
- testACL2.add(new ACL(ZooDefs.Perms.READ, new Id("scheme", "ro")));
-
- assertFalse(aclId.equals(cache.convertAcls(testACL2)));
- }
-
- @Test
- public void testBidirectionality() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
-
- assertEquals(testACL, cache.convertLong(aclId));
- }
-
- @Test
- public void testCacheSize() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
- assertEquals(1, cache.size());
-
- List<ACL> testACL2 = createACL("myid");
-
- assertEquals(aclId, cache.convertAcls(testACL2));
- assertEquals(1, cache.size());
-
- List<ACL> testACL3 = createACL("differentId");
-
- Long aclId3 = cache.convertAcls(testACL3);
- assertFalse(aclId3.equals(aclId));
- assertEquals(2, cache.size());
- }
-
- @Test
- public void testAddThenRemove() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
- assertEquals(1, cache.size());
-
- cache.removeUsage(aclId);
- assertEquals(0, cache.size());
- }
-
- @Test
- public void testMultipleAddsAndRemove() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
- assertEquals(1, cache.size());
-
- cache.convertAcls(testACL);
- assertEquals(1, cache.size());
-
- List<ACL> testACL2 = createACL("anotherId");
- cache.convertAcls(testACL2);
-
- cache.removeUsage(aclId);
- assertEquals(2, cache.size());
- cache.removeUsage(aclId);
- assertEquals(1, cache.size());
-
- Long newId = cache.convertAcls(testACL);
- assertFalse(aclId.equals(newId));
- }
-
- @Test
- public void testAddUsage() {
- List<ACL> testACL = createACL("myid");
-
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- Long aclId = cache.convertAcls(testACL);
- assertEquals(1, cache.size());
-
- cache.addUsage(aclId);
- assertEquals(1, cache.size());
-
- cache.removeUsage(aclId);
- assertEquals(1, cache.size());
- cache.removeUsage(aclId);
- assertEquals(0, cache.size());
- }
-
- @Test
- public void testAddNonExistentUsage() {
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
- cache.addUsage(1L);
-
- assertEquals(0, cache.size());
- /*
- On startup, it's possible that we'll try calling addUsage of an ID not in the cache. This is safe to ignore
- as it'll be added later when we traverse the tranlog. See discussion here:
- http://mail-archives.apache.org/mod_mbox/zookeeper-user/201507.mbox/%3CCAB5oV2_ujhvBA1sEkCG2WRakPjCy%2BNR10620WK2G1GGgmEO44g%40mail.gmail.com%3E
-
- This test makes sure that we don't add the ID to the cache in this case as that would result in dupes later
- and consequently incorrect counts and entries that will never be cleaned out.
- */
- }
-
- @Test
- public void testSerializeDeserialize() throws IOException {
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
-
- List<ACL> acl1 = createACL("one");
- List<ACL> acl2 = createACL("two");
- List<ACL> acl3 = createACL("three");
- List<ACL> acl4 = createACL("four");
- List<ACL> acl5 = createACL("five");
-
- Long aclId1 = convertACLsNTimes(cache, acl1, 1);
- Long aclId2 = convertACLsNTimes(cache, acl2, 2);
- Long aclId3 = convertACLsNTimes(cache, acl3, 3);
- Long aclId4 = convertACLsNTimes(cache, acl4, 4);
- Long aclId5 = convertACLsNTimes(cache, acl5, 5);
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- BinaryOutputArchive archive = BinaryOutputArchive.getArchive(baos);
- cache.serialize(archive);
-
- BinaryInputArchive inArchive = BinaryInputArchive.getArchive(new ByteArrayInputStream(baos.toByteArray()));
- ReferenceCountedACLCache deserializedCache = new ReferenceCountedACLCache();
- deserializedCache.deserialize(inArchive);
- callAddUsageNTimes(deserializedCache, aclId1, 1);
- callAddUsageNTimes(deserializedCache, aclId2, 2);
- callAddUsageNTimes(deserializedCache, aclId3, 3);
- callAddUsageNTimes(deserializedCache, aclId4, 4);
- callAddUsageNTimes(deserializedCache, aclId5, 5);
- assertCachesEqual(cache, deserializedCache);
- }
-
- @Test
- public void testNPEInDeserialize() throws IOException {
- ReferenceCountedACLCache serializeCache = new ReferenceCountedACLCache(){
- @Override
- public synchronized void serialize(OutputArchive oa) throws IOException {
- oa.writeInt(1, "map");
- oa.writeLong(1, "long");
- oa.startVector(null, "acls");
- oa.endVector(null, "acls");
- }
- };
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- BinaryOutputArchive archive = BinaryOutputArchive.getArchive(baos);
- serializeCache.serialize(archive);
- BinaryInputArchive inArchive = BinaryInputArchive.getArchive(new ByteArrayInputStream(baos.toByteArray()));
- ReferenceCountedACLCache deserializedCache = new ReferenceCountedACLCache();
- try {
- deserializedCache.deserialize(inArchive);
- } catch (NullPointerException e){
- fail("should not throw NPE while do deserialized");
- } catch (RuntimeException e) {
- // do nothing.
- }
- }
-
-
- private void assertCachesEqual(ReferenceCountedACLCache expected, ReferenceCountedACLCache actual){
- assertEquals(expected.aclIndex, actual.aclIndex);
- assertEquals(expected.aclKeyMap, actual.aclKeyMap);
- assertEquals(expected.longKeyMap, actual.longKeyMap);
- assertEquals(expected.referenceCounter, actual.referenceCounter);
- }
-
- @Test
- public void testPurgeUnused() throws IOException {
- ReferenceCountedACLCache cache = new ReferenceCountedACLCache();
-
- List<ACL> acl1 = createACL("one");
- List<ACL> acl2 = createACL("two");
- List<ACL> acl3 = createACL("three");
- List<ACL> acl4 = createACL("four");
- List<ACL> acl5 = createACL("five");
-
- Long aclId1 = convertACLsNTimes(cache, acl1, 1);
- Long aclId2 = convertACLsNTimes(cache, acl2, 2);
- Long aclId3 = convertACLsNTimes(cache, acl3, 3);
- Long aclId4 = convertACLsNTimes(cache, acl4, 4);
- Long aclId5 = convertACLsNTimes(cache, acl5, 5);
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- BinaryOutputArchive archive = BinaryOutputArchive.getArchive(baos);
- cache.serialize(archive);
-
- BinaryInputArchive inArchive = BinaryInputArchive.getArchive(new ByteArrayInputStream(baos.toByteArray()));
- ReferenceCountedACLCache deserializedCache = new ReferenceCountedACLCache();
- deserializedCache.deserialize(inArchive);
- callAddUsageNTimes(deserializedCache, aclId1, 1);
- callAddUsageNTimes(deserializedCache, aclId2, 2);
- deserializedCache.purgeUnused();
-
- assertEquals(2, deserializedCache.size());
- assertEquals(aclId1, deserializedCache.convertAcls(acl1));
- assertEquals(aclId2, deserializedCache.convertAcls(acl2));
- assertFalse(acl3.equals(deserializedCache.convertAcls(acl3)));
- assertFalse(acl4.equals(deserializedCache.convertAcls(acl4)));
- assertFalse(acl5.equals(deserializedCache.convertAcls(acl5)));
- }
-
- private void callAddUsageNTimes(ReferenceCountedACLCache deserializedCache, Long aclId, int num) {
- for (int i = 0; i < num; i++) {
- deserializedCache.addUsage(aclId);
- }
- }
-
- private Long convertACLsNTimes(ReferenceCountedACLCache cache, List<ACL> acl, int num) {
- if (num <= 0) {
- return -1L;
- }
-
- for (int i = 0; i < num -1; i++) {
- cache.convertAcls(acl);
- }
-
- return cache.convertAcls(acl);
- }
-
- private List<ACL> createACL(String id) {
- List<ACL> acl1 = new ArrayList<ACL>();
- acl1.add(new ACL(ZooDefs.Perms.ADMIN, new Id("scheme", id)));
- return acl1;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/SerializationPerfTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/SerializationPerfTest.java b/src/java/test/org/apache/zookeeper/server/SerializationPerfTest.java
deleted file mode 100644
index 4398232..0000000
--- a/src/java/test/org/apache/zookeeper/server/SerializationPerfTest.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.jute.BinaryOutputArchive;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZKTestCase;
-import org.junit.Test;
-
-public class SerializationPerfTest extends ZKTestCase {
- protected static final Logger LOG = LoggerFactory.getLogger(SerializationPerfTest.class);
-
- private static class NullOutputStream extends OutputStream {
- public void write(int b) {
- // do nothing - exclude persistence from perf
- }
- }
-
- static int createNodes(DataTree tree, String path, int depth,
- int childcount, int parentCVersion, byte[] data) throws KeeperException.NodeExistsException, KeeperException.NoNodeException {
- path += "node" + depth;
- tree.createNode(path, data, null, -1, ++parentCVersion, 1, 1);
-
- if (--depth == 0) {
- return 1;
- }
-
- path += "/";
-
- int count = 1;
- for (int i = 0; i < childcount; i++) {
- count += createNodes(tree, path + i, depth, childcount, 1, data);
- }
-
- return count;
- }
-
- private static void serializeTree(int depth, int width, int len)
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- DataTree tree = new DataTree();
- createNodes(tree, "/", depth, width, tree.getNode("/").stat.getCversion(), new byte[len]);
- int count = tree.getNodeCount();
-
- BinaryOutputArchive oa =
- BinaryOutputArchive.getArchive(new NullOutputStream());
- System.gc();
- long start = System.nanoTime();
- tree.serialize(oa, "test");
- long end = System.nanoTime();
- long durationms = (end - start)/1000000L;
- long pernodeus = ((end - start)/1000L)/count;
- LOG.info("Serialized " + count + " nodes in "
- + durationms + " ms (" + pernodeus + "us/node), depth="
- + depth + " width=" + width + " datalen=" + len);
- }
-
- @Test
- public void testSingleSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(1, 0, 20);
- }
-
- @Test
- public void testWideSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(2, 10000, 20);
- }
-
- @Test
- public void testDeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(400, 1, 20);
- }
-
- @Test
- public void test10Wide5DeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(5, 10, 20);
- }
-
- @Test
- public void test15Wide5DeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(5, 15, 20);
- }
-
- @Test
- public void test25Wide4DeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(4, 25, 20);
- }
-
- @Test
- public void test40Wide4DeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(4, 40, 20);
- }
-
- @Test
- public void test300Wide3DeepSerialize()
- throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
- serializeTree(3, 300, 20);
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/ServerCnxnFactoryAccessor.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/ServerCnxnFactoryAccessor.java b/src/java/test/org/apache/zookeeper/server/ServerCnxnFactoryAccessor.java
deleted file mode 100644
index 0dfdcfe..0000000
--- a/src/java/test/org/apache/zookeeper/server/ServerCnxnFactoryAccessor.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server;
-
-public class ServerCnxnFactoryAccessor {
- public static ZooKeeperServer getZkServer(ServerCnxnFactory fac) {
- return fac.zkServer;
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/ServerStatsTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/ServerStatsTest.java b/src/java/test/org/apache/zookeeper/server/ServerStatsTest.java
deleted file mode 100644
index 89c78f9..0000000
--- a/src/java/test/org/apache/zookeeper/server/ServerStatsTest.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.common.Time;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.mockito.Mockito.mock;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-
-public class ServerStatsTest extends ZKTestCase {
-
- private ServerStats.Provider providerMock;
-
- @Before
- public void setUp() {
- providerMock = mock(ServerStats.Provider.class);
- }
-
- @Test
- public void testPacketsMetrics() {
- // Given ...
- ServerStats serverStats = new ServerStats(providerMock);
- int incrementCount = 20;
-
- // When increment ...
- for (int i = 0; i < incrementCount; i++) {
- serverStats.incrementPacketsSent();
- serverStats.incrementPacketsReceived();
- serverStats.incrementPacketsReceived();
- }
-
- // Then ...
- Assert.assertEquals(incrementCount, serverStats.getPacketsSent());
- Assert.assertEquals(incrementCount*2, serverStats.getPacketsReceived());
-
- // When reset ...
- serverStats.resetRequestCounters();
-
- // Then ...
- assertAllPacketsZero(serverStats);
-
- }
-
- @Test
- public void testLatencyMetrics() {
- // Given ...
- ServerStats serverStats = new ServerStats(providerMock);
-
- // When incremented...
- serverStats.updateLatency(Time.currentElapsedTime()-1000);
- serverStats.updateLatency(Time.currentElapsedTime()-2000);
-
- // Then ...
- assertThat("Max latency check", 2000L,
- greaterThanOrEqualTo(serverStats.getMaxLatency()));
- assertThat("Min latency check", 1000L,
- greaterThanOrEqualTo(serverStats.getMinLatency()));
- assertThat("Avg latency check", 1500L,
- greaterThanOrEqualTo(serverStats.getAvgLatency()));
-
- // When reset...
- serverStats.resetLatency();
-
- // Then ...
- assertAllLatencyZero(serverStats);
- }
-
- @Test
- public void testFsyncThresholdExceedMetrics() {
- // Given ...
- ServerStats serverStats = new ServerStats(providerMock);
- int incrementCount = 30;
-
- // When increment ...
- for (int i = 0; i < incrementCount; i++) {
- serverStats.incrementFsyncThresholdExceedCount();
- }
-
- // Then ...
- Assert.assertEquals(incrementCount, serverStats.getFsyncThresholdExceedCount());
-
- // When reset ...
- serverStats.resetFsyncThresholdExceedCount();
-
- // Then ...
- assertFsyncThresholdExceedCountZero(serverStats);
-
- }
-
- @Test
- public void testReset() {
- // Given ...
- ServerStats serverStats = new ServerStats(providerMock);
-
- assertAllPacketsZero(serverStats);
- assertAllLatencyZero(serverStats);
-
- // When ...
- serverStats.incrementPacketsSent();
- serverStats.incrementPacketsReceived();
- serverStats.updateLatency(Time.currentElapsedTime()-1000);
-
- serverStats.reset();
-
- // Then ...
- assertAllPacketsZero(serverStats);
- assertAllLatencyZero(serverStats);
- }
-
- private void assertAllPacketsZero(ServerStats serverStats) {
- Assert.assertEquals(0L, serverStats.getPacketsSent());
- Assert.assertEquals(0L, serverStats.getPacketsReceived());
- }
-
- private void assertAllLatencyZero(ServerStats serverStats) {
- Assert.assertEquals(0L, serverStats.getMaxLatency());
- Assert.assertEquals(0L, serverStats.getMinLatency());
- Assert.assertEquals(0L, serverStats.getAvgLatency());
- }
-
- private void assertFsyncThresholdExceedCountZero(ServerStats serverStats) {
- Assert.assertEquals(0L, serverStats.getFsyncThresholdExceedCount());
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/SessionTrackerTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/SessionTrackerTest.java b/src/java/test/org/apache/zookeeper/server/SessionTrackerTest.java
deleted file mode 100644
index f427102..0000000
--- a/src/java/test/org/apache/zookeeper/server/SessionTrackerTest.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import junit.framework.Assert;
-
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.server.SessionTrackerImpl.SessionImpl;
-import org.apache.zookeeper.test.ClientBase;
-import org.junit.Test;
-
-/**
- * Testing zk client session logic in sessiontracker
- */
-public class SessionTrackerTest extends ZKTestCase {
-
- private final long sessionId = 339900;
- private final int sessionTimeout = 3000;
- private FirstProcessor firstProcessor;
- private CountDownLatch latch;
-
- /**
- * Verify the create session call in the Leader.FinalRequestProcessor after
- * the session expiration.
- */
- @Test(timeout = 20000)
- public void testAddSessionAfterSessionExpiry() throws Exception {
- ZooKeeperServer zks = setupSessionTracker();
-
- latch = new CountDownLatch(1);
- zks.sessionTracker.addSession(sessionId, sessionTimeout);
- SessionTrackerImpl sessionTrackerImpl = (SessionTrackerImpl) zks.sessionTracker;
- SessionImpl sessionImpl = sessionTrackerImpl.sessionsById
- .get(sessionId);
- Assert.assertNotNull("Sessionid:" + sessionId
- + " doesn't exists in sessiontracker", sessionImpl);
-
- // verify the session existence
- Object sessionOwner = new Object();
- sessionTrackerImpl.checkSession(sessionId, sessionOwner);
-
- // waiting for the session expiry
- latch.await(sessionTimeout * 2, TimeUnit.MILLISECONDS);
-
- // Simulating FinalRequestProcessor logic: create session request has
- // delayed and now reaches FinalRequestProcessor. Here the leader zk
- // will do sessionTracker.addSession(id, timeout)
- sessionTrackerImpl.addSession(sessionId, sessionTimeout);
- try {
- sessionTrackerImpl.checkSession(sessionId, sessionOwner);
- Assert.fail("Should throw session expiry exception "
- + "as the session has expired and closed");
- } catch (KeeperException.SessionExpiredException e) {
- // expected behaviour
- }
- Assert.assertTrue("Session didn't expired", sessionImpl.isClosing());
- Assert.assertFalse("Session didn't expired", sessionTrackerImpl
- .touchSession(sessionId, sessionTimeout));
- Assert.assertEquals(
- "Duplicate session expiry request has been generated", 1,
- firstProcessor.getCountOfCloseSessionReq());
- }
-
- /**
- * Verify the session closure request has reached PrepRequestProcessor soon
- * after session expiration by the session tracker
- */
- @Test(timeout = 20000)
- public void testCloseSessionRequestAfterSessionExpiry() throws Exception {
- ZooKeeperServer zks = setupSessionTracker();
-
- latch = new CountDownLatch(1);
- zks.sessionTracker.addSession(sessionId, sessionTimeout);
- SessionTrackerImpl sessionTrackerImpl = (SessionTrackerImpl) zks.sessionTracker;
- SessionImpl sessionImpl = sessionTrackerImpl.sessionsById
- .get(sessionId);
- Assert.assertNotNull("Sessionid:" + sessionId
- + " doesn't exists in sessiontracker", sessionImpl);
-
- // verify the session existence
- Object sessionOwner = new Object();
- sessionTrackerImpl.checkSession(sessionId, sessionOwner);
-
- // waiting for the session expiry
- latch.await(sessionTimeout * 2, TimeUnit.MILLISECONDS);
-
- // Simulating close session request: removeSession() will be executed
- // while OpCode.closeSession
- sessionTrackerImpl.removeSession(sessionId);
- SessionImpl actualSession = sessionTrackerImpl.sessionsById
- .get(sessionId);
- Assert.assertNull("Session:" + sessionId
- + " still exists after removal", actualSession);
- }
-
- private ZooKeeperServer setupSessionTracker() throws IOException {
- File tmpDir = ClientBase.createTmpDir();
- ClientBase.setupTestEnv();
- ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
- zks.setupRequestProcessors();
- firstProcessor = new FirstProcessor(zks, null);
- zks.firstProcessor = firstProcessor;
-
- // setup session tracker
- zks.createSessionTracker();
- zks.startSessionTracker();
- return zks;
- }
-
- // Mock processor used in zookeeper server
- private class FirstProcessor extends PrepRequestProcessor {
- private volatile int countOfCloseSessionReq = 0;
-
- public FirstProcessor(ZooKeeperServer zks,
- RequestProcessor nextProcessor) {
- super(zks, nextProcessor);
- }
-
- @Override
- public void processRequest(Request request) {
- // check session close request
- if (request.type == OpCode.closeSession) {
- countOfCloseSessionReq++;
- latch.countDown();
- }
- }
-
- // return number of session expiry calls
- int getCountOfCloseSessionReq() {
- return countOfCloseSessionReq;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/ToStringTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/ToStringTest.java b/src/java/test/org/apache/zookeeper/server/ToStringTest.java
deleted file mode 100644
index f19469b..0000000
--- a/src/java/test/org/apache/zookeeper/server/ToStringTest.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.proto.SetDataRequest;
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * A misc place to verify toString methods - mainly to make sure they don't
- * fail.
- */
-public class ToStringTest extends ZKTestCase {
- /** Verify jute - which we've had particular problems with in the past
- * wrt null fields */
- @Test
- public void testJuteToString() {
- SetDataRequest req = new SetDataRequest(null, null, 0);
- Assert.assertNotSame("ERROR", req.toString());
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/test/org/apache/zookeeper/server/ZooKeeperServerMainTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/ZooKeeperServerMainTest.java b/src/java/test/org/apache/zookeeper/server/ZooKeeperServerMainTest.java
deleted file mode 100644
index 2ffb91e..0000000
--- a/src/java/test/org/apache/zookeeper/server/ZooKeeperServerMainTest.java
+++ /dev/null
@@ -1,615 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zookeeper.server;
-
-import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
-import static org.junit.Assert.fail;
-
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.jute.Record;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.PortAssignment;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZKTestCase;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.proto.ReplyHeader;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-import org.apache.zookeeper.server.persistence.Util;
-import org.apache.zookeeper.server.util.ZxidUtils;
-import org.apache.zookeeper.test.ClientBase;
-import org.apache.zookeeper.txn.SetDataTxn;
-import org.apache.zookeeper.txn.TxnHeader;
-import org.jboss.netty.channel.Channel;
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * Test stand-alone server.
- *
- */
-public class ZooKeeperServerMainTest extends ZKTestCase implements Watcher {
- protected static final Logger LOG =
- LoggerFactory.getLogger(ZooKeeperServerMainTest.class);
-
- public static class MainThread extends Thread {
- final File confFile;
- final TestZKSMain main;
- final File tmpDir;
- final File dataDir;
- final File logDir;
-
- public MainThread(int clientPort, boolean preCreateDirs) throws IOException {
- this(clientPort, preCreateDirs, ClientBase.createTmpDir());
- }
-
- public MainThread(int clientPort, boolean preCreateDirs, File tmpDir) throws IOException {
- super("Standalone server with clientPort:" + clientPort);
- this.tmpDir = tmpDir;
- confFile = new File(this.tmpDir, "zoo.cfg");
-
- FileWriter fwriter = new FileWriter(confFile);
- fwriter.write("tickTime=2000\n");
- fwriter.write("initLimit=10\n");
- fwriter.write("syncLimit=5\n");
-
- dataDir = new File(this.tmpDir, "data");
- logDir = new File(dataDir.toString() + "_txnlog");
- if (preCreateDirs) {
- if (!dataDir.mkdir()) {
- throw new IOException("unable to mkdir " + dataDir);
- }
- if (!logDir.mkdir()) {
- throw new IOException("unable to mkdir " + logDir);
- }
- }
-
- String dataDirPath = dataDir.toString();
- String logDirPath = logDir.toString();
-
- // Convert windows path to UNIX to avoid problems with "\"
- String osname = java.lang.System.getProperty("os.name");
- if (osname.toLowerCase().contains("windows")) {
- dataDirPath = dataDirPath.replace('\\', '/');
- logDirPath = logDirPath.replace('\\', '/');
- }
- fwriter.write("dataDir=" + dataDirPath + "\n");
- fwriter.write("dataLogDir=" + logDirPath + "\n");
- fwriter.write("clientPort=" + clientPort + "\n");
- fwriter.flush();
- fwriter.close();
-
- main = new TestZKSMain();
- }
-
- public void run() {
- String args[] = new String[1];
- args[0] = confFile.toString();
- try {
- main.initializeAndRun(args);
- } catch (Exception e) {
- // test will still fail even though we just log/ignore
- LOG.error("unexpected exception in run", e);
- }
- }
-
- public void shutdown() throws IOException {
- main.shutdown();
- }
-
- void deleteDirs() throws IOException{
- delete(tmpDir);
- }
-
- void delete(File f) throws IOException {
- if (f.isDirectory()) {
- for (File c : f.listFiles())
- delete(c);
- }
- if (!f.delete())
- // double check for the file existence
- if (f.exists()) {
- throw new IOException("Failed to delete file: " + f);
- }
- }
-
- ServerCnxnFactory getCnxnFactory() {
- return main.getCnxnFactory();
- }
- }
-
- public static class TestZKSMain extends ZooKeeperServerMain {
- public void shutdown() {
- super.shutdown();
- }
- }
-
- /**
- * Test case for https://issues.apache.org/jira/browse/ZOOKEEPER-2247.
- * Test to verify that even after non recoverable error (error while
- * writing transaction log) on ZooKeeper service will be available
- */
- @Test(timeout = 30000)
- public void testNonRecoverableError() throws Exception {
- ClientBase.setupTestEnv();
-
- final int CLIENT_PORT = PortAssignment.unique();
-
- MainThread main = new MainThread(CLIENT_PORT, true);
- main.start();
-
- Assert.assertTrue("waiting for server being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
- CONNECTION_TIMEOUT));
-
-
- ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT, this);
-
- zk.create("/foo1", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- Assert.assertEquals(new String(zk.getData("/foo1", null, null)), "foobar");
-
- // inject problem in server
- ZooKeeperServer zooKeeperServer = main.getCnxnFactory()
- .getZooKeeperServer();
- FileTxnSnapLog snapLog = zooKeeperServer.getTxnLogFactory();
- FileTxnSnapLog fileTxnSnapLogWithError = new FileTxnSnapLog(
- snapLog.getDataDir(), snapLog.getSnapDir()) {
- @Override
- public void commit() throws IOException {
- throw new IOException("Input/output error");
- }
- };
- ZKDatabase newDB = new ZKDatabase(fileTxnSnapLogWithError);
- zooKeeperServer.setZKDatabase(newDB);
-
- try {
- // do create operation, so that injected IOException is thrown
- zk.create("/foo2", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- fail("IOException is expected as error is injected in transaction log commit funtionality");
- } catch (Exception e) {
- // do nothing
- }
- zk.close();
- Assert.assertTrue("waiting for server down",
- ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT));
- fileTxnSnapLogWithError.close();
- main.shutdown();
- main.deleteDirs();
- }
-
- /**
- * 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);
- 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);
- 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);
- 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();
- }
-
- /**
- * Verify the ability to start a standalone server instance.
- */
- @Test
- public void testStandalone() throws Exception {
- ClientBase.setupTestEnv();
-
- final int CLIENT_PORT = PortAssignment.unique();
-
- MainThread main = new MainThread(CLIENT_PORT, true);
- main.start();
-
- Assert.assertTrue("waiting for server being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
- CONNECTION_TIMEOUT));
-
-
- ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT, this);
-
- zk.create("/foo", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- Assert.assertEquals(new String(zk.getData("/foo", null, null)), "foobar");
- zk.close();
-
- main.shutdown();
- main.join();
- main.deleteDirs();
-
- Assert.assertTrue("waiting for server down",
- ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT));
- }
-
- /**
- * Test verifies the auto creation of data dir and data log dir.
- */
- @Test(timeout = 30000)
- public void testAutoCreateDataLogDir() throws Exception {
- ClientBase.setupTestEnv();
- final int CLIENT_PORT = PortAssignment.unique();
-
- MainThread main = new MainThread(CLIENT_PORT, false);
- String args[] = new String[1];
- args[0] = main.confFile.toString();
- main.start();
-
- Assert.assertTrue("waiting for server being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
- CONNECTION_TIMEOUT));
-
- ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT, this);
-
- zk.create("/foo", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- Assert.assertEquals(new String(zk.getData("/foo", null, null)),
- "foobar");
- zk.close();
-
- main.shutdown();
- main.join();
- main.deleteDirs();
-
- Assert.assertTrue("waiting for server down", ClientBase
- .waitForServerDown("127.0.0.1:" + CLIENT_PORT,
- ClientBase.CONNECTION_TIMEOUT));
- }
-
- @Test
- public void testJMXRegistrationWithNIO() throws Exception {
- ClientBase.setupTestEnv();
- File tmpDir_1 = ClientBase.createTmpDir();
- ServerCnxnFactory server_1 = startServer(tmpDir_1);
- File tmpDir_2 = ClientBase.createTmpDir();
- ServerCnxnFactory server_2 = startServer(tmpDir_2);
-
- server_1.shutdown();
- server_2.shutdown();
-
- deleteFile(tmpDir_1);
- deleteFile(tmpDir_2);
- }
-
- @Test
- public void testJMXRegistrationWithNetty() throws Exception {
- String originalServerCnxnFactory = System
- .getProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
- System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY,
- NettyServerCnxnFactory.class.getName());
- try {
- ClientBase.setupTestEnv();
- File tmpDir_1 = ClientBase.createTmpDir();
- ServerCnxnFactory server_1 = startServer(tmpDir_1);
- File tmpDir_2 = ClientBase.createTmpDir();
- ServerCnxnFactory server_2 = startServer(tmpDir_2);
-
- server_1.shutdown();
- server_2.shutdown();
-
- deleteFile(tmpDir_1);
- deleteFile(tmpDir_2);
- } finally {
- // setting back
- if (originalServerCnxnFactory == null
- || originalServerCnxnFactory.isEmpty()) {
- System.clearProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
- } else {
- System.setProperty(
- ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY,
- originalServerCnxnFactory);
- }
- }
- }
-
- /**
- * Test case to verify that ZooKeeper server is able to shutdown properly
- * when there are pending request(s) in the RequestProcessor chain.
- *
- * {@link https://issues.apache.org/jira/browse/ZOOKEEPER-2347}
- */
- @Test(timeout = 30000)
- public void testRaceBetweenSyncFlushAndZKShutdown() throws Exception {
- File tmpDir = ClientBase.createTmpDir();
- File testDir = File.createTempFile("test", ".dir", tmpDir);
- testDir.delete();
-
- // Following are the sequence of steps to simulate the deadlock
- // situation - SyncRequestProcessor#shutdown holds a lock and waits on
- // FinalRequestProcessor to complete a pending operation, which in turn
- // also needs the ZooKeeperServer lock
-
- // 1. start zk server
- FileTxnSnapLog ftsl = new FileTxnSnapLog(testDir, testDir);
- final SimpleZooKeeperServer zkServer = new SimpleZooKeeperServer(ftsl);
- zkServer.startup();
- // 2. Wait for setting up request processor chain. At the end of setup,
- // it will add a mock request into the chain
- // 3. Also, waiting for FinalRequestProcessor to start processing request
- zkServer.waitForFinalProcessRequest();
- // 4. Above step ensures that there is a request in the processor chain.
- // Now invoke shutdown, which will acquire zks lock
- Thread shutdownThread = new Thread() {
- public void run() {
- zkServer.shutdown();
- };
- };
- shutdownThread.start();
- // 5. Wait for SyncRequestProcessor to trigger shutdown function.
- // This is to ensure that zks lock is acquired
- zkServer.waitForSyncReqProcessorShutdown();
- // 6. Now resume FinalRequestProcessor which in turn call
- // zks#decInProcess() function and tries to acquire zks lock.
- // This results in deadlock
- zkServer.resumeFinalProcessRequest();
- // 7. Waiting to finish server shutdown. Testing that
- // SyncRequestProcessor#shutdown holds a lock and waits on
- // FinalRequestProcessor to complete a pending operation, which in turn
- // also needs the ZooKeeperServer lock
- shutdownThread.join();
- }
-
- private class SimpleZooKeeperServer extends ZooKeeperServer {
- private SimpleSyncRequestProcessor syncProcessor;
- private SimpleFinalRequestProcessor finalProcessor;
-
- SimpleZooKeeperServer(FileTxnSnapLog ftsl) throws IOException {
- super(ftsl, 2000, 2000, 4000, null, new ZKDatabase(ftsl));
- }
-
- @Override
- protected void setupRequestProcessors() {
- finalProcessor = new SimpleFinalRequestProcessor(this);
- syncProcessor = new SimpleSyncRequestProcessor(this,
- finalProcessor);
- syncProcessor.start();
- firstProcessor = new PrepRequestProcessor(this, syncProcessor);
- ((PrepRequestProcessor) firstProcessor).start();
-
- // add request to the chain
- addRequestToSyncProcessor();
- }
-
- private void addRequestToSyncProcessor() {
- long zxid = ZxidUtils.makeZxid(3, 7);
- TxnHeader hdr = new TxnHeader(1, 1, zxid, 1,
- ZooDefs.OpCode.setData);
- Record txn = new SetDataTxn("/foo" + zxid, new byte[0], 1);
- byte[] buf;
- try {
- buf = Util.marshallTxnEntry(hdr, txn);
- } catch (IOException e) {
- LOG.error("IOException while adding request to SyncRequestProcessor", e);
- Assert.fail("IOException while adding request to SyncRequestProcessor!");
- return;
- }
- NettyServerCnxnFactory factory = new NettyServerCnxnFactory();
- final MockNettyServerCnxn nettyCnxn = new MockNettyServerCnxn(null,
- this, factory);
- Request req = new Request(nettyCnxn, 1, 1, ZooDefs.OpCode.setData,
- ByteBuffer.wrap(buf), null);
- req.hdr = hdr;
- req.txn = txn;
- syncProcessor.processRequest(req);
- }
-
- void waitForFinalProcessRequest() throws InterruptedException {
- Assert.assertTrue("Waiting for FinalRequestProcessor to start processing request",
- finalProcessor.waitForProcessRequestToBeCalled());
- }
-
- void waitForSyncReqProcessorShutdown() throws InterruptedException {
- Assert.assertTrue("Waiting for SyncRequestProcessor to shut down",
- syncProcessor.waitForShutdownToBeCalled());
- }
-
- void resumeFinalProcessRequest() throws InterruptedException {
- finalProcessor.resumeProcessRequest();
- }
- }
-
- private class MockNettyServerCnxn extends NettyServerCnxn {
- public MockNettyServerCnxn(Channel channel, ZooKeeperServer zks,
- NettyServerCnxnFactory factory) {
- super(null, null, factory);
- }
-
- @Override
- protected synchronized void updateStatsForResponse(long cxid, long zxid,
- String op, long start, long end) {
- return;
- }
-
- @Override
- public synchronized void sendResponse(ReplyHeader h, Record r,
- String tag) {
- return;
- }
- }
-
- private class SimpleFinalRequestProcessor extends FinalRequestProcessor {
- private CountDownLatch finalReqProcessCalled = new CountDownLatch(1);
- private CountDownLatch resumeFinalReqProcess = new CountDownLatch(1);
- private volatile boolean interrupted = false;
- public SimpleFinalRequestProcessor(ZooKeeperServer zks) {
- super(zks);
- }
-
- @Override
- public void processRequest(Request request) {
- finalReqProcessCalled.countDown();
- try {
- resumeFinalReqProcess.await(ClientBase.CONNECTION_TIMEOUT,
- TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- LOG.error("Interrupted while waiting to process request", e);
- interrupted = true; // Marked as interrupted
- resumeFinalReqProcess.countDown();
- return;
- }
- super.processRequest(request);
- }
-
- boolean waitForProcessRequestToBeCalled() throws InterruptedException {
- return finalReqProcessCalled.await(ClientBase.CONNECTION_TIMEOUT,
- TimeUnit.MILLISECONDS);
- }
-
- void resumeProcessRequest() throws InterruptedException {
- resumeFinalReqProcess.countDown();
- resumeFinalReqProcess.await(ClientBase.CONNECTION_TIMEOUT,
- TimeUnit.MILLISECONDS);
- Assert.assertFalse("Interrupted while waiting to process request",
- interrupted);
- }
- }
-
- private class SimpleSyncRequestProcessor extends SyncRequestProcessor {
- private final CountDownLatch shutdownCalled = new CountDownLatch(1);
-
- public SimpleSyncRequestProcessor(ZooKeeperServer zks,
- RequestProcessor nextProcessor) {
- super(zks, nextProcessor);
- }
-
- @Override
- public void shutdown() {
- shutdownCalled.countDown();
- super.shutdown();
- }
-
- boolean waitForShutdownToBeCalled() throws InterruptedException {
- return shutdownCalled.await(ClientBase.CONNECTION_TIMEOUT / 3,
- TimeUnit.MILLISECONDS);
- }
- }
-
- private void deleteFile(File f) throws IOException {
- if (f.isDirectory()) {
- for (File c : f.listFiles())
- deleteFile(c);
- }
- if (!f.delete())
- // double check for the file existence
- if (f.exists()) {
- throw new IOException("Failed to delete file: " + f);
- }
- }
-
- private ServerCnxnFactory startServer(File tmpDir) throws IOException,
- InterruptedException {
- final int CLIENT_PORT = PortAssignment.unique();
- ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
- ServerCnxnFactory f = ServerCnxnFactory.createFactory(CLIENT_PORT, -1);
- f.startup(zks);
- Assert.assertNotNull("JMX initialization failed!", zks.jmxServerBean);
- Assert.assertTrue("waiting for server being up",
- ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
- CONNECTION_TIMEOUT));
- return f;
- }
-
- public void process(WatchedEvent event) {
- // ignore for this test
- }
-}