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/24 09:30:59 UTC
[04/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch-3.5 - zookeeper-server
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiTransactionTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiTransactionTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiTransactionTest.java
new file mode 100644
index 0000000..fb864f5
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/MultiTransactionTest.java
@@ -0,0 +1,806 @@
+/**
+ * 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.test;
+
+import static org.junit.Assert.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.AsyncCallback.MultiCallback;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.Transaction;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.OpResult.CheckResult;
+import org.apache.zookeeper.OpResult.CreateResult;
+import org.apache.zookeeper.OpResult.DeleteResult;
+import org.apache.zookeeper.OpResult.ErrorResult;
+import org.apache.zookeeper.OpResult.SetDataResult;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.SyncRequestProcessor;
+import org.apache.zookeeper.ZKParameterized;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(Parameterized.class)
+@Parameterized.UseParametersRunnerFactory(ZKParameterized.RunnerFactory.class)
+public class MultiTransactionTest extends ClientBase {
+ private static final Logger LOG = LoggerFactory.getLogger(MultiTransactionTest.class);
+ private ZooKeeper zk;
+ private ZooKeeper zk_chroot;
+
+ private final boolean useAsync;
+
+ public MultiTransactionTest(boolean useAsync) {
+ this.useAsync = useAsync;
+ }
+
+ @Parameters
+ public static Collection<Object[]> configs() {
+ return Arrays.asList(new Object[][] {
+ { false }, { true },
+ });
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ SyncRequestProcessor.setSnapCount(150);
+ super.setUp();
+ zk = createClient();
+ }
+
+ static class MultiResult {
+ int rc;
+ List<OpResult> results;
+ boolean finished = false;
+ }
+
+ private List<OpResult> multi(ZooKeeper zk, Iterable<Op> ops)
+ throws KeeperException, InterruptedException {
+ if (useAsync) {
+ final MultiResult res = new MultiResult();
+ zk.multi(ops, new MultiCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<OpResult> opResults) {
+ synchronized (res) {
+ res.rc = rc;
+ res.results = opResults;
+ res.finished = true;
+ res.notifyAll();
+ }
+ }
+ }, null);
+ synchronized (res) {
+ while (!res.finished) {
+ res.wait();
+ }
+ }
+ if (KeeperException.Code.OK.intValue() != res.rc) {
+ KeeperException ke = KeeperException.create(KeeperException.Code.get(res.rc));
+ throw ke;
+ }
+ return res.results;
+ } else {
+ return zk.multi(ops);
+ }
+ }
+
+ private void multiHavingErrors(ZooKeeper zk, Iterable<Op> ops,
+ List<Integer> expectedResultCodes, String expectedErr)
+ throws KeeperException, InterruptedException {
+ if (useAsync) {
+ final MultiResult res = new MultiResult();
+ zk.multi(ops, new MultiCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<OpResult> opResults) {
+ synchronized (res) {
+ res.rc = rc;
+ res.results = opResults;
+ res.finished = true;
+ res.notifyAll();
+ }
+ }
+ }, null);
+ synchronized (res) {
+ while (!res.finished) {
+ res.wait();
+ }
+ }
+ for (int i = 0; i < res.results.size(); i++) {
+ OpResult opResult = res.results.get(i);
+ Assert.assertTrue("Did't recieve proper error response",
+ opResult instanceof ErrorResult);
+ ErrorResult errRes = (ErrorResult) opResult;
+ Assert.assertEquals("Did't recieve proper error code",
+ expectedResultCodes.get(i).intValue(), errRes.getErr());
+ }
+ } else {
+ try {
+ zk.multi(ops);
+ Assert.fail("Shouldn't have validated in ZooKeeper client!");
+ } catch (KeeperException e) {
+ Assert.assertEquals("Wrong exception", expectedErr, e.code()
+ .name());
+ } catch (IllegalArgumentException e) {
+ Assert.assertEquals("Wrong exception", expectedErr,
+ e.getMessage());
+ }
+ }
+ }
+
+ private List<OpResult> commit(Transaction txn)
+ throws KeeperException, InterruptedException {
+ if (useAsync) {
+ final MultiResult res = new MultiResult();
+ txn.commit(new MultiCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<OpResult> opResults) {
+ synchronized (res) {
+ res.rc = rc;
+ res.results = opResults;
+ res.finished = true;
+ res.notifyAll();
+ }
+ }
+ }, null);
+ synchronized (res) {
+ while (!res.finished) {
+ res.wait();
+ }
+ }
+ if (KeeperException.Code.OK.intValue() != res.rc) {
+ KeeperException ke = KeeperException.create(KeeperException.Code.get(res.rc));
+ throw ke;
+ }
+ return res.results;
+ } else {
+ return txn.commit();
+ }
+ }
+
+ /**
+ * Test verifies the multi calls with invalid znode path
+ */
+ @Test(timeout = 90000)
+ public void testInvalidPath() throws Exception {
+ List<Integer> expectedResultCodes = new ArrayList<Integer>();
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+ expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+ // create with CreateMode
+ List<Op> opList = Arrays.asList(Op.create("/multi0", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT), Op.create(
+ "/multi1/", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT), Op.create("/multi2", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
+ String expectedErr = "Path must not end with / character";
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+
+ // create with valid sequential flag
+ opList = Arrays.asList(Op.create("/multi0", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT), Op.create(
+ "multi1/", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.EPHEMERAL_SEQUENTIAL.toFlag()), Op.create("/multi2",
+ new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
+ expectedErr = "Path must start with / character";
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+
+ // check
+ opList = Arrays.asList(Op.check("/multi0", -1),
+ Op.check("/multi1/", 100), Op.check("/multi2", 5));
+ expectedErr = "Path must not end with / character";
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+
+ // delete
+ opList = Arrays.asList(Op.delete("/multi0", -1),
+ Op.delete("/multi1/", 100), Op.delete("/multi2", 5));
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+
+ // Multiple bad arguments
+ expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
+
+ // setdata
+ opList = Arrays.asList(Op.setData("/multi0", new byte[0], -1),
+ Op.setData("/multi1/", new byte[0], -1),
+ Op.setData("/multi2", new byte[0], -1),
+ Op.setData("multi3", new byte[0], -1));
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+ }
+
+ /**
+ * ZOOKEEPER-2052:
+ * Multi abort shouldn't have any side effect.
+ * We fix a bug in rollback and the following scenario should work:
+ * 1. multi delete abort because of not empty directory
+ * 2. ephemeral nodes under that directory are deleted
+ * 3. multi delete should succeed.
+ */
+ @Test
+ public void testMultiRollback() throws Exception {
+ zk.create("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ ZooKeeper epheZk = createClient();
+ epheZk.create("/foo/bar", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+
+ List<Op> opList = Arrays.asList(Op.delete("/foo", -1));
+ try {
+ zk.multi(opList);
+ Assert.fail("multi delete should failed for not empty directory");
+ } catch (KeeperException.NotEmptyException e) {
+ }
+
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ zk.exists("/foo/bar", new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getType() == Event.EventType.NodeDeleted){
+ latch.countDown();
+ }
+ }
+ });
+
+ epheZk.close();
+
+ latch.await();
+
+ try {
+ zk.getData("/foo/bar", false, null);
+ Assert.fail("ephemeral node should have been deleted");
+ } catch (KeeperException.NoNodeException e) {
+ }
+
+ zk.multi(opList);
+
+ try {
+ zk.getData("/foo", false, null);
+ Assert.fail("persistent node should have been deleted after multi");
+ } catch (KeeperException.NoNodeException e) {
+ }
+ }
+
+ /**
+ * Test verifies the multi calls with blank znode path
+ */
+ @Test(timeout = 90000)
+ public void testBlankPath() throws Exception {
+ List<Integer> expectedResultCodes = new ArrayList<Integer>();
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+ expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+ expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
+
+ // delete
+ String expectedErr = "Path cannot be null";
+ List<Op> opList = Arrays.asList(Op.delete("/multi0", -1),
+ Op.delete(null, 100), Op.delete("/multi2", 5),
+ Op.delete("", -1));
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+ }
+
+ /**
+ * Test verifies the multi.create with invalid createModeFlag
+ */
+ @Test(timeout = 90000)
+ public void testInvalidCreateModeFlag() throws Exception {
+ List<Integer> expectedResultCodes = new ArrayList<Integer>();
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+ expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
+ expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
+ .intValue());
+
+ int createModeFlag = 6789;
+ List<Op> opList = Arrays.asList(Op.create("/multi0", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT), Op.create(
+ "/multi1", new byte[0], Ids.OPEN_ACL_UNSAFE, createModeFlag),
+ Op.create("/multi2", new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT));
+ String expectedErr = KeeperException.Code.BADARGUMENTS.name();
+ multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
+ }
+
+ @Test
+ public void testChRootCreateDelete() throws Exception {
+ // creating the subtree for chRoot clients.
+ String chRoot = createNameSpace();
+ // Creating child using chRoot client.
+ zk_chroot = createClient(this.hostPort + chRoot);
+ Op createChild = Op.create("/myid", new byte[0],
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ multi(zk_chroot, Arrays.asList(createChild));
+
+ Assert.assertNotNull("zNode is not created under chroot:" + chRoot, zk
+ .exists(chRoot + "/myid", false));
+ Assert.assertNotNull("zNode is not created under chroot:" + chRoot,
+ zk_chroot.exists("/myid", false));
+ Assert.assertNull("zNode is created directly under '/', ignored configured chroot",
+ zk.exists("/myid", false));
+
+ // Deleting child using chRoot client.
+ Op deleteChild = Op.delete("/myid", 0);
+ multi(zk_chroot, Arrays.asList(deleteChild));
+ Assert.assertNull("zNode exists under chroot:" + chRoot, zk.exists(
+ chRoot + "/myid", false));
+ Assert.assertNull("zNode exists under chroot:" + chRoot, zk_chroot
+ .exists("/myid", false));
+ }
+
+ @Test
+ public void testChRootSetData() throws Exception {
+ // creating the subtree for chRoot clients.
+ String chRoot = createNameSpace();
+ // setData using chRoot client.
+ zk_chroot = createClient(this.hostPort + chRoot);
+ String[] names = {"/multi0", "/multi1", "/multi2"};
+ List<Op> ops = new ArrayList<Op>();
+
+ for (int i = 0; i < names.length; i++) {
+ ops.add(Op.create(names[i], new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT));
+ ops.add(Op.setData(names[i], names[i].getBytes(), 0));
+ }
+
+ multi(zk_chroot, ops) ;
+
+ for (int i = 0; i < names.length; i++) {
+ Assert.assertArrayEquals("zNode data not matching", names[i]
+ .getBytes(), zk_chroot.getData(names[i], false, null));
+ }
+ }
+
+ @Test
+ public void testChRootCheck() throws Exception {
+ // creating the subtree for chRoot clients.
+ String chRoot = createNameSpace();
+ // checking the child version using chRoot client.
+ zk_chroot = createClient(this.hostPort + chRoot);
+ String[] names = {"/multi0", "/multi1", "/multi2"};
+ List<Op> ops = new ArrayList<Op>();
+
+ for (int i = 0; i < names.length; i++) {
+ zk.create(chRoot + names[i], new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ }
+ for (int i = 0; i < names.length; i++) {
+ ops.add(Op.check(names[i], 0));
+ }
+ multi(zk_chroot, ops) ;
+ }
+
+ @Test
+ public void testChRootTransaction() throws Exception {
+ // creating the subtree for chRoot clients.
+ String chRoot = createNameSpace();
+ // checking the child version using chRoot client.
+ zk_chroot = createClient(this.hostPort + chRoot);
+ String childPath = "/myid";
+ Transaction transaction = zk_chroot.transaction();
+ transaction.create(childPath, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ transaction.check(childPath, 0);
+ transaction.setData(childPath, childPath.getBytes(), 0);
+ commit(transaction);
+
+ Assert.assertNotNull("zNode is not created under chroot:" + chRoot, zk
+ .exists(chRoot + childPath, false));
+ Assert.assertNotNull("zNode is not created under chroot:" + chRoot,
+ zk_chroot.exists(childPath, false));
+ Assert.assertNull("zNode is created directly under '/', ignored configured chroot",
+ zk.exists(childPath, false));
+ Assert.assertArrayEquals("zNode data not matching", childPath
+ .getBytes(), zk_chroot.getData(childPath, false, null));
+
+ transaction = zk_chroot.transaction();
+ // Deleting child using chRoot client.
+ transaction.delete(childPath, 1);
+ commit(transaction);
+
+ Assert.assertNull("chroot:" + chRoot + " exists after delete", zk
+ .exists(chRoot + "/myid", false));
+ Assert.assertNull("chroot:" + chRoot + " exists after delete",
+ zk_chroot.exists("/myid", false));
+ }
+
+ private String createNameSpace() throws InterruptedException,
+ KeeperException {
+ // creating the subtree for chRoot clients.
+ String chRoot = "/appsX";
+ Op createChRoot = Op.create(chRoot, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ multi(zk, Arrays.asList(createChRoot));
+ return chRoot;
+ }
+
+
+ @Test
+ public void testCreate() throws Exception {
+ multi(zk, Arrays.asList(
+ Op.create("/multi0", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.create("/multi1", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.create("/multi2", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
+ ));
+ zk.getData("/multi0", false, null);
+ zk.getData("/multi1", false, null);
+ zk.getData("/multi2", false, null);
+ }
+
+ @Test
+ public void testCreateDelete() throws Exception {
+
+ multi(zk, Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/multi", 0)
+ ));
+
+ // '/multi' should have been deleted
+ Assert.assertNull(zk.exists("/multi", null));
+ }
+
+ @Test
+ public void testInvalidVersion() throws Exception {
+
+ try {
+ multi(zk, Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/multi", 1)
+ ));
+ Assert.fail("delete /multi should have failed");
+ } catch (KeeperException e) {
+ /* PASS */
+ }
+ }
+
+ @Test
+ public void testNestedCreate() throws Exception {
+
+ multi(zk, Arrays.asList(
+ /* Create */
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.create("/multi/a", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.create("/multi/a/1", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+
+ /* Delete */
+ Op.delete("/multi/a/1", 0),
+ Op.delete("/multi/a", 0),
+ Op.delete("/multi", 0)
+ ));
+
+ //Verify tree deleted
+ Assert.assertNull(zk.exists("/multi/a/1", null));
+ Assert.assertNull(zk.exists("/multi/a", null));
+ Assert.assertNull(zk.exists("/multi", null));
+ }
+
+ @Test
+ public void testSetData() throws Exception {
+
+ String[] names = {"/multi0", "/multi1", "/multi2"};
+ List<Op> ops = new ArrayList<Op>();
+
+ for (int i = 0; i < names.length; i++) {
+ ops.add(Op.create(names[i], new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
+ ops.add(Op.setData(names[i], names[i].getBytes(), 0));
+ }
+
+ multi(zk, ops) ;
+
+ for (int i = 0; i < names.length; i++) {
+ Assert.assertArrayEquals(names[i].getBytes(), zk.getData(names[i], false, null));
+ }
+ }
+
+ @Test
+ public void testUpdateConflict() throws Exception {
+
+ Assert.assertNull(zk.exists("/multi", null));
+
+ try {
+ multi(zk, Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.setData("/multi", "X".getBytes(), 0),
+ Op.setData("/multi", "Y".getBytes(), 0)
+ ));
+ Assert.fail("Should have thrown a KeeperException for invalid version");
+ } catch (KeeperException e) {
+ //PASS
+ LOG.error("STACKTRACE: ", e);
+ }
+
+ Assert.assertNull(zk.exists("/multi", null));
+
+ //Updating version solves conflict -- order matters
+ multi(zk, Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.setData("/multi", "X".getBytes(), 0),
+ Op.setData("/multi", "Y".getBytes(), 1)
+ ));
+
+ Assert.assertArrayEquals(zk.getData("/multi", false, null), "Y".getBytes());
+ }
+
+ @Test
+ public void testDeleteUpdateConflict() throws Exception {
+
+ /* Delete of a node folowed by an update of the (now) deleted node */
+ try {
+ multi(zk, Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/multi", 0),
+ Op.setData("/multi", "Y".getBytes(), 0)
+ ));
+ Assert.fail("/multi should have been deleted so setData should have failed");
+ } catch (KeeperException e) {
+ /* PASS */
+ }
+
+ // '/multi' should never have been created as entire op should fail
+ Assert.assertNull(zk.exists("/multi", null)) ;
+ }
+
+ @Test
+ public void testGetResults() throws Exception {
+ /* Delete of a node folowed by an update of the (now) deleted node */
+ Iterable<Op> ops = Arrays.asList(
+ Op.create("/multi", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/multi", 0),
+ Op.setData("/multi", "Y".getBytes(), 0),
+ Op.create("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
+ );
+ List<OpResult> results = null;
+ if (useAsync) {
+ final MultiResult res = new MultiResult();
+ zk.multi(ops, new MultiCallback() {
+ @Override
+ public void processResult(int rc, String path, Object ctx,
+ List<OpResult> opResults) {
+ synchronized (res) {
+ res.rc = rc;
+ res.results = opResults;
+ res.finished = true;
+ res.notifyAll();
+ }
+ }
+ }, null);
+ synchronized (res) {
+ while (!res.finished) {
+ res.wait();
+ }
+ }
+ Assert.assertFalse("/multi should have been deleted so setData should have failed",
+ KeeperException.Code.OK.intValue() == res.rc);
+ Assert.assertNull(zk.exists("/multi", null));
+ results = res.results;
+ } else {
+ try {
+ zk.multi(ops);
+ Assert.fail("/multi should have been deleted so setData should have failed");
+ } catch (KeeperException e) {
+ // '/multi' should never have been created as entire op should fail
+ Assert.assertNull(zk.exists("/multi", null));
+ results = e.getResults();
+ }
+ }
+
+ Assert.assertNotNull(results);
+ for (OpResult r : results) {
+ LOG.info("RESULT==> {}", r);
+ if (r instanceof ErrorResult) {
+ ErrorResult er = (ErrorResult) r;
+ LOG.info("ERROR RESULT: {} ERR=>{}", er, KeeperException.Code.get(er.getErr()));
+ }
+ }
+ }
+
+ /**
+ * Exercise the equals methods of OpResult classes.
+ */
+ @Test
+ public void testOpResultEquals() {
+ opEquals(new CreateResult("/foo"),
+ new CreateResult("/foo"),
+ new CreateResult("nope"));
+
+ opEquals(new CreateResult("/foo"),
+ new CreateResult("/foo"),
+ new CreateResult("/foo", new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)));
+
+ opEquals(new CreateResult("/foo", new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new CreateResult("/foo", new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new CreateResult("nope", new Stat(11, 12, 13, 14, 15, 16, 17, 18, 19, 110, 111)));
+
+ opEquals(new CreateResult("/foo", new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new CreateResult("/foo", new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new CreateResult("/foo"));
+
+ opEquals(new CheckResult(),
+ new CheckResult(),
+ null);
+
+ opEquals(new SetDataResult(new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new SetDataResult(new Stat(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)),
+ new SetDataResult(new Stat(11, 12, 13, 14, 15, 16, 17, 18, 19, 110, 111)));
+
+ opEquals(new ErrorResult(1),
+ new ErrorResult(1),
+ new ErrorResult(2));
+
+ opEquals(new DeleteResult(),
+ new DeleteResult(),
+ null);
+
+ opEquals(new ErrorResult(1),
+ new ErrorResult(1),
+ new ErrorResult(2));
+ }
+
+ private void opEquals(OpResult expected, OpResult value, OpResult near) {
+ assertEquals(value, value);
+ assertFalse(value.equals(new Object()));
+ assertFalse(value.equals(near));
+ assertFalse(value.equals(value instanceof CreateResult ?
+ new ErrorResult(1) : new CreateResult("nope2")));
+ assertTrue(value.equals(expected));
+ }
+
+ @Test
+ public void testWatchesTriggered() throws KeeperException, InterruptedException {
+ HasTriggeredWatcher watcher = new HasTriggeredWatcher();
+ zk.getChildren("/", watcher);
+ multi(zk, Arrays.asList(
+ Op.create("/t", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/t", -1)
+ ));
+ assertTrue(watcher.triggered.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ public void testNoWatchesTriggeredForFailedMultiRequest() throws InterruptedException, KeeperException {
+ HasTriggeredWatcher watcher = new HasTriggeredWatcher();
+ zk.getChildren("/", watcher);
+ try {
+ multi(zk, Arrays.asList(
+ Op.create("/t", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT),
+ Op.delete("/nonexisting", -1)
+ ));
+ fail("expected previous multi op to fail!");
+ } catch (KeeperException.NoNodeException e) {
+ // expected
+ }
+ SyncCallback cb = new SyncCallback();
+ zk.sync("/", cb, null);
+
+ // by waiting for the callback we're assured that the event queue is flushed
+ cb.done.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
+ assertEquals(1, watcher.triggered.getCount());
+ }
+
+ @Test
+ public void testTransactionBuilder() throws Exception {
+ List<OpResult> results = commit(zk.transaction()
+ .create("/t1", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
+ .create("/t1/child", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
+ .create("/t2", null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL));
+ assertEquals(3, results.size());
+ for (OpResult r : results) {
+ CreateResult c = (CreateResult)r;
+ assertTrue(c.getPath().startsWith("/t"));
+ assertNotNull(c.toString());
+ }
+ assertNotNull(zk.exists("/t1", false));
+ assertNotNull(zk.exists("/t1/child", false));
+ assertNotNull(zk.exists("/t2", false));
+
+ results = commit(zk.transaction()
+ .check("/t1", 0)
+ .check("/t1/child", 0)
+ .check("/t2", 0));
+ assertEquals(3, results.size());
+ for (OpResult r : results) {
+ CheckResult c = (CheckResult)r;
+ assertNotNull(c.toString());
+ }
+
+ try {
+ results = commit(zk.transaction()
+ .check("/t1", 0)
+ .check("/t1/child", 0)
+ .check("/t2", 1));
+ fail();
+ } catch (KeeperException.BadVersionException e) {
+ // expected
+ }
+
+ results = commit(zk.transaction()
+ .check("/t1", 0)
+ .setData("/t1", new byte[0], 0));
+ assertEquals(2, results.size());
+ for (OpResult r : results) {
+ assertNotNull(r.toString());
+ }
+
+ try {
+ results = commit(zk.transaction()
+ .check("/t1", 1)
+ .setData("/t1", new byte[0], 2));
+ fail();
+ } catch (KeeperException.BadVersionException e) {
+ // expected
+ }
+
+ results = commit(zk.transaction()
+ .check("/t1", 1)
+ .check("/t1/child", 0)
+ .check("/t2", 0));
+ assertEquals(3, results.size());
+
+ results = commit(zk.transaction()
+ .delete("/t2", -1)
+ .delete("/t1/child", -1));
+ assertEquals(2, results.size());
+ for (OpResult r : results) {
+ DeleteResult d = (DeleteResult)r;
+ assertNotNull(d.toString());
+ }
+ assertNotNull(zk.exists("/t1", false));
+ assertNull(zk.exists("/t1/child", false));
+ assertNull(zk.exists("/t2", false));
+ }
+
+ private static class HasTriggeredWatcher implements Watcher {
+ private final CountDownLatch triggered = new CountDownLatch(1);
+
+ @Override
+ public void process(WatchedEvent event) {
+ triggered.countDown();
+ }
+ }
+ private static class SyncCallback implements AsyncCallback.VoidCallback {
+ private final CountDownLatch done = new CountDownLatch(1);
+
+ @Override
+ public void processResult(int rc, String path, Object ctx) {
+ done.countDown();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NIOConnectionFactoryFdLeakTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NIOConnectionFactoryFdLeakTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NIOConnectionFactoryFdLeakTest.java
new file mode 100644
index 0000000..d11f71e
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NIOConnectionFactoryFdLeakTest.java
@@ -0,0 +1,69 @@
+/**
+ * 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.test;
+
+import java.net.InetSocketAddress;
+
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.util.OSMXBean;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ZOOKEEPER-1620 - Acceptor and Selector thread don't call selector.close()
+ * causing fd leakage
+ */
+public class NIOConnectionFactoryFdLeakTest extends ZKTestCase {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(NIOConnectionFactoryFdLeakTest.class);
+
+ @Test
+ public void testFileDescriptorLeak() throws Exception {
+
+ OSMXBean osMbean = new OSMXBean();
+ if (osMbean.getUnix() != true) {
+ LOG.info("Unable to run test on non-unix system");
+ return;
+ }
+
+ long startFdCount = osMbean.getOpenFileDescriptorCount();
+ LOG.info("Start fdcount is: " + startFdCount);
+
+ for (int i = 0; i < 50; ++i) {
+ NIOServerCnxnFactory factory = new NIOServerCnxnFactory();
+ factory.configure(
+ new InetSocketAddress(
+ "127.0.0.1", PortAssignment.unique()), 10);
+ factory.start();
+ Thread.sleep(100);
+ factory.shutdown();
+ }
+
+ long endFdCount = osMbean.getOpenFileDescriptorCount();
+ LOG.info("End fdcount is: " + endFdCount);
+
+ // On my box, if selector.close() is not called fd diff is > 700.
+ Assert.assertTrue("Possible fd leakage",
+ ((endFdCount - startFdCount) < 50));
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteBase.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteBase.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteBase.java
new file mode 100644
index 0000000..684d67a
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteBase.java
@@ -0,0 +1,49 @@
+/**
+ * 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.test;
+
+import org.apache.zookeeper.ClientCnxnSocketNetty;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.server.NettyServerCnxnFactory;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Netty Client against Netty server
+ */
+@RunWith(Suite.class)
+public class NettyNettySuiteBase {
+ @BeforeClass
+ public static void setUp() {
+ System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY,
+ NettyServerCnxnFactory.class.getName());
+ System.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET,
+ ClientCnxnSocketNetty.class.getName());
+ System.setProperty("zookeeper.admin.enableServer", "false");
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ System.clearProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
+ System.clearProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteHammerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteHammerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteHammerTest.java
new file mode 100644
index 0000000..586e584
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteHammerTest.java
@@ -0,0 +1,30 @@
+/**
+ * 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.test;
+
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Netty Client against Netty server
+ */
+@Suite.SuiteClasses({
+ AsyncHammerTest.class
+})
+public class NettyNettySuiteHammerTest extends NettyNettySuiteBase {
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteTest.java
new file mode 100644
index 0000000..35710ba
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NettyNettySuiteTest.java
@@ -0,0 +1,38 @@
+/**
+ * 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.test;
+
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Netty Client against Netty server
+ */
+@Suite.SuiteClasses({
+ ACLTest.class,
+ AsyncOpsTest.class,
+ ChrootClientTest.class,
+ ClientTest.class,
+ FourLetterWordsTest.class,
+ NullDataTest.class,
+ ReconfigTest.class,
+ SessionTest.class,
+ WatcherTest.class
+})
+public class NettyNettySuiteTest extends NettyNettySuiteBase {
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteBase.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteBase.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteBase.java
new file mode 100644
index 0000000..5725c17
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteBase.java
@@ -0,0 +1,44 @@
+/**
+ * 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.test;
+
+import org.apache.zookeeper.server.NettyServerCnxnFactory;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Nio Client against Netty server
+ */
+@RunWith(Suite.class)
+public class NioNettySuiteBase {
+ @BeforeClass
+ public static void setUp() {
+ System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY,
+ NettyServerCnxnFactory.class.getName());
+ System.setProperty("zookeeper.admin.enableServer", "false");
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ System.clearProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteHammerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteHammerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteHammerTest.java
new file mode 100644
index 0000000..96e7b0c
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteHammerTest.java
@@ -0,0 +1,30 @@
+/**
+ * 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.test;
+
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Nio Client against Netty server
+ */
+@Suite.SuiteClasses({
+ AsyncHammerTest.class
+ })
+public class NioNettySuiteHammerTest extends NioNettySuiteBase {
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteTest.java
new file mode 100644
index 0000000..a1ef33d
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NioNettySuiteTest.java
@@ -0,0 +1,38 @@
+/**
+ * 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.test;
+
+import org.junit.runners.Suite;
+
+/**
+ * Run tests with: Nio Client against Netty server
+ */
+@Suite.SuiteClasses({
+ ACLTest.class,
+ AsyncOpsTest.class,
+ ChrootClientTest.class,
+ ClientTest.class,
+ FourLetterWordsTest.class,
+ NullDataTest.class,
+ ReconfigTest.class,
+ SessionTest.class,
+ WatcherTest.class
+ })
+public class NioNettySuiteTest extends NioNettySuiteBase {
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NonRecoverableErrorTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NonRecoverableErrorTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NonRecoverableErrorTest.java
new file mode 100644
index 0000000..31790d2
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NonRecoverableErrorTest.java
@@ -0,0 +1,185 @@
+/**
+ * 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.test;
+
+import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.UUID;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.apache.zookeeper.server.quorum.QuorumPeerTestBase;
+import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * This class tests the non-recoverable error behavior of quorum server.
+ */
+public class NonRecoverableErrorTest extends QuorumPeerTestBase {
+ private static final String NODE_PATH = "/noLeaderIssue";
+
+ /**
+ * 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), ZooKeeper is still available.
+ */
+ @Test(timeout = 30000)
+ public void testZooKeeperServiceAvailableOnLeader() throws Exception {
+ int SERVER_COUNT = 3;
+ final int clientPorts[] = new int[SERVER_COUNT];
+ StringBuilder sb = new StringBuilder();
+ String server;
+
+ for (int i = 0; i < SERVER_COUNT; i++) {
+ clientPorts[i] = PortAssignment.unique();
+ server = "server." + i + "=127.0.0.1:" + PortAssignment.unique()
+ + ":" + PortAssignment.unique() + ":participant;127.0.0.1:"
+ + clientPorts[i];
+ sb.append(server + "\n");
+ }
+ String currentQuorumCfgSection = sb.toString();
+ MainThread mt[] = new MainThread[SERVER_COUNT];
+
+ for (int i = 0; i < SERVER_COUNT; i++) {
+ mt[i] = new MainThread(i, clientPorts[i], currentQuorumCfgSection,
+ false);
+ mt[i].start();
+ }
+
+ // ensure server started
+ for (int i = 0; i < SERVER_COUNT; i++) {
+ Assert.assertTrue("waiting for server " + i + " being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i],
+ CONNECTION_TIMEOUT));
+ }
+
+ CountdownWatcher watcher = new CountdownWatcher();
+ ZooKeeper zk = new ZooKeeper("127.0.0.1:" + clientPorts[0],
+ ClientBase.CONNECTION_TIMEOUT, watcher);
+ watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
+
+ String data = "originalData";
+ zk.create(NODE_PATH, data.getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ // get information of current leader
+ QuorumPeer leader = getLeaderQuorumPeer(mt);
+ assertNotNull("Leader must have been elected by now", leader);
+
+ // inject problem in leader
+ FileTxnSnapLog snapLog = leader.getActiveServer().getTxnLogFactory();
+ FileTxnSnapLog fileTxnSnapLogWithError = new FileTxnSnapLog(
+ snapLog.getDataDir(), snapLog.getSnapDir()) {
+ @Override
+ public void commit() throws IOException {
+ throw new IOException("Input/output error");
+ }
+ };
+ ZKDatabase originalZKDatabase = leader.getActiveServer()
+ .getZKDatabase();
+ long leaderCurrentEpoch = leader.getCurrentEpoch();
+
+ ZKDatabase newDB = new ZKDatabase(fileTxnSnapLogWithError);
+ leader.getActiveServer().setZKDatabase(newDB);
+
+ try {
+ // do create operation, so that injected IOException is thrown
+ zk.create(uniqueZnode(), data.getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ fail("IOException is expected due to error injected to transaction log commit");
+ } catch (Exception e) {
+ // do nothing
+ }
+
+ // resetting watcher so that this watcher can be again used to ensure
+ // that the zkClient is able to re-establish connection with the
+ // newly elected zookeeper quorum.
+ watcher.reset();
+ waitForNewLeaderElection(leader, leaderCurrentEpoch);
+
+ // ensure server started, give enough time, so that new leader election
+ // takes place
+ for (int i = 0; i < SERVER_COUNT; i++) {
+ Assert.assertTrue("waiting for server " + i + " being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i],
+ CONNECTION_TIMEOUT));
+ }
+
+ // revert back the error
+ leader.getActiveServer().setZKDatabase(originalZKDatabase);
+
+ // verify that now ZooKeeper service is up and running
+ leader = getLeaderQuorumPeer(mt);
+ assertNotNull("New leader must have been elected by now", leader);
+
+ String uniqueNode = uniqueZnode();
+ watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
+ String createNode = zk.create(uniqueNode, data.getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ // if node is created successfully then it means that ZooKeeper service
+ // is available
+ assertEquals("Failed to create znode", uniqueNode, createNode);
+ zk.close();
+ // stop all severs
+ for (int i = 0; i < SERVER_COUNT; i++) {
+ mt[i].shutdown();
+ }
+ }
+
+ private void waitForNewLeaderElection(QuorumPeer peer,
+ long leaderCurrentEpoch) throws IOException, InterruptedException {
+ LOG.info("Waiting for new LE cycle..");
+ int count = 100; // giving a grace period of 10seconds
+ while (count > 0) {
+ if (leaderCurrentEpoch == peer.getCurrentEpoch()) {
+ Thread.sleep(100);
+ }
+ count--;
+ }
+ Assert.assertNotEquals("New LE cycle must have triggered",
+ leaderCurrentEpoch, peer.getCurrentEpoch());
+ }
+
+ 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;
+ }
+
+ private String uniqueZnode() {
+ UUID randomUUID = UUID.randomUUID();
+ String node = NODE_PATH + "/" + randomUUID.toString();
+ return node;
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/NullDataTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/NullDataTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NullDataTest.java
new file mode 100644
index 0000000..8f1a1bb
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/NullDataTest.java
@@ -0,0 +1,75 @@
+/**
+ * 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.test;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.AsyncCallback.StatCallback;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class NullDataTest extends ClientBase implements StatCallback {
+ String snapCount;
+ CountDownLatch cn = new CountDownLatch(1);
+
+ @Override
+ public void setUp() throws Exception {
+ // Change the snapcount to happen more often
+ snapCount = System.getProperty("zookeeper.snapCount", "1024");
+ System.setProperty("zookeeper.snapCount", "10");
+ super.setUp();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ System.setProperty("zookeeper.snapCount", snapCount);
+ super.tearDown();
+ }
+
+ @Test
+ public void testNullData() throws IOException,
+ InterruptedException, KeeperException {
+ String path = "/SIZE";
+ ZooKeeper zk = null;
+ zk = createClient();
+ try {
+ zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ // try sync zk exists
+ zk.exists(path, false);
+ zk.exists(path, false, this , null);
+ cn.await(10, TimeUnit.SECONDS);
+ Assert.assertSame(0L, cn.getCount());
+ } finally {
+ if(zk != null)
+ zk.close();
+ }
+
+ }
+
+ public void processResult(int rc, String path, Object ctx, Stat stat) {
+ cn.countDown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/OOMTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/OOMTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/OOMTest.java
new file mode 100644
index 0000000..465df98
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/OOMTest.java
@@ -0,0 +1,161 @@
+/**
+ * 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.test;
+
+import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OOMTest extends ZKTestCase implements Watcher {
+ @Test
+ public void testOOM() throws IOException, InterruptedException, KeeperException {
+ // This test takes too long tos run!
+ if (true)
+ return;
+ File tmpDir = ClientBase.createTmpDir();
+ // Grab some memory so that it is easier to cause an
+ // OOM condition;
+ ArrayList<byte[]> hog = new ArrayList<byte[]>();
+ while (true) {
+ try {
+ hog.add(new byte[1024 * 1024 * 2]);
+ } catch (OutOfMemoryError e) {
+ hog.remove(0);
+ break;
+ }
+ }
+ ClientBase.setupTestEnv();
+ ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+
+ final int PORT = PortAssignment.unique();
+ ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
+ f.startup(zks);
+ Assert.assertTrue("waiting for server up",
+ ClientBase.waitForServerUp("127.0.0.1:" + PORT,
+ CONNECTION_TIMEOUT));
+
+ System.err.println("OOM Stage 0");
+ utestPrep(PORT);
+ System.out.println("Free = " + Runtime.getRuntime().freeMemory()
+ + " total = " + Runtime.getRuntime().totalMemory() + " max = "
+ + Runtime.getRuntime().maxMemory());
+ System.err.println("OOM Stage 1");
+ for (int i = 0; i < 1000; i++) {
+ System.out.println(i);
+ utestExists(PORT);
+ }
+ System.out.println("Free = " + Runtime.getRuntime().freeMemory()
+ + " total = " + Runtime.getRuntime().totalMemory() + " max = "
+ + Runtime.getRuntime().maxMemory());
+ System.err.println("OOM Stage 2");
+ for (int i = 0; i < 1000; i++) {
+ System.out.println(i);
+ utestGet(PORT);
+ }
+ System.out.println("Free = " + Runtime.getRuntime().freeMemory()
+ + " total = " + Runtime.getRuntime().totalMemory() + " max = "
+ + Runtime.getRuntime().maxMemory());
+ System.err.println("OOM Stage 3");
+ for (int i = 0; i < 1000; i++) {
+ System.out.println(i);
+ utestChildren(PORT);
+ }
+ System.out.println("Free = " + Runtime.getRuntime().freeMemory()
+ + " total = " + Runtime.getRuntime().totalMemory() + " max = "
+ + Runtime.getRuntime().maxMemory());
+ hog.get(0)[0] = (byte) 1;
+
+ f.shutdown();
+ zks.shutdown();
+ Assert.assertTrue("waiting for server down",
+ ClientBase.waitForServerDown("127.0.0.1:" + PORT,
+ CONNECTION_TIMEOUT));
+ }
+
+ private void utestExists(int port)
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk =
+ new ZooKeeper("127.0.0.1:" + port, CONNECTION_TIMEOUT, this);
+ for (int i = 0; i < 10000; i++) {
+ zk.exists("/this/path/doesnt_exist!", true);
+ }
+ zk.close();
+ }
+
+ private void utestPrep(int port)
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk =
+ new ZooKeeper("127.0.0.1:" + port, CONNECTION_TIMEOUT, this);
+ for (int i = 0; i < 10000; i++) {
+ zk.create("/" + i, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ }
+ zk.close();
+ }
+
+ private void utestGet(int port)
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk =
+ new ZooKeeper("127.0.0.1:" + port, CONNECTION_TIMEOUT, this);
+ for (int i = 0; i < 10000; i++) {
+ Stat stat = new Stat();
+ zk.getData("/" + i, true, stat);
+ }
+ zk.close();
+ }
+
+ private void utestChildren(int port)
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk =
+ new ZooKeeper("127.0.0.1:" + port, CONNECTION_TIMEOUT, this);
+ for (int i = 0; i < 10000; i++) {
+ zk.getChildren("/" + i, true);
+ }
+ zk.close();
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
+ */
+ public void process(WatchedEvent event) {
+ System.err.println("Got event " + event.getType() + " "
+ + event.getState() + " " + event.getPath());
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/OSMXBeanTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/OSMXBeanTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/OSMXBeanTest.java
new file mode 100644
index 0000000..371c902
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/OSMXBeanTest.java
@@ -0,0 +1,71 @@
+/**
+ * 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.test;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.Before;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.zookeeper.server.util.OSMXBean;
+
+public class OSMXBeanTest extends ZKTestCase {
+
+ private OSMXBean osMbean;
+ private Long ofdc = 0L;
+ private Long mfdc = 0L;
+ protected static final Logger LOG = LoggerFactory.getLogger(OSMXBeanTest.class);
+
+ @Before
+ public void initialize() {
+ this.osMbean = new OSMXBean();
+ Assert.assertNotNull("Could not initialize OSMXBean object!", osMbean);
+ }
+
+ @Test
+ public final void testGetUnix() {
+ boolean isUnix = osMbean.getUnix();
+ if (!isUnix) {
+ LOG.info("Running in a Windows system! Output won't be printed!");
+ } else {
+ LOG.info("Running in a Unix or Linux system!");
+ }
+ }
+
+ @Test
+ public final void testGetOpenFileDescriptorCount() {
+ if (osMbean != null && osMbean.getUnix() == true) {
+ ofdc = osMbean.getOpenFileDescriptorCount();
+ LOG.info("open fdcount is: " + ofdc);
+ }
+ Assert.assertFalse("The number of open file descriptor is negative",(ofdc < 0));
+ }
+
+ @Test
+ public final void testGetMaxFileDescriptorCount() {
+ if (osMbean != null && osMbean.getUnix() == true) {
+ mfdc = osMbean.getMaxFileDescriptorCount();
+ LOG.info("max fdcount is: " + mfdc);
+ }
+ Assert.assertFalse("The max file descriptor number is negative",(mfdc < 0));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverHierarchicalQuorumTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverHierarchicalQuorumTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverHierarchicalQuorumTest.java
new file mode 100644
index 0000000..acd7d86
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverHierarchicalQuorumTest.java
@@ -0,0 +1,39 @@
+/* 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.test;
+
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.junit.Test;
+
+public class ObserverHierarchicalQuorumTest extends HierarchicalQuorumTest {
+ /**
+ * startServers(true) puts two observers into a 5 peer ensemble
+ */
+ void startServers() throws Exception {
+ startServers(true);
+ }
+
+ protected void shutdown(QuorumPeer qp) {
+ QuorumBase.shutdown(qp);
+ }
+
+ @Test
+ public void testHierarchicalQuorum() throws Throwable {
+ cht.runHammer(5, 10);
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverLETest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverLETest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverLETest.java
new file mode 100644
index 0000000..123ba0b
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverLETest.java
@@ -0,0 +1,76 @@
+/**
+ * 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.test;
+
+import static org.junit.Assert.*;
+
+import java.util.Arrays;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumStats;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ObserverLETest extends ZKTestCase {
+ final QuorumBase qb = new QuorumBase();
+ final ClientTest ct = new ClientTest();
+
+ @Before
+ public void establishThreeParticipantOneObserverEnsemble() throws Exception {
+ qb.setUp(true);
+ ct.hostPort = qb.hostPort;
+ ct.setUpAll();
+ qb.s5.shutdown();
+ }
+
+ @After
+ public void shutdownQuorum() throws Exception {
+ ct.tearDownAll();
+ qb.tearDown();
+ }
+
+ /**
+ * 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));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverQuorumHammerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverQuorumHammerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverQuorumHammerTest.java
new file mode 100644
index 0000000..03b4000
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverQuorumHammerTest.java
@@ -0,0 +1,39 @@
+/**
+ * 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.test;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ObserverQuorumHammerTest extends QuorumHammerTest {
+ public static final long CONNECTION_TIMEOUT = ClientTest.CONNECTION_TIMEOUT;
+
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ qb.setUp(true);
+ cht.hostPort = qb.hostPort;
+ cht.setUpAll();
+ }
+
+ @Test
+ public void testHammerBasic() throws Throwable {
+ cht.testHammerBasic();
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverTest.java
new file mode 100644
index 0000000..e9d8332
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverTest.java
@@ -0,0 +1,232 @@
+/**
+ * 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.test;
+
+import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException.ConnectionLossException;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooKeeper.States;
+import org.apache.zookeeper.server.quorum.QuorumPeerTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ObserverTest extends QuorumPeerTestBase implements Watcher{
+ protected static final Logger LOG =
+ LoggerFactory.getLogger(ObserverTest.class);
+
+ CountDownLatch latch;
+ ZooKeeper zk;
+ WatchedEvent lastEvent = null;
+
+ /**
+ * This test ensures two things:
+ * 1. That Observers can successfully proxy requests to the ensemble.
+ * 2. That Observers don't participate in leader elections.
+ * The second is tested by constructing an ensemble where a leader would
+ * be elected if and only if an Observer voted.
+ * @throws Exception
+ */
+ @Test
+ public void testObserver() throws Exception {
+ ClientBase.setupTestEnv();
+ // We expect two notifications before we want to continue
+ latch = new CountDownLatch(2);
+
+ final int PORT_QP1 = PortAssignment.unique();
+ final int PORT_QP2 = PortAssignment.unique();
+ final int PORT_OBS = PortAssignment.unique();
+ final int PORT_QP_LE1 = PortAssignment.unique();
+ final int PORT_QP_LE2 = PortAssignment.unique();
+ final int PORT_OBS_LE = PortAssignment.unique();
+
+ final int CLIENT_PORT_QP1 = PortAssignment.unique();
+ final int CLIENT_PORT_QP2 = PortAssignment.unique();
+ final int CLIENT_PORT_OBS = PortAssignment.unique();
+
+
+ String quorumCfgSection =
+ "server.1=127.0.0.1:" + (PORT_QP1)
+ + ":" + (PORT_QP_LE1) + ";" + CLIENT_PORT_QP1
+ + "\nserver.2=127.0.0.1:" + (PORT_QP2)
+ + ":" + (PORT_QP_LE2) + ";" + CLIENT_PORT_QP2
+ + "\nserver.3=127.0.0.1:"
+ + (PORT_OBS)+ ":" + (PORT_OBS_LE) + ":observer" + ";" + CLIENT_PORT_OBS;
+
+ MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+ MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSection);
+ MainThread q3 = new MainThread(3, CLIENT_PORT_OBS, quorumCfgSection);
+ q1.start();
+ q2.start();
+ q3.start();
+ Assert.assertTrue("waiting for server 1 being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+ CONNECTION_TIMEOUT));
+ Assert.assertTrue("waiting for server 2 being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
+ CONNECTION_TIMEOUT));
+ Assert.assertTrue("waiting for server 3 being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_OBS,
+ CONNECTION_TIMEOUT));
+
+ zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT_OBS,
+ ClientBase.CONNECTION_TIMEOUT, this);
+ zk.create("/obstest", "test".getBytes(),Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ // Assert that commands are getting forwarded correctly
+ Assert.assertEquals(new String(zk.getData("/obstest", null, null)), "test");
+
+ // Now check that other commands don't blow everything up
+ zk.sync("/", null, null);
+ zk.setData("/obstest", "test2".getBytes(), -1);
+ zk.getChildren("/", false);
+
+ Assert.assertEquals(zk.getState(), States.CONNECTED);
+
+ LOG.info("Shutting down server 2");
+ // Now kill one of the other real servers
+ q2.shutdown();
+
+ Assert.assertTrue("Waiting for server 2 to shut down",
+ ClientBase.waitForServerDown("127.0.0.1:"+CLIENT_PORT_QP2,
+ ClientBase.CONNECTION_TIMEOUT));
+
+ LOG.info("Server 2 down");
+
+ // Now the resulting ensemble shouldn't be quorate
+ latch.await();
+ Assert.assertNotSame("Client is still connected to non-quorate cluster",
+ KeeperState.SyncConnected,lastEvent.getState());
+
+ LOG.info("Latch returned");
+
+ try {
+ Assert.assertFalse("Shouldn't get a response when cluster not quorate!",
+ new String(zk.getData("/obstest", null, null)).equals("test"));
+ }
+ catch (ConnectionLossException c) {
+ LOG.info("Connection loss exception caught - ensemble not quorate (this is expected)");
+ }
+
+ latch = new CountDownLatch(1);
+
+ LOG.info("Restarting server 2");
+
+ // Bring it back
+ q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSection);
+ q2.start();
+
+ LOG.info("Waiting for server 2 to come up");
+ Assert.assertTrue("waiting for server 2 being up",
+ ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
+ CONNECTION_TIMEOUT));
+
+ LOG.info("Server 2 started, waiting for latch");
+
+ latch.await();
+ // It's possible our session expired - but this is ok, shows we
+ // were able to talk to the ensemble
+ Assert.assertTrue("Client didn't reconnect to quorate ensemble (state was" +
+ lastEvent.getState() + ")",
+ (KeeperState.SyncConnected==lastEvent.getState() ||
+ KeeperState.Expired==lastEvent.getState()));
+
+ LOG.info("Shutting down all servers");
+
+ q1.shutdown();
+ q2.shutdown();
+ q3.shutdown();
+
+ LOG.info("Closing zk client");
+
+ zk.close();
+ Assert.assertTrue("Waiting for server 1 to shut down",
+ ClientBase.waitForServerDown("127.0.0.1:"+CLIENT_PORT_QP1,
+ ClientBase.CONNECTION_TIMEOUT));
+ Assert.assertTrue("Waiting for server 2 to shut down",
+ ClientBase.waitForServerDown("127.0.0.1:"+CLIENT_PORT_QP2,
+ ClientBase.CONNECTION_TIMEOUT));
+ Assert.assertTrue("Waiting for server 3 to shut down",
+ ClientBase.waitForServerDown("127.0.0.1:"+CLIENT_PORT_OBS,
+ ClientBase.CONNECTION_TIMEOUT));
+
+ }
+
+ /**
+ * Implementation of watcher interface.
+ */
+ public void process(WatchedEvent event) {
+ lastEvent = event;
+ latch.countDown();
+ LOG.info("Latch got event :: " + event);
+ }
+
+ /**
+ * This test ensures that an Observer does not elect itself as a leader, or
+ * indeed come up properly, if it is the lone member of an ensemble.
+ * @throws Exception
+ */
+ @Test
+ public void testObserverOnly() throws Exception {
+ ClientBase.setupTestEnv();
+ final int CLIENT_PORT_QP1 = PortAssignment.unique();
+
+ String quorumCfgSection =
+ "server.1=127.0.0.1:" + (PortAssignment.unique())
+ + ":" + (PortAssignment.unique()) + ":observer;" + CLIENT_PORT_QP1 + "\n";
+
+ MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+ q1.start();
+ q1.join(ClientBase.CONNECTION_TIMEOUT);
+ Assert.assertFalse(q1.isAlive());
+ }
+
+ /**
+ * Ensure that observer only comes up when a proper ensemble is configured.
+ * (and will not come up with standalone server).
+ */
+ @Test
+ public void testObserverWithStandlone() throws Exception {
+ ClientBase.setupTestEnv();
+ final int CLIENT_PORT_QP1 = PortAssignment.unique();
+
+ String quorumCfgSection =
+ "server.1=127.0.0.1:" + (PortAssignment.unique())
+ + ":" + (PortAssignment.unique()) + ":observer\n"
+ + "server.2=127.0.0.1:" + (PortAssignment.unique())
+ + ":" + (PortAssignment.unique()) + "\npeerType=observer\n";
+
+ MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+ q1.start();
+ q1.join(ClientBase.CONNECTION_TIMEOUT);
+ Assert.assertFalse(q1.isAlive());
+ }
+
+}