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:31:03 UTC
[08/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/ClientTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientTest.java
new file mode 100644
index 0000000..6373bb3
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientTest.java
@@ -0,0 +1,870 @@
+/**
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.KeeperException.InvalidACLException;
+import org.apache.zookeeper.TestableZooKeeper;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooDefs.Perms;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.ExistsRequest;
+import org.apache.zookeeper.proto.ExistsResponse;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.server.PrepRequestProcessor;
+import org.apache.zookeeper.server.util.OSMXBean;
+import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClientTest extends ClientBase {
+ protected static final Logger LOG = LoggerFactory.getLogger(ClientTest.class);
+ private boolean skipACL = System.getProperty("zookeeper.skipACL", "no").equals("yes");
+
+ /** Verify that pings are sent, keeping the "idle" client alive */
+ @Test
+ public void testPing() throws Exception {
+ ZooKeeper zkIdle = null;
+ ZooKeeper zkWatchCreator = null;
+ try {
+ CountdownWatcher watcher = new CountdownWatcher();
+ zkIdle = createClient(watcher, hostPort, 10000);
+
+ zkWatchCreator = createClient();
+
+ for (int i = 0; i < 10; i++) {
+ zkWatchCreator.create("/" + i, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ }
+ for (int i = 0; i < 10; i++) {
+ zkIdle.exists("/" + i, true);
+ }
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(1000);
+ zkWatchCreator.delete("/" + i, -1);
+ }
+ // The bug will manifest itself here because zkIdle will expire
+ zkIdle.exists("/0", false);
+ } finally {
+ if (zkIdle != null) {
+ zkIdle.close();
+ }
+ if (zkWatchCreator != null) {
+ zkWatchCreator.close();
+ }
+ }
+ }
+
+ @Test
+ public void testClientwithoutWatcherObj() throws IOException,
+ InterruptedException, KeeperException
+ {
+ performClientTest(false);
+ }
+
+ @Test
+ public void testClientWithWatcherObj() throws IOException,
+ InterruptedException, KeeperException
+ {
+ performClientTest(true);
+ }
+
+ /** Exercise the testable functions, verify tostring, etc... */
+ @Test
+ public void testTestability() throws Exception {
+ TestableZooKeeper zk = createClient();
+ try {
+ LOG.info("{}",zk.testableLocalSocketAddress());
+ LOG.info("{}",zk.testableRemoteSocketAddress());
+ LOG.info("{}",zk.toString());
+ } finally {
+ zk.close(CONNECTION_TIMEOUT);
+ LOG.info("{}",zk.testableLocalSocketAddress());
+ LOG.info("{}",zk.testableRemoteSocketAddress());
+ LOG.info("{}",zk.toString());
+ }
+ }
+
+ @Test
+ public void testACLs() throws Exception {
+ ZooKeeper zk = null;
+ try {
+ zk = createClient();
+ try {
+ zk.create("/acltest", new byte[0], Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+ Assert.fail("Should have received an invalid acl error");
+ } catch(InvalidACLException e) {
+ LOG.info("Test successful, invalid acl received : "
+ + e.getMessage());
+ }
+ try {
+ ArrayList<ACL> testACL = new ArrayList<ACL>();
+ testACL.add(new ACL(Perms.ALL | Perms.ADMIN, Ids.AUTH_IDS));
+ testACL.add(new ACL(Perms.ALL | Perms.ADMIN, new Id("ip", "127.0.0.1/8")));
+ zk.create("/acltest", new byte[0], testACL, CreateMode.PERSISTENT);
+ Assert.fail("Should have received an invalid acl error");
+ } catch(InvalidACLException e) {
+ LOG.info("Test successful, invalid acl received : "
+ + e.getMessage());
+ }
+ try {
+ ArrayList<ACL> testACL = new ArrayList<ACL>();
+ testACL.add(new ACL(Perms.ALL | Perms.ADMIN, new Id()));
+ zk.create("/nullidtest", new byte[0], testACL, CreateMode.PERSISTENT);
+ Assert.fail("Should have received an invalid acl error");
+ } catch(InvalidACLException e) {
+ LOG.info("Test successful, invalid acl received : "
+ + e.getMessage());
+ }
+ zk.addAuthInfo("digest", "ben:passwd".getBytes());
+ ArrayList<ACL> testACL = new ArrayList<ACL>();
+ testACL.add(new ACL(Perms.ALL, new Id("auth","")));
+ testACL.add(new ACL(Perms.WRITE, new Id("ip", "127.0.0.1")));
+ zk.create("/acltest", new byte[0], testACL, CreateMode.PERSISTENT);
+ zk.close();
+ zk = createClient();
+ zk.addAuthInfo("digest", "ben:passwd2".getBytes());
+ if (skipACL) {
+ try {
+ zk.getData("/acltest", false, null);
+ } catch (KeeperException e) {
+ Assert.fail("Badauth reads should succeed with skipACL.");
+ }
+ } else {
+ try {
+ zk.getData("/acltest", false, null);
+ Assert.fail("Should have received a permission error");
+ } catch (KeeperException e) {
+ Assert.assertEquals(Code.NOAUTH, e.code());
+ }
+ }
+ zk.addAuthInfo("digest", "ben:passwd".getBytes());
+ zk.getData("/acltest", false, null);
+ zk.setACL("/acltest", Ids.OPEN_ACL_UNSAFE, -1);
+ zk.close();
+ zk = createClient();
+ zk.getData("/acltest", false, null);
+ List<ACL> acls = zk.getACL("/acltest", new Stat());
+ Assert.assertEquals(1, acls.size());
+ Assert.assertEquals(Ids.OPEN_ACL_UNSAFE, acls);
+
+ // The stat parameter should be optional.
+ acls = zk.getACL("/acltest", null);
+ Assert.assertEquals(1, acls.size());
+ Assert.assertEquals(Ids.OPEN_ACL_UNSAFE, acls);
+
+ zk.close();
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+ }
+
+ @Test
+ public void testNullAuthId() throws Exception {
+ ZooKeeper zk = null;
+ try {
+ zk = createClient();
+ zk.addAuthInfo("digest", "ben:passwd".getBytes());
+ ArrayList<ACL> testACL = new ArrayList<ACL>();
+ testACL.add(new ACL(Perms.ALL, new Id("auth", null)));
+ zk.create("/acltest", new byte[0], testACL, CreateMode.PERSISTENT);
+ zk.close();
+ zk = createClient();
+ zk.addAuthInfo("digest", "ben:passwd2".getBytes());
+ if (skipACL) {
+ try {
+ zk.getData("/acltest", false, null);
+ } catch (KeeperException e) {
+ Assert.fail("Badauth reads should succeed with skipACL.");
+ }
+ } else {
+ try {
+ zk.getData("/acltest", false, null);
+ Assert.fail("Should have received a permission error");
+ } catch (KeeperException e) {
+ Assert.assertEquals(Code.NOAUTH, e.code());
+ }
+ }
+ zk.addAuthInfo("digest", "ben:passwd".getBytes());
+ zk.getData("/acltest", false, null);
+ zk.setACL("/acltest", Ids.OPEN_ACL_UNSAFE, -1);
+ zk.close();
+ zk = createClient();
+ zk.getData("/acltest", false, null);
+ List<ACL> acls = zk.getACL("/acltest", new Stat());
+ Assert.assertEquals(1, acls.size());
+ Assert.assertEquals(Ids.OPEN_ACL_UNSAFE, acls);
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+ }
+
+ private class MyWatcher extends CountdownWatcher {
+ LinkedBlockingQueue<WatchedEvent> events =
+ new LinkedBlockingQueue<WatchedEvent>();
+
+ public void process(WatchedEvent event) {
+ super.process(event);
+ if (event.getType() != EventType.None) {
+ try {
+ events.put(event);
+ } catch (InterruptedException e) {
+ LOG.warn("ignoring interrupt during event.put");
+ }
+ }
+ }
+ }
+
+ /**
+ * Register multiple watchers and verify that they all get notified and
+ * in the right order.
+ */
+ @Test
+ public void testMutipleWatcherObjs()
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk = createClient(new CountdownWatcher(), hostPort);
+ try {
+ MyWatcher watchers[] = new MyWatcher[100];
+ MyWatcher watchers2[] = new MyWatcher[watchers.length];
+ for (int i = 0; i < watchers.length; i++) {
+ watchers[i] = new MyWatcher();
+ watchers2[i] = new MyWatcher();
+ zk.create("/foo-" + i, ("foodata" + i).getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ }
+ Stat stat = new Stat();
+
+ //
+ // test get/exists with single set of watchers
+ // get all, then exists all
+ //
+ for (int i = 0; i < watchers.length; i++) {
+ Assert.assertNotNull(zk.getData("/foo-" + i, watchers[i], stat));
+ }
+ for (int i = 0; i < watchers.length; i++) {
+ Assert.assertNotNull(zk.exists("/foo-" + i, watchers[i]));
+ }
+ // trigger the watches
+ for (int i = 0; i < watchers.length; i++) {
+ zk.setData("/foo-" + i, ("foodata2-" + i).getBytes(), -1);
+ zk.setData("/foo-" + i, ("foodata3-" + i).getBytes(), -1);
+ }
+ for (int i = 0; i < watchers.length; i++) {
+ WatchedEvent event =
+ watchers[i].events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/foo-" + i, event.getPath());
+ Assert.assertEquals(EventType.NodeDataChanged, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+
+ // small chance that an unexpected message was delivered
+ // after this check, but we would catch that next time
+ // we check events
+ Assert.assertEquals(0, watchers[i].events.size());
+ }
+
+ //
+ // test get/exists with single set of watchers
+ // get/exists together
+ //
+ for (int i = 0; i < watchers.length; i++) {
+ Assert.assertNotNull(zk.getData("/foo-" + i, watchers[i], stat));
+ Assert.assertNotNull(zk.exists("/foo-" + i, watchers[i]));
+ }
+ // trigger the watches
+ for (int i = 0; i < watchers.length; i++) {
+ zk.setData("/foo-" + i, ("foodata4-" + i).getBytes(), -1);
+ zk.setData("/foo-" + i, ("foodata5-" + i).getBytes(), -1);
+ }
+ for (int i = 0; i < watchers.length; i++) {
+ WatchedEvent event =
+ watchers[i].events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/foo-" + i, event.getPath());
+ Assert.assertEquals(EventType.NodeDataChanged, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+
+ // small chance that an unexpected message was delivered
+ // after this check, but we would catch that next time
+ // we check events
+ Assert.assertEquals(0, watchers[i].events.size());
+ }
+
+ //
+ // test get/exists with two sets of watchers
+ //
+ for (int i = 0; i < watchers.length; i++) {
+ Assert.assertNotNull(zk.getData("/foo-" + i, watchers[i], stat));
+ Assert.assertNotNull(zk.exists("/foo-" + i, watchers2[i]));
+ }
+ // trigger the watches
+ for (int i = 0; i < watchers.length; i++) {
+ zk.setData("/foo-" + i, ("foodata6-" + i).getBytes(), -1);
+ zk.setData("/foo-" + i, ("foodata7-" + i).getBytes(), -1);
+ }
+ for (int i = 0; i < watchers.length; i++) {
+ WatchedEvent event =
+ watchers[i].events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/foo-" + i, event.getPath());
+ Assert.assertEquals(EventType.NodeDataChanged, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+
+ // small chance that an unexpected message was delivered
+ // after this check, but we would catch that next time
+ // we check events
+ Assert.assertEquals(0, watchers[i].events.size());
+
+ // watchers2
+ WatchedEvent event2 =
+ watchers2[i].events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/foo-" + i, event2.getPath());
+ Assert.assertEquals(EventType.NodeDataChanged, event2.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event2.getState());
+
+ // small chance that an unexpected message was delivered
+ // after this check, but we would catch that next time
+ // we check events
+ Assert.assertEquals(0, watchers2[i].events.size());
+ }
+
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+ }
+
+ private void performClientTest(boolean withWatcherObj)
+ throws IOException, InterruptedException, KeeperException
+ {
+ ZooKeeper zk = null;
+ try {
+ MyWatcher watcher = new MyWatcher();
+ zk = createClient(watcher, hostPort);
+ LOG.info("Before create /benwashere");
+ zk.create("/benwashere", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ LOG.info("After create /benwashere");
+ try {
+ zk.setData("/benwashere", "hi".getBytes(), 57);
+ Assert.fail("Should have gotten BadVersion exception");
+ } catch(KeeperException.BadVersionException e) {
+ // expected that
+ } catch (KeeperException e) {
+ Assert.fail("Should have gotten BadVersion exception");
+ }
+ LOG.info("Before delete /benwashere");
+ zk.delete("/benwashere", 0);
+ LOG.info("After delete /benwashere");
+ zk.close();
+ //LOG.info("Closed client: " + zk.describeCNXN());
+ Thread.sleep(2000);
+
+ zk = createClient(watcher, hostPort);
+ //LOG.info("Created a new client: " + zk.describeCNXN());
+ LOG.info("Before delete /");
+
+ try {
+ zk.delete("/", -1);
+ Assert.fail("deleted root!");
+ } catch(KeeperException.BadArgumentsException e) {
+ // good, expected that
+ }
+ Stat stat = new Stat();
+ // Test basic create, ls, and getData
+ zk.create("/pat", "Pat was here".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ LOG.info("Before create /ben");
+ zk.create("/pat/ben", "Ben was here".getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ LOG.info("Before getChildren /pat");
+ List<String> children = zk.getChildren("/pat", false);
+ Assert.assertEquals(1, children.size());
+ Assert.assertEquals("ben", children.get(0));
+ List<String> children2 = zk.getChildren("/pat", false, null);
+ Assert.assertEquals(children, children2);
+ String value = new String(zk.getData("/pat/ben", false, stat));
+ Assert.assertEquals("Ben was here", value);
+ // Test stat and watch of non existent node
+
+ try {
+ if (withWatcherObj) {
+ Assert.assertEquals(null, zk.exists("/frog", watcher));
+ } else {
+ Assert.assertEquals(null, zk.exists("/frog", true));
+ }
+ LOG.info("Comment: asseting passed for frog setting /");
+ } catch (KeeperException.NoNodeException e) {
+ // OK, expected that
+ }
+ zk.create("/frog", "hi".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ // the first poll is just a session delivery
+ LOG.info("Comment: checking for events length "
+ + watcher.events.size());
+ WatchedEvent event = watcher.events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/frog", event.getPath());
+ Assert.assertEquals(EventType.NodeCreated, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+ // Test child watch and create with sequence
+ zk.getChildren("/pat/ben", true);
+ for (int i = 0; i < 10; i++) {
+ zk.create("/pat/ben/" + i + "-", Integer.toString(i).getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ }
+ children = zk.getChildren("/pat/ben", false);
+ Collections.sort(children);
+ Assert.assertEquals(10, children.size());
+ for (int i = 0; i < 10; i++) {
+ final String name = children.get(i);
+ Assert.assertTrue("starts with -", name.startsWith(i + "-"));
+ byte b[];
+ if (withWatcherObj) {
+ b = zk.getData("/pat/ben/" + name, watcher, stat);
+ } else {
+ b = zk.getData("/pat/ben/" + name, true, stat);
+ }
+ Assert.assertEquals(Integer.toString(i), new String(b));
+ zk.setData("/pat/ben/" + name, "new".getBytes(),
+ stat.getVersion());
+ if (withWatcherObj) {
+ stat = zk.exists("/pat/ben/" + name, watcher);
+ } else {
+ stat = zk.exists("/pat/ben/" + name, true);
+ }
+ zk.delete("/pat/ben/" + name, stat.getVersion());
+ }
+ event = watcher.events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/pat/ben", event.getPath());
+ Assert.assertEquals(EventType.NodeChildrenChanged, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+ for (int i = 0; i < 10; i++) {
+ event = watcher.events.poll(10, TimeUnit.SECONDS);
+ final String name = children.get(i);
+ Assert.assertEquals("/pat/ben/" + name, event.getPath());
+ Assert.assertEquals(EventType.NodeDataChanged, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+ event = watcher.events.poll(10, TimeUnit.SECONDS);
+ Assert.assertEquals("/pat/ben/" + name, event.getPath());
+ Assert.assertEquals(EventType.NodeDeleted, event.getType());
+ Assert.assertEquals(KeeperState.SyncConnected, event.getState());
+ }
+ zk.create("/good\u0040path", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ zk.create("/duplicate", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ try {
+ zk.create("/duplicate", "".getBytes(), Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ Assert.fail("duplicate create allowed");
+ } catch(KeeperException.NodeExistsException e) {
+ // OK, expected that
+ }
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+ }
+
+ // Test that sequential filenames are being created correctly,
+ // with 0-padding in the filename
+ @Test
+ public void testSequentialNodeNames()
+ throws IOException, InterruptedException, KeeperException
+ {
+ String path = "/SEQUENCE";
+ String file = "TEST";
+ String filepath = path + "/" + file;
+
+ ZooKeeper zk = null;
+ try {
+ zk = createClient();
+ zk.create(path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create(filepath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ List<String> children = zk.getChildren(path, false);
+ Assert.assertEquals(1, children.size());
+ Assert.assertEquals(file + "0000000000", children.get(0));
+
+ zk.create(filepath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ children = zk.getChildren(path, false);
+ Assert.assertEquals(2, children.size());
+ Assert.assertTrue("contains child 1", children.contains(file + "0000000001"));
+
+ zk.create(filepath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ children = zk.getChildren(path, false);
+ Assert.assertEquals(3, children.size());
+ Assert.assertTrue("contains child 2",
+ children.contains(file + "0000000002"));
+
+ // The pattern is holding so far. Let's run the counter a bit
+ // to be sure it continues to spit out the correct answer
+ for(int i = children.size(); i < 105; i++)
+ zk.create(filepath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+
+ children = zk.getChildren(path, false);
+ Assert.assertTrue("contains child 104",
+ children.contains(file + "0000000104"));
+
+ }
+ finally {
+ if(zk != null)
+ zk.close();
+ }
+ }
+
+ // Test that data provided when
+ // creating sequential nodes is stored properly
+ @Test
+ public void testSequentialNodeData() throws Exception {
+ ZooKeeper zk= null;
+ String queue_handle = "/queue";
+ try {
+ zk = createClient();
+
+ zk.create(queue_handle, new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ zk.create(queue_handle + "/element", "0".getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ zk.create(queue_handle + "/element", "1".getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
+ List<String> children = zk.getChildren(queue_handle, true);
+ Assert.assertEquals(children.size(), 2);
+ String child1 = children.get(0);
+ String child2 = children.get(1);
+ int compareResult = child1.compareTo(child2);
+ Assert.assertNotSame(compareResult, 0);
+ if (compareResult < 0) {
+ } else {
+ String temp = child1;
+ child1 = child2;
+ child2 = temp;
+ }
+ String child1data = new String(zk.getData(queue_handle
+ + "/" + child1, false, null));
+ String child2data = new String(zk.getData(queue_handle
+ + "/" + child2, false, null));
+ Assert.assertEquals(child1data, "0");
+ Assert.assertEquals(child2data, "1");
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+
+ }
+
+ @Test
+ public void testLargeNodeData() throws Exception {
+ ZooKeeper zk= null;
+ String queue_handle = "/large";
+ try {
+ zk = createClient();
+
+ zk.create(queue_handle, new byte[500000], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ } finally {
+ if (zk != null) {
+ zk.close();
+ }
+ }
+
+ }
+
+ private void verifyCreateFails(String path, ZooKeeper zk) throws Exception {
+ try {
+ zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ } catch (IllegalArgumentException e) {
+ // this is good
+ return;
+ }
+ Assert.fail("bad path \"" + path + "\" not caught");
+ }
+
+ // Test that the path string is validated
+ @Test
+ public void testPathValidation() throws Exception {
+ ZooKeeper zk = createClient();
+
+ verifyCreateFails(null, zk);
+ verifyCreateFails("", zk);
+ verifyCreateFails("//", zk);
+ verifyCreateFails("///", zk);
+ verifyCreateFails("////", zk);
+ verifyCreateFails("/.", zk);
+ verifyCreateFails("/..", zk);
+ verifyCreateFails("/./", zk);
+ verifyCreateFails("/../", zk);
+ verifyCreateFails("/foo/./", zk);
+ verifyCreateFails("/foo/../", zk);
+ verifyCreateFails("/foo/.", zk);
+ verifyCreateFails("/foo/..", zk);
+ verifyCreateFails("/./.", zk);
+ verifyCreateFails("/../..", zk);
+ verifyCreateFails("/\u0001foo", zk);
+ verifyCreateFails("/foo/bar/", zk);
+ verifyCreateFails("/foo//bar", zk);
+ verifyCreateFails("/foo/bar//", zk);
+
+ verifyCreateFails("foo", zk);
+ verifyCreateFails("a", zk);
+
+ zk.create("/createseqpar", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ // next two steps - related to sequential processing
+ // 1) verify that empty child name Assert.fails if not sequential
+ try {
+ zk.create("/createseqpar/", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ Assert.assertTrue(false);
+ } catch(IllegalArgumentException be) {
+ // catch this.
+ }
+
+ // 2) verify that empty child name success if sequential
+ zk.create("/createseqpar/", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ zk.create("/createseqpar/.", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ zk.create("/createseqpar/..", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ try {
+ zk.create("/createseqpar//", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ Assert.assertTrue(false);
+ } catch(IllegalArgumentException be) {
+ // catch this.
+ }
+ try {
+ zk.create("/createseqpar/./", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ Assert.assertTrue(false);
+ } catch(IllegalArgumentException be) {
+ // catch this.
+ }
+ try {
+ zk.create("/createseqpar/../", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ Assert.assertTrue(false);
+ } catch(IllegalArgumentException be) {
+ // catch this.
+ }
+
+
+ //check for the code path that throws at server
+ PrepRequestProcessor.setFailCreate(true);
+ try {
+ zk.create("/m", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ Assert.assertTrue(false);
+ } catch(KeeperException.BadArgumentsException be) {
+ // catch this.
+ }
+ PrepRequestProcessor.setFailCreate(false);
+ zk.create("/.foo", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/.f.", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/..f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/..f..", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f.c", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f\u0040f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/.f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/f.", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/..f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/f..", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/.f/f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/f/f./f", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ }
+
+// private void notestConnections()
+// throws IOException, InterruptedException, KeeperException
+// {
+// ZooKeeper zk;
+// for(int i = 0; i < 2000; i++) {
+// if (i % 100 == 0) {
+// LOG.info("Testing " + i + " connections");
+// }
+// // We want to make sure socket descriptors are going away
+// zk = new ZooKeeper(hostPort, 30000, this);
+// zk.getData("/", false, new Stat());
+// zk.close();
+// }
+// }
+
+ @Test
+ public void testDeleteWithChildren() throws Exception {
+ ZooKeeper zk = createClient();
+ zk.create("/parent", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ zk.create("/parent/child", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ try {
+ zk.delete("/parent", -1);
+ Assert.fail("Should have received a not equals message");
+ } catch (KeeperException e) {
+ Assert.assertEquals(KeeperException.Code.NOTEMPTY, e.code());
+ }
+ zk.delete("/parent/child", -1);
+ zk.delete("/parent", -1);
+ zk.close();
+ }
+
+ private class VerifyClientCleanup extends Thread {
+ int count;
+ int current = 0;
+
+ VerifyClientCleanup(String name, int count) {
+ super(name);
+ this.count = count;
+ }
+
+ public void run() {
+ try {
+ for (; current < count; current++) {
+ TestableZooKeeper zk = createClient();
+ // we've asked to close, wait for it to finish closing
+ // all the sub-threads otw the selector may not be
+ // closed when we check (false positive on test Assert.failure
+ zk.close(CONNECTION_TIMEOUT);
+ }
+ } catch (Throwable t) {
+ LOG.error("test Assert.failed", t);
+ }
+ }
+ }
+
+ /**
+ * Verify that the client is cleaning up properly. Open/close a large
+ * number of sessions. Essentially looking to see if sockets/selectors
+ * are being cleaned up properly during close.
+ *
+ * @throws Throwable
+ */
+ @Test
+ public void testClientCleanup() throws Throwable {
+ OSMXBean osMbean = new OSMXBean();
+ if (osMbean.getUnix() == false) {
+ LOG.warn("skipping testClientCleanup, only available on Unix");
+ return;
+ }
+
+ final int threadCount = 3;
+ final int clientCount = 10;
+
+ /* Log the number of fds used before and after a test is run. Verifies
+ * we are freeing resources correctly. Unfortunately this only works
+ * on unix systems (the only place sun has implemented as part of the
+ * mgmt bean api).
+ */
+ long initialFdCount = osMbean.getOpenFileDescriptorCount();
+
+ VerifyClientCleanup threads[] = new VerifyClientCleanup[threadCount];
+
+ for (int i = 0; i < threads.length; i++) {
+ threads[i] = new VerifyClientCleanup("VCC" + i, clientCount);
+ threads[i].start();
+ }
+
+ for (int i = 0; i < threads.length; i++) {
+ threads[i].join(CONNECTION_TIMEOUT);
+ Assert.assertTrue(threads[i].current == threads[i].count);
+ }
+
+ // if this Assert.fails it means we are not cleaning up after the closed
+ // sessions.
+ long currentCount = osMbean.getOpenFileDescriptorCount();
+ final String logmsg = "open fds after test ({}) are not significantly higher than before ({})";
+
+ if (currentCount > initialFdCount + 10) {
+ // consider as error
+ LOG.error(logmsg,Long.valueOf(currentCount),Long.valueOf(initialFdCount));
+ } else {
+ LOG.info(logmsg,Long.valueOf(currentCount),Long.valueOf(initialFdCount));
+ }
+ }
+
+
+ /**
+ * We create a perfectly valid 'exists' request, except that the opcode is wrong.
+ * @return
+ * @throws Exception
+ */
+ @Test
+ public void testNonExistingOpCode() throws Exception {
+ final CountDownLatch clientDisconnected = new CountDownLatch(1);
+ Watcher watcher = new Watcher() {
+ @Override
+ public synchronized void process(WatchedEvent event) {
+ if (event.getState() == KeeperState.Disconnected) {
+ clientDisconnected.countDown();
+ }
+ }
+ };
+ TestableZooKeeper zk = new TestableZooKeeper(hostPort, CONNECTION_TIMEOUT, watcher);
+
+ final String path = "/m1";
+
+ RequestHeader h = new RequestHeader();
+ h.setType(888); // This code does not exists
+ ExistsRequest request = new ExistsRequest();
+ request.setPath(path);
+ request.setWatch(false);
+ ExistsResponse response = new ExistsResponse();
+
+ ReplyHeader r = zk.submitRequest(h, request, response, null);
+
+ Assert.assertEquals(r.getErr(), Code.UNIMPLEMENTED.intValue());
+
+ // Sending a nonexisting opcode should cause the server to disconnect
+ Assert.assertTrue("failed to disconnect",
+ clientDisconnected.await(5000, TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ public void testTryWithResources() throws Exception {
+ ZooKeeper zooKeeper;
+ try (ZooKeeper zk = createClient()) {
+ zooKeeper = zk;
+ Assert.assertTrue(zooKeeper.getState().isAlive());
+ }
+
+ Assert.assertFalse(zooKeeper.getState().isAlive());
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/CnxManagerTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/CnxManagerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CnxManagerTest.java
new file mode 100644
index 0000000..a072bc0
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CnxManagerTest.java
@@ -0,0 +1,523 @@
+/**
+ * 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.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.net.Socket;
+
+import org.apache.zookeeper.common.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.quorum.QuorumCnxManager;
+import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
+import org.apache.zookeeper.server.quorum.QuorumCnxManager.InitialMessage;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CnxManagerTest extends ZKTestCase {
+ protected static final Logger LOG = LoggerFactory.getLogger(FLENewEpochTest.class);
+ protected static final int THRESHOLD = 4;
+
+ int count;
+ HashMap<Long,QuorumServer> peers;
+ File peerTmpdir[];
+ int peerQuorumPort[];
+ int peerClientPort[];
+ @Before
+ public void setUp() throws Exception {
+
+ this.count = 3;
+ this.peers = new HashMap<Long,QuorumServer>(count);
+ peerTmpdir = new File[count];
+ peerQuorumPort = new int[count];
+ peerClientPort = new int[count];
+
+ for(int i = 0; i < count; i++) {
+ peerQuorumPort[i] = PortAssignment.unique();
+ peerClientPort[i] = PortAssignment.unique();
+ peers.put(Long.valueOf(i),
+ new QuorumServer(i,
+ new InetSocketAddress(
+ "127.0.0.1", peerQuorumPort[i]),
+ new InetSocketAddress(
+ "127.0.0.1", PortAssignment.unique()),
+ new InetSocketAddress(
+ "127.0.0.1", peerClientPort[i])));
+ peerTmpdir[i] = ClientBase.createTmpDir();
+ }
+ }
+
+ ByteBuffer createMsg(int state, long leader, long zxid, long epoch){
+ byte requestBytes[] = new byte[28];
+ ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
+
+ /*
+ * Building notification packet to send
+ */
+
+ requestBuffer.clear();
+ requestBuffer.putInt(state);
+ requestBuffer.putLong(leader);
+ requestBuffer.putLong(zxid);
+ requestBuffer.putLong(epoch);
+
+ return requestBuffer;
+ }
+
+ class CnxManagerThread extends Thread {
+
+ boolean failed;
+ CnxManagerThread(){
+ failed = false;
+ }
+
+ public void run(){
+ try {
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[0], peerTmpdir[0], peerClientPort[0], 3, 0, 1000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if(listener != null){
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+
+ long sid = 1;
+ cnxManager.toSend(sid, createMsg(ServerState.LOOKING.ordinal(), 0, -1, 1));
+
+ Message m = null;
+ int numRetries = 1;
+ while((m == null) && (numRetries++ <= THRESHOLD)){
+ m = cnxManager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
+ if(m == null) cnxManager.connectAll();
+ }
+
+ if(numRetries > THRESHOLD){
+ failed = true;
+ return;
+ }
+
+ cnxManager.testInitiateConnection(sid);
+
+ m = cnxManager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
+ if(m == null){
+ failed = true;
+ return;
+ }
+ } catch (Exception e) {
+ LOG.error("Exception while running mock thread", e);
+ Assert.fail("Unexpected exception");
+ }
+ }
+ }
+
+ @Test
+ public void testCnxManager() throws Exception {
+ CnxManagerThread thread = new CnxManagerThread();
+
+ thread.start();
+
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if(listener != null){
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+
+ cnxManager.toSend(0L, createMsg(ServerState.LOOKING.ordinal(), 1, -1, 1));
+
+ Message m = null;
+ int numRetries = 1;
+ while((m == null) && (numRetries++ <= THRESHOLD)){
+ m = cnxManager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
+ if(m == null) cnxManager.connectAll();
+ }
+
+ Assert.assertTrue("Exceeded number of retries", numRetries <= THRESHOLD);
+
+ thread.join(5000);
+ if (thread.isAlive()) {
+ Assert.fail("Thread didn't join");
+ } else {
+ if(thread.failed)
+ Assert.fail("Did not receive expected message");
+ }
+ cnxManager.halt();
+ Assert.assertFalse(cnxManager.listener.isAlive());
+ }
+
+ @Test
+ public void testCnxManagerTimeout() throws Exception {
+ Random rand = new Random();
+ byte b = (byte) rand.nextInt();
+ int deadPort = PortAssignment.unique();
+ String deadAddress = "10.1.1." + b;
+
+ LOG.info("This is the dead address I'm trying: " + deadAddress);
+
+ peers.put(Long.valueOf(2),
+ new QuorumServer(2,
+ new InetSocketAddress(deadAddress, deadPort),
+ new InetSocketAddress(deadAddress, PortAssignment.unique()),
+ new InetSocketAddress(deadAddress, PortAssignment.unique())));
+ peerTmpdir[2] = ClientBase.createTmpDir();
+
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if(listener != null){
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+
+ long begin = Time.currentElapsedTime();
+ cnxManager.toSend(2L, createMsg(ServerState.LOOKING.ordinal(), 1, -1, 1));
+ long end = Time.currentElapsedTime();
+
+ if((end - begin) > 6000) Assert.fail("Waited more than necessary");
+ cnxManager.halt();
+ Assert.assertFalse(cnxManager.listener.isAlive());
+ }
+
+ /**
+ * Tests a bug in QuorumCnxManager that causes a spin lock
+ * when a negative value is sent. This test checks if the
+ * connection is being closed upon a message with negative
+ * length.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testCnxManagerSpinLock() throws Exception {
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if(listener != null){
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+
+ int port = peers.get(peer.getId()).electionAddr.getPort();
+ LOG.info("Election port: " + port);
+
+ Thread.sleep(1000);
+
+ SocketChannel sc = SocketChannel.open();
+ sc.socket().connect(peers.get(1L).electionAddr, 5000);
+
+ InetSocketAddress otherAddr = peers.get(new Long(2)).electionAddr;
+ DataOutputStream dout = new DataOutputStream(sc.socket().getOutputStream());
+ dout.writeLong(QuorumCnxManager.PROTOCOL_VERSION);
+ dout.writeLong(new Long(2));
+ String addr = otherAddr.getHostString()+ ":" + otherAddr.getPort();
+ byte[] addr_bytes = addr.getBytes();
+ dout.writeInt(addr_bytes.length);
+ dout.write(addr_bytes);
+ dout.flush();
+
+
+ ByteBuffer msgBuffer = ByteBuffer.wrap(new byte[4]);
+ msgBuffer.putInt(-20);
+ msgBuffer.position(0);
+ sc.write(msgBuffer);
+
+ Thread.sleep(1000);
+
+ try{
+ /*
+ * Write a number of times until it
+ * detects that the socket is broken.
+ */
+ for(int i = 0; i < 100; i++){
+ msgBuffer.position(0);
+ sc.write(msgBuffer);
+ }
+ Assert.fail("Socket has not been closed");
+ } catch (Exception e) {
+ LOG.info("Socket has been closed as expected");
+ }
+ peer.shutdown();
+ cnxManager.halt();
+ Assert.assertFalse(cnxManager.listener.isAlive());
+ }
+
+ /**
+ * Tests a bug in QuorumCnxManager that causes a NPE when a 3.4.6
+ * observer connects to a 3.5.0 server.
+ * {@link https://issues.apache.org/jira/browse/ZOOKEEPER-1789}
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testCnxManagerNPE() throws Exception {
+ // the connecting peer (id = 2) is a 3.4.6 observer
+ peers.get(2L).type = LearnerType.OBSERVER;
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1],
+ peerClientPort[1], 3, 1, 1000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if (listener != null) {
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+ int port = peers.get(peer.getId()).electionAddr.getPort();
+ LOG.info("Election port: " + port);
+
+ Thread.sleep(1000);
+
+ SocketChannel sc = SocketChannel.open();
+ sc.socket().connect(peers.get(1L).electionAddr, 5000);
+
+ /*
+ * Write id (3.4.6 protocol). This previously caused a NPE in
+ * QuorumCnxManager.
+ */
+ byte[] msgBytes = new byte[8];
+ ByteBuffer msgBuffer = ByteBuffer.wrap(msgBytes);
+ msgBuffer.putLong(2L);
+ msgBuffer.position(0);
+ sc.write(msgBuffer);
+
+ msgBuffer = ByteBuffer.wrap(new byte[8]);
+ // write length of message
+ msgBuffer.putInt(4);
+ // write message
+ msgBuffer.putInt(5);
+ msgBuffer.position(0);
+ sc.write(msgBuffer);
+
+ Message m = cnxManager.pollRecvQueue(1000, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(m);
+
+ peer.shutdown();
+ cnxManager.halt();
+ Assert.assertFalse(cnxManager.listener.isAlive());
+ }
+
+ /*
+ * Test if a receiveConnection is able to timeout on socket errors
+ */
+ @Test
+ public void testSocketTimeout() throws Exception {
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 2000, 2, 2);
+ QuorumCnxManager cnxManager = peer.createCnxnManager();
+ QuorumCnxManager.Listener listener = cnxManager.listener;
+ if(listener != null){
+ listener.start();
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+ int port = peers.get(peer.getId()).electionAddr.getPort();
+ LOG.info("Election port: " + port);
+ Thread.sleep(1000);
+
+ Socket sock = new Socket();
+ sock.connect(peers.get(1L).electionAddr, 5000);
+ long begin = Time.currentElapsedTime();
+ // Read without sending data. Verify timeout.
+ cnxManager.receiveConnection(sock);
+ long end = Time.currentElapsedTime();
+ if((end - begin) > ((peer.getSyncLimit() * peer.getTickTime()) + 500)) Assert.fail("Waited more than necessary");
+ cnxManager.halt();
+ Assert.assertFalse(cnxManager.listener.isAlive());
+ }
+
+ /*
+ * Test if Worker threads are getting killed after connection loss
+ */
+ @Test
+ public void testWorkerThreads() throws Exception {
+ ArrayList<QuorumPeer> peerList = new ArrayList<QuorumPeer>();
+ try {
+ for (int sid = 0; sid < 3; sid++) {
+ QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[sid],
+ peerTmpdir[sid], peerClientPort[sid], 3, sid, 1000, 2,
+ 2);
+ LOG.info("Starting peer {}", peer.getId());
+ peer.start();
+ peerList.add(sid, peer);
+ }
+ String failure = verifyThreadCount(peerList, 4);
+ Assert.assertNull(failure, failure);
+ for (int myid = 0; myid < 3; myid++) {
+ for (int i = 0; i < 5; i++) {
+ // halt one of the listeners and verify count
+ QuorumPeer peer = peerList.get(myid);
+ LOG.info("Round {}, halting peer ",
+ new Object[] { i, peer.getId() });
+ peer.shutdown();
+ peerList.remove(myid);
+ failure = verifyThreadCount(peerList, 2);
+ Assert.assertNull(failure, failure);
+ // Restart halted node and verify count
+ peer = new QuorumPeer(peers, peerTmpdir[myid],
+ peerTmpdir[myid], peerClientPort[myid], 3, myid,
+ 1000, 2, 2);
+ LOG.info("Round {}, restarting peer ",
+ new Object[] { i, peer.getId() });
+ peer.start();
+ peerList.add(myid, peer);
+ failure = verifyThreadCount(peerList, 4);
+ Assert.assertNull(failure, failure);
+ }
+ }
+ } finally {
+ for (QuorumPeer quorumPeer : peerList) {
+ quorumPeer.shutdown();
+ }
+ }
+ }
+
+ /**
+ * Returns null on success, otw the message assoc with the failure
+ * @throws InterruptedException
+ */
+ public String verifyThreadCount(ArrayList<QuorumPeer> peerList, long ecnt)
+ throws InterruptedException
+ {
+ String failure = null;
+ for (int i = 0; i < 480; i++) {
+ Thread.sleep(500);
+
+ failure = _verifyThreadCount(peerList, ecnt);
+ if (failure == null) {
+ return null;
+ }
+ }
+ return failure;
+ }
+ public String _verifyThreadCount(ArrayList<QuorumPeer> peerList, long ecnt) {
+ for (int myid = 0; myid < peerList.size(); myid++) {
+ QuorumPeer peer = peerList.get(myid);
+ QuorumCnxManager cnxManager = peer.getQuorumCnxManager();
+ long cnt = cnxManager.getThreadCount();
+ if (cnt != ecnt) {
+ return new Date()
+ + " Incorrect number of Worker threads for sid=" + myid
+ + " expected " + ecnt + " found " + cnt;
+ }
+ }
+ return null;
+ }
+
+ @Test
+ public void testInitialMessage() throws Exception {
+ InitialMessage msg;
+ ByteArrayOutputStream bos;
+ DataInputStream din;
+ DataOutputStream dout;
+ String hostport;
+
+ // message with bad protocol version
+ try {
+
+ // the initial message (without the protocol version)
+ hostport = "10.0.0.2:3888";
+ bos = new ByteArrayOutputStream();
+ dout = new DataOutputStream(bos);
+ dout.writeLong(5L); // sid
+ dout.writeInt(hostport.getBytes().length);
+ dout.writeBytes(hostport);
+
+ // now parse it
+ din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
+ msg = InitialMessage.parse(-65530L, din);
+ Assert.fail("bad protocol version accepted");
+ } catch (InitialMessage.InitialMessageException ex) {}
+
+ // message too long
+ try {
+
+ hostport = createLongString(1048576);
+ bos = new ByteArrayOutputStream();
+ dout = new DataOutputStream(bos);
+ dout.writeLong(5L); // sid
+ dout.writeInt(hostport.getBytes().length);
+ dout.writeBytes(hostport);
+
+ din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
+ msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+ Assert.fail("long message accepted");
+ } catch (InitialMessage.InitialMessageException ex) {}
+
+ // bad hostport string
+ try {
+
+ hostport = "what's going on here?";
+ bos = new ByteArrayOutputStream();
+ dout = new DataOutputStream(bos);
+ dout.writeLong(5L); // sid
+ dout.writeInt(hostport.getBytes().length);
+ dout.writeBytes(hostport);
+
+ din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
+ msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+ Assert.fail("bad hostport accepted");
+ } catch (InitialMessage.InitialMessageException ex) {}
+
+ // good message
+ try {
+
+ hostport = "10.0.0.2:3888";
+ bos = new ByteArrayOutputStream();
+ dout = new DataOutputStream(bos);
+ dout.writeLong(5L); // sid
+ dout.writeInt(hostport.getBytes().length);
+ dout.writeBytes(hostport);
+
+ // now parse it
+ din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
+ msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+ } catch (InitialMessage.InitialMessageException ex) {
+ Assert.fail(ex.toString());
+ }
+ }
+
+ private String createLongString(int size) {
+ StringBuilder sb = new StringBuilder(size);
+ for (int i=0; i < size; i++) {
+ sb.append('x');
+ }
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/ConnectStringParserTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ConnectStringParserTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ConnectStringParserTest.java
new file mode 100644
index 0000000..393cc03
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ConnectStringParserTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.apache.zookeeper.client.ConnectStringParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ConnectStringParserTest extends ZKTestCase{
+
+ @Test
+ public void testSingleServerChrootPath(){
+ String chrootPath = "/hallo/welt";
+ String servers = "10.10.10.1";
+ assertChrootPath(chrootPath,
+ new ConnectStringParser(servers+chrootPath));
+ }
+
+ @Test
+ public void testMultipleServersChrootPath(){
+ String chrootPath = "/hallo/welt";
+ String servers = "10.10.10.1,10.10.10.2";
+ assertChrootPath(chrootPath,
+ new ConnectStringParser(servers+chrootPath));
+ }
+
+ @Test
+ public void testParseServersWithoutPort(){
+ String servers = "10.10.10.1,10.10.10.2";
+ ConnectStringParser parser = new ConnectStringParser(servers);
+
+ Assert.assertEquals("10.10.10.1", parser.getServerAddresses().get(0).getHostString());
+ Assert.assertEquals("10.10.10.2", parser.getServerAddresses().get(1).getHostString());
+ }
+
+ @Test
+ public void testParseServersWithPort(){
+ String servers = "10.10.10.1:112,10.10.10.2:110";
+ ConnectStringParser parser = new ConnectStringParser(servers);
+
+ Assert.assertEquals("10.10.10.1", parser.getServerAddresses().get(0).getHostString());
+ Assert.assertEquals("10.10.10.2", parser.getServerAddresses().get(1).getHostString());
+
+ Assert.assertEquals(112, parser.getServerAddresses().get(0).getPort());
+ Assert.assertEquals(110, parser.getServerAddresses().get(1).getPort());
+ }
+
+ private void assertChrootPath(String expected, ConnectStringParser parser){
+ Assert.assertEquals(expected, parser.getChrootPath());
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateModeTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateModeTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateModeTest.java
new file mode 100644
index 0000000..fc61adf
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateModeTest.java
@@ -0,0 +1,91 @@
+/**
+ * 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.util.EnumSet;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.KeeperException.Code;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CreateModeTest extends ZKTestCase {
+
+ @Test
+ public void testBasicCreateMode() {
+ CreateMode cm = CreateMode.PERSISTENT;
+ Assert.assertEquals(cm.toFlag(), 0);
+ Assert.assertFalse(cm.isEphemeral());
+ Assert.assertFalse(cm.isSequential());
+ Assert.assertFalse(cm.isContainer());
+
+ cm = CreateMode.EPHEMERAL;
+ Assert.assertEquals(cm.toFlag(), 1);
+ Assert.assertTrue(cm.isEphemeral());
+ Assert.assertFalse(cm.isSequential());
+ Assert.assertFalse(cm.isContainer());
+
+ cm = CreateMode.PERSISTENT_SEQUENTIAL;
+ Assert.assertEquals(cm.toFlag(), 2);
+ Assert.assertFalse(cm.isEphemeral());
+ Assert.assertTrue(cm.isSequential());
+ Assert.assertFalse(cm.isContainer());
+
+ cm = CreateMode.EPHEMERAL_SEQUENTIAL;
+ Assert.assertEquals(cm.toFlag(), 3);
+ Assert.assertTrue(cm.isEphemeral());
+ Assert.assertTrue(cm.isSequential());
+ Assert.assertFalse(cm.isContainer());
+
+ cm = CreateMode.CONTAINER;
+ Assert.assertEquals(cm.toFlag(), 4);
+ Assert.assertFalse(cm.isEphemeral());
+ Assert.assertFalse(cm.isSequential());
+ Assert.assertTrue(cm.isContainer());
+ }
+
+ @Test
+ public void testFlagConversion() throws KeeperException {
+ // Ensure we get the same value back after round trip conversion
+ EnumSet<CreateMode> allModes = EnumSet.allOf(CreateMode.class);
+
+ for(CreateMode cm : allModes) {
+ Assert.assertEquals(cm, CreateMode.fromFlag( cm.toFlag() ) );
+ }
+ }
+
+ @Test
+ public void testInvalidFlagConversion() throws KeeperException {
+ try {
+ CreateMode.fromFlag(99);
+ Assert.fail("Shouldn't be able to convert 99 to a CreateMode.");
+ } catch(KeeperException ke) {
+ Assert.assertEquals(Code.BADARGUMENTS, ke.code());
+ }
+
+ try {
+ CreateMode.fromFlag(-1);
+ Assert.fail("Shouldn't be able to convert -1 to a CreateMode.");
+ } catch(KeeperException ke) {
+ Assert.assertEquals(Code.BADARGUMENTS, ke.code());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateTest.java
new file mode 100644
index 0000000..1b427a1
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/CreateTest.java
@@ -0,0 +1,115 @@
+/**
+ * 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 org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.Stat;
+import org.junit.Assert;
+import org.junit.Test;
+/**
+ * Test suite for validating the Create API.
+ */
+public class CreateTest extends ClientBase {
+ private ZooKeeper zk;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ zk = createClient();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ zk.close();
+ }
+
+ @Test
+ public void testCreate()
+ throws IOException, KeeperException, InterruptedException {
+ createNoStatVerifyResult("/foo");
+ createNoStatVerifyResult("/foo/child");
+ }
+
+ @Test
+ public void testCreateWithStat()
+ throws IOException, KeeperException, InterruptedException {
+ String name = "/foo";
+ Stat stat = createWithStatVerifyResult("/foo");
+ Stat childStat = createWithStatVerifyResult("/foo/child");
+ // Don't expect to get the same stats for different creates.
+ Assert.assertFalse(stat.equals(childStat));
+ }
+
+ @Test
+ public void testCreateWithNullStat()
+ throws IOException, KeeperException, InterruptedException {
+ String name = "/foo";
+ Assert.assertNull(zk.exists(name, false));
+
+ Stat stat = null;
+ // If a null Stat object is passed the create should still
+ // succeed, but no Stat info will be returned.
+ String path = zk.create(name, name.getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, stat);
+ Assert.assertNull(stat);
+ Assert.assertNotNull(zk.exists(name, false));
+ }
+
+ private void createNoStatVerifyResult(String newName)
+ throws KeeperException, InterruptedException {
+ Assert.assertNull("Node existed before created", zk.exists(newName, false));
+ String path = zk.create(newName, newName.getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ Assert.assertEquals(path, newName);
+ Assert.assertNotNull("Node was not created as expected",
+ zk.exists(newName, false));
+ }
+
+ private Stat createWithStatVerifyResult(String newName)
+ throws KeeperException, InterruptedException {
+ Assert.assertNull("Node existed before created", zk.exists(newName, false));
+ Stat stat = new Stat();
+ String path = zk.create(newName, newName.getBytes(),
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, stat);
+ Assert.assertEquals(path, newName);
+ validateCreateStat(stat, newName);
+
+ Stat referenceStat = zk.exists(newName, false);
+ Assert.assertNotNull("Node was not created as expected", referenceStat);
+ Assert.assertEquals(referenceStat, stat);
+
+ return stat;
+ }
+
+ private void validateCreateStat(Stat stat, String name) {
+ Assert.assertEquals(stat.getCzxid(), stat.getMzxid());
+ Assert.assertEquals(stat.getCzxid(), stat.getPzxid());
+ Assert.assertEquals(stat.getCtime(), stat.getMtime());
+ Assert.assertEquals(0, stat.getCversion());
+ Assert.assertEquals(0, stat.getVersion());
+ Assert.assertEquals(0, stat.getAversion());
+ Assert.assertEquals(0, stat.getEphemeralOwner());
+ Assert.assertEquals(name.length(), stat.getDataLength());
+ Assert.assertEquals(0, stat.getNumChildren());
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectableZooKeeper.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectableZooKeeper.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectableZooKeeper.java
new file mode 100644
index 0000000..619bdc6
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectableZooKeeper.java
@@ -0,0 +1,58 @@
+/**
+ * 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 org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+
+public class DisconnectableZooKeeper extends ZooKeeper {
+ public DisconnectableZooKeeper(String host, int sessionTimeout, Watcher watcher)
+ throws IOException
+ {
+ super(host, sessionTimeout, watcher);
+ }
+
+ public DisconnectableZooKeeper(String host, int sessionTimeout, Watcher watcher,
+ long sessionId, byte[] sessionPasswd)
+ throws IOException
+ {
+ super(host, sessionTimeout, watcher, sessionId, sessionPasswd);
+ }
+
+ /** Testing only!!! Really!!!! This is only here to test when the client
+ * disconnects from the server w/o sending a session disconnect (ie
+ * ending the session cleanly). The server will eventually notice the
+ * client is no longer pinging and will timeout the session.
+ */
+ public void disconnect() throws IOException {
+ cnxn.disconnect();
+ }
+
+ /**
+ * Prevent the client from automatically reconnecting if the connection to the
+ * server is lost
+ */
+ public void dontReconnect() throws Exception {
+ java.lang.reflect.Field f = cnxn.getClass().getDeclaredField("closing");
+ f.setAccessible(true);
+ f.setBoolean(cnxn, true);
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectedWatcherTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectedWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectedWatcherTest.java
new file mode 100644
index 0000000..aa65e21
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DisconnectedWatcherTest.java
@@ -0,0 +1,255 @@
+/**
+ * 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.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DisconnectedWatcherTest extends ClientBase {
+ protected static final Logger LOG = LoggerFactory.getLogger(DisconnectedWatcherTest.class);
+ final int TIMEOUT = 5000;
+
+ private class MyWatcher extends CountdownWatcher {
+ LinkedBlockingQueue<WatchedEvent> events =
+ new LinkedBlockingQueue<WatchedEvent>();
+
+ public void process(WatchedEvent event) {
+ super.process(event);
+ if (event.getType() != Event.EventType.None) {
+ try {
+ events.put(event);
+ } catch (InterruptedException e) {
+ LOG.warn("ignoring interrupt during event.put");
+ }
+ }
+ }
+ }
+
+ // @see jira issue ZOOKEEPER-961
+
+ @Test
+ public void testChildWatcherAutoResetWithChroot() throws Exception {
+ ZooKeeper zk1 = createClient();
+
+ zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ MyWatcher watcher = new MyWatcher();
+ ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1");
+ zk2.getChildren("/", true );
+
+ // this call shouldn't trigger any error or watch
+ zk1.create("/youdontmatter1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ // this should trigger the watch
+ zk1.create("/ch1/youshouldmatter1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/", e.getPath());
+
+ MyWatcher childWatcher = new MyWatcher();
+ zk2.getChildren("/", childWatcher);
+
+ stopServer();
+ watcher.waitForDisconnected(3000);
+ startServer();
+ watcher.waitForConnected(3000);
+
+ // this should trigger the watch
+ zk1.create("/ch1/youshouldmatter2", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/", e.getPath());
+ }
+
+ @Test
+ public void testDefaultWatcherAutoResetWithChroot() throws Exception {
+ ZooKeeper zk1 = createClient();
+
+ zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ MyWatcher watcher = new MyWatcher();
+ ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1");
+ zk2.getChildren("/", true );
+
+ // this call shouldn't trigger any error or watch
+ zk1.create("/youdontmatter1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ // this should trigger the watch
+ zk1.create("/ch1/youshouldmatter1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/", e.getPath());
+
+ zk2.getChildren("/", true );
+
+ stopServer();
+ watcher.waitForDisconnected(3000);
+ startServer();
+ watcher.waitForConnected(3000);
+
+ // this should trigger the watch
+ zk1.create("/ch1/youshouldmatter2", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/", e.getPath());
+ }
+
+ @Test
+ public void testDeepChildWatcherAutoResetWithChroot() throws Exception {
+ ZooKeeper zk1 = createClient();
+
+ zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ zk1.create("/ch1/here", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ zk1.create("/ch1/here/we", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ zk1.create("/ch1/here/we/are", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ MyWatcher watcher = new MyWatcher();
+ ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1/here/we");
+ zk2.getChildren("/are", true );
+
+ // this should trigger the watch
+ zk1.create("/ch1/here/we/are/now", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/are", e.getPath());
+
+ MyWatcher childWatcher = new MyWatcher();
+ zk2.getChildren("/are", childWatcher);
+
+ stopServer();
+ watcher.waitForDisconnected(3000);
+ startServer();
+ watcher.waitForConnected(3000);
+
+ // this should trigger the watch
+ zk1.create("/ch1/here/we/are/again", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals("/are", e.getPath());
+ }
+
+ // @see jira issue ZOOKEEPER-706. Test auto reset of a large number of
+ // watches which require multiple SetWatches calls.
+ @Test(timeout = 840000)
+ public void testManyChildWatchersAutoReset() throws Exception {
+ ZooKeeper zk1 = createClient();
+
+ MyWatcher watcher = new MyWatcher();
+ ZooKeeper zk2 = createClient(watcher);
+
+ // 110 character base path
+ String pathBase = "/long-path-000000000-111111111-222222222-333333333-444444444-"
+ + "555555555-666666666-777777777-888888888-999999999";
+
+ zk1.create(pathBase, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ // Create 10,000 nodes. This should ensure the length of our
+ // watches set below exceeds 1MB.
+ List<String> paths = new ArrayList<String>();
+ for (int i = 0; i < 10000; i++) {
+ String path = zk1.create(pathBase + "/ch-", null, Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT_SEQUENTIAL);
+ paths.add(path);
+ }
+ LOG.info("Created 10,000 nodes.");
+
+ MyWatcher childWatcher = new MyWatcher();
+
+ // Set a combination of child/exists/data watches
+ int i = 0;
+ for (String path : paths) {
+ if (i % 3 == 0) {
+ zk2.getChildren(path, childWatcher);
+ } else if (i % 3 == 1) {
+ zk2.exists(path + "/foo", childWatcher);
+ } else if (i % 3 == 2) {
+ zk2.getData(path, childWatcher, null);
+ }
+
+ i++;
+ }
+
+ stopServer();
+ watcher.waitForDisconnected(30000);
+ startServer();
+ watcher.waitForConnected(30000);
+
+ // Trigger the watches and ensure they properly propagate to the client
+ i = 0;
+ for (String path : paths) {
+ if (i % 3 == 0) {
+ zk1.create(path + "/ch", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ WatchedEvent e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+ Assert.assertEquals(path, e.getPath());
+ } else if (i % 3 == 1) {
+ zk1.create(path + "/foo", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ WatchedEvent e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeCreated, e.getType());
+ Assert.assertEquals(path + "/foo", e.getPath());
+ } else if (i % 3 == 2) {
+ zk1.setData(path, new byte[]{1, 2, 3}, -1);
+
+ WatchedEvent e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull(e);
+ Assert.assertEquals(EventType.NodeDataChanged, e.getType());
+ Assert.assertEquals(path, e.getPath());
+ }
+
+ i++;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/43d71c2e/zookeeper-server/src/test/java/org/apache/zookeeper/test/DuplicateLocalSessionUpgradeTest.java
----------------------------------------------------------------------
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/DuplicateLocalSessionUpgradeTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DuplicateLocalSessionUpgradeTest.java
new file mode 100644
index 0000000..a21acdc
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/DuplicateLocalSessionUpgradeTest.java
@@ -0,0 +1,113 @@
+/**
+ * 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.CreateMode;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * When request are route incorrectly, both follower and the leader will perform
+ * local session upgrade. So we saw CreateSession twice in txnlog This doesn't
+ * affect the correctness but cause the ensemble to see more load than
+ * necessary.
+ */
+public class DuplicateLocalSessionUpgradeTest extends ZKTestCase {
+ protected static final Logger LOG = LoggerFactory
+ .getLogger(DuplicateLocalSessionUpgradeTest.class);
+
+ private final QuorumBase qb = new QuorumBase();
+
+ private static final int CONNECTION_TIMEOUT = ClientBase.CONNECTION_TIMEOUT;
+
+ @Before
+ public void setUp() throws Exception {
+ LOG.info("STARTING quorum " + getClass().getName());
+ qb.localSessionsEnabled = true;
+ qb.localSessionsUpgradingEnabled = true;
+ qb.setUp();
+ ClientBase.waitForServerUp(qb.hostPort, 10000);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ LOG.info("STOPPING quorum " + getClass().getName());
+ qb.tearDown();
+ }
+
+ @Test
+ public void testLocalSessionUpgradeOnFollower() throws Exception {
+ testLocalSessionUpgrade(false);
+ }
+
+ @Test
+ public void testLocalSessionUpgradeOnLeader() throws Exception {
+ testLocalSessionUpgrade(true);
+ }
+
+ private void testLocalSessionUpgrade(boolean testLeader) throws Exception {
+
+ int leaderIdx = qb.getLeaderIndex();
+ Assert.assertFalse("No leader in quorum?", leaderIdx == -1);
+ int followerIdx = (leaderIdx + 1) % 5;
+ int testPeerIdx = testLeader ? leaderIdx : followerIdx;
+ String hostPorts[] = qb.hostPort.split(",");
+
+ CountdownWatcher watcher = new CountdownWatcher();
+ ZooKeeper zk = qb.createClient(watcher, hostPorts[testPeerIdx],
+ CONNECTION_TIMEOUT);
+ watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+ final String firstPath = "/first";
+ final String secondPath = "/ephemeral";
+
+ // Just create some node so that we know the current zxid
+ zk.create(firstPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+
+ // Now, try an ephemeral node. This will trigger session upgrade
+ // so there will be createSession request inject into the pipeline
+ // prior to this request
+ zk.create(secondPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
+ CreateMode.EPHEMERAL);
+
+ Stat firstStat = zk.exists(firstPath, null);
+ Assert.assertNotNull(firstStat);
+
+ Stat secondStat = zk.exists(secondPath, null);
+ Assert.assertNotNull(secondStat);
+
+ long zxidDiff = secondStat.getCzxid() - firstStat.getCzxid();
+
+ // If there is only one createSession request in between, zxid diff
+ // will be exactly 2. The alternative way of checking is to actually
+ // read txnlog but this should be sufficient
+ Assert.assertEquals(2L, zxidDiff);
+
+ }
+}