zookeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h...@apache.org
Subject [1/2] zookeeper git commit: ZOOKEEPER-1177: Add the memory optimized watch manager for concentrate watches scenario
Date Fri, 28 Sep 2018 21:38:33 GMT
Repository: zookeeper
Updated Branches:
  refs/heads/master 4ebb847bc -> fdde8b006


http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/main/org/apache/zookeeper/server/watch/WatcherOrBitSet.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/watch/WatcherOrBitSet.java b/src/java/main/org/apache/zookeeper/server/watch/WatcherOrBitSet.java
new file mode 100644
index 0000000..83b186f
--- /dev/null
+++ b/src/java/main/org/apache/zookeeper/server/watch/WatcherOrBitSet.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.watch;
+
+import java.util.Set;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.server.util.BitHashSet;
+
+public class WatcherOrBitSet {
+
+    private Set<Watcher> watchers;
+    private BitHashSet watcherBits;
+
+    public WatcherOrBitSet(final Set<Watcher> watchers) {
+        this.watchers = watchers;
+    }
+
+    public WatcherOrBitSet(final BitHashSet watcherBits) {
+        this.watcherBits = watcherBits;
+    }
+
+    public boolean contains(Watcher watcher) {
+        if (watchers == null) {
+            return false;
+        }
+        return watchers.contains(watcher);
+    }
+
+    public boolean contains(int watcherBit) {
+        if (watcherBits == null) {
+            return false;
+        }
+        return watcherBits.contains(watcherBit);
+    }
+
+    public int size() {
+        if (watchers != null) {
+            return watchers.size();
+        }
+        if (watcherBits != null) {
+            return watcherBits.size();
+        }
+        return 0;
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/main/org/apache/zookeeper/server/watch/WatchesPathReport.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/watch/WatchesPathReport.java b/src/java/main/org/apache/zookeeper/server/watch/WatchesPathReport.java
new file mode 100644
index 0000000..38f02de
--- /dev/null
+++ b/src/java/main/org/apache/zookeeper/server/watch/WatchesPathReport.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.watch;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A watch report, essentially a mapping of path to session IDs of sessions that
+ * have set a watch on that path. This class is immutable.
+ */
+public class WatchesPathReport {
+
+    private final Map<String, Set<Long>> path2Ids;
+
+    /**
+     * Creates a new report.
+     *
+     * @param path2Ids map of paths to session IDs of sessions that have set a
+     * watch on that path
+     */
+    WatchesPathReport(Map<String, Set<Long>> path2Ids) {
+        this.path2Ids = Collections.unmodifiableMap(deepCopy(path2Ids));
+    }
+
+    private static Map<String, Set<Long>> deepCopy(Map<String, Set<Long>> m) {
+        Map<String, Set<Long>> m2 = new HashMap<String, Set<Long>>();
+        for (Map.Entry<String, Set<Long>> e : m.entrySet()) {
+            m2.put(e.getKey(), new HashSet<Long>(e.getValue()));
+        }
+        return m2;
+    }
+
+    /**
+     * Checks if the given path has watches set.
+     *
+     * @param path path
+     * @return true if path has watch set
+     */
+    public boolean hasSessions(String path) {
+        return path2Ids.containsKey(path);
+    }
+    /**
+     * Gets the session IDs of sessions that have set watches on the given path.
+     * The returned set is immutable.
+     *
+     * @param path session ID
+     * @return session IDs of sessions that have set watches on the path, or
+     * null if none
+     */
+    public Set<Long> getSessions(String path) {
+        Set<Long> s = path2Ids.get(path);
+        return s != null ? Collections.unmodifiableSet(s) : null;
+    }
+
+    /**
+     * Converts this report to a map. The returned map is mutable, and changes
+     * to it do not reflect back into this report.
+     *
+     * @return map representation of report
+     */
+    public Map<String, Set<Long>> toMap() {
+        return deepCopy(path2Ids);
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/main/org/apache/zookeeper/server/watch/WatchesReport.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/watch/WatchesReport.java b/src/java/main/org/apache/zookeeper/server/watch/WatchesReport.java
new file mode 100644
index 0000000..ac888d3
--- /dev/null
+++ b/src/java/main/org/apache/zookeeper/server/watch/WatchesReport.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.watch;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A watch report, essentially a mapping of session ID to paths that the session
+ * has set a watch on. This class is immutable.
+ */
+public class WatchesReport {
+
+    private final Map<Long, Set<String>> id2paths;
+
+    /**
+     * Creates a new report.
+     *
+     * @param id2paths map of session IDs to paths that each session has set
+     * a watch on
+     */
+    WatchesReport(Map<Long, Set<String>> id2paths) {
+        this.id2paths = Collections.unmodifiableMap(deepCopy(id2paths));
+    }
+
+    private static Map<Long, Set<String>> deepCopy(Map<Long, Set<String>> m) {
+        Map<Long, Set<String>> m2 = new HashMap<Long, Set<String>>();
+        for (Map.Entry<Long, Set<String>> e : m.entrySet()) {
+            m2.put(e.getKey(), new HashSet<String>(e.getValue()));
+        }
+        return m2;
+    }
+
+    /**
+     * Checks if the given session has watches set.
+     *
+     * @param sessionId session ID
+     * @return true if session has paths with watches set
+     */
+    public boolean hasPaths(long sessionId) {
+        return id2paths.containsKey(sessionId);
+    }
+
+    /**
+     * Gets the paths that the given session has set watches on. The returned
+     * set is immutable.
+     *
+     * @param sessionId session ID
+     * @return paths that have watches set by the session, or null if none
+     */
+    public Set<String> getPaths(long sessionId) {
+        Set<String> s = id2paths.get(sessionId);
+        return s != null ? Collections.unmodifiableSet(s) : null;
+    }
+
+    /**
+     * Converts this report to a map. The returned map is mutable, and changes
+     * to it do not reflect back into this report.
+     *
+     * @return map representation of report
+     */
+    public Map<Long, Set<String>> toMap() {
+        return deepCopy(id2paths);
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/main/org/apache/zookeeper/server/watch/WatchesSummary.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/watch/WatchesSummary.java b/src/java/main/org/apache/zookeeper/server/watch/WatchesSummary.java
new file mode 100644
index 0000000..b2449ba
--- /dev/null
+++ b/src/java/main/org/apache/zookeeper/server/watch/WatchesSummary.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.watch;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * A summary of watch information. This class is immutable.
+ */
+public class WatchesSummary {
+
+    /**
+     * The key in the map returned by {@link #toMap()} for the number of
+     * connections.
+     */
+    public static final String KEY_NUM_CONNECTIONS = "num_connections";
+    /**
+     * The key in the map returned by {@link #toMap()} for the number of paths.
+     */
+    public static final String KEY_NUM_PATHS = "num_paths";
+    /**
+     * The key in the map returned by {@link #toMap()} for the total number of
+     * watches.
+     */
+    public static final String KEY_NUM_TOTAL_WATCHES = "num_total_watches";
+
+    private final int numConnections;
+    private final int numPaths;
+    private final int totalWatches;
+
+    /**
+     * Creates a new summary.
+     *
+     * @param numConnections the number of sessions that have set watches
+     * @param numPaths the number of paths that have watches set on them
+     * @param totalWatches the total number of watches set
+     */
+    WatchesSummary(int numConnections, int numPaths, int totalWatches) {
+        this.numConnections = numConnections;
+        this.numPaths = numPaths;
+        this.totalWatches = totalWatches;
+    }
+
+    /**
+     * Gets the number of connections (sessions) that have set watches.
+     *
+     * @return number of connections
+     */
+    public int getNumConnections() {
+        return numConnections;
+    }
+    /**
+     * Gets the number of paths that have watches set on them.
+     *
+     * @return number of paths
+     */
+    public int getNumPaths() {
+        return numPaths;
+    }
+    /**
+     * Gets the total number of watches set.
+     *
+     * @return total watches
+     */
+    public int getTotalWatches() {
+        return totalWatches;
+    }
+
+    /**
+     * Converts this summary to a map. The returned map is mutable, and changes
+     * to it do not reflect back into this summary.
+     *
+     * @return map representation of summary
+     */
+    public Map<String, Object> toMap() {
+        Map<String, Object> summary = new LinkedHashMap<String, Object>();
+        summary.put(KEY_NUM_CONNECTIONS, numConnections);
+        summary.put(KEY_NUM_PATHS, numPaths);
+        summary.put(KEY_NUM_TOTAL_WATCHES, totalWatches);
+        return summary;
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/config/findbugsExcludeFile.xml
----------------------------------------------------------------------
diff --git a/src/java/test/config/findbugsExcludeFile.xml b/src/java/test/config/findbugsExcludeFile.xml
index 4ab5a5e..a3f45a6 100644
--- a/src/java/test/config/findbugsExcludeFile.xml
+++ b/src/java/test/config/findbugsExcludeFile.xml
@@ -53,7 +53,13 @@
     <Method name="run" />
     <Bug pattern="DM_EXIT" />
   </Match>
-  
+
+   <!-- Failed to create watch manager is a unrecoverable error -->
+   <Match>
+     <Class name="org.apache.zookeeper.server.DataTree" />
+     <Bug pattern="DM_EXIT" />
+   </Match>
+
   <Match>
     <Package name="org.apache.jute.compiler.generated" />
   </Match>
@@ -85,7 +91,7 @@
 
   <Match>
     <Class name="org.apache.zookeeper.server.DataNode"/>
-      <Field name="children"/> 
+      <Field name="children"/>
       <Bug code="IS"/>
   </Match>
  <Match>
@@ -98,6 +104,15 @@
     <Field name="serverStats"/>
     <Bug code="IS"/>
   </Match>
+
+  <!-- The iterate function is non-thread safe, the caller will synchronize
+       on the BitHHashSet object -->
+  <Match>
+    <Class name="org.apache.zookeeper.server.util.BitHashSet" />
+    <Field name="elementCount" />
+    <Bug code="IS" />
+  </Match>
+
   <Match>
      <Class name="org.apache.zookeeper.server.quorum.LearnerSessionTracker"/>
        <Bug code="UrF"/>
@@ -111,7 +126,7 @@
   <!-- these are old classes just for upgrading and should go away -->
   <Match>
     <Class name="org.apache.zookeeper.server.upgrade.DataNodeV1"/>
-  </Match> 
+  </Match>
 
   <Match>
     <Class name="org.apache.zookeeper.server.upgrade.DataTreeV1"/>
@@ -134,6 +149,23 @@
     </Or>
   </Match>
 
+  <!-- Synchronize on the AtomicInteger to do wait/notify, but not relying
+       on the synchronization to control the AtomicInteger value update,
+       so it's not a problem -->
+  <Match>
+    <Class name="org.apache.zookeeper.server.watch.WatcherCleaner" />
+    <Bug code="JLM" />
+    <Method name="addDeadWatcher" />
+  </Match>
+
+  <Match>
+    <Class name="org.apache.zookeeper.server.watch.WatcherCleaner$1" />
+    <Bug code="JLM" />
+    <Method name="doWork" />
+  </Match>
+
+
+
   <Match>
     <Class name="org.apache.zookeeper.server.quorum.QuorumPeer"/>
     <Bug pattern="OS_OPEN_STREAM" />

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/WatchesPathReportTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/WatchesPathReportTest.java b/src/java/test/org/apache/zookeeper/server/WatchesPathReportTest.java
deleted file mode 100644
index c0b107d..0000000
--- a/src/java/test/org/apache/zookeeper/server/WatchesPathReportTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import org.apache.zookeeper.ZKTestCase;
-import org.junit.Before;
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-public class WatchesPathReportTest extends ZKTestCase {
-    private Map<String, Set<Long>> m;
-    private WatchesPathReport r;
-    @Before public void setUp() {
-        m = new HashMap<String, Set<Long>>();
-        Set<Long> s = new HashSet<Long>();
-        s.add(101L);
-        s.add(102L);
-        m.put("path1", s);
-        s = new HashSet<Long>();
-        s.add(201L);
-        m.put("path2", s);
-        r = new WatchesPathReport(m);
-    }
-    @Test public void testHasSessions() {
-        assertTrue(r.hasSessions("path1"));
-        assertTrue(r.hasSessions("path2"));
-        assertFalse(r.hasSessions("path3"));
-    }
-    @Test public void testGetSessions() {
-        Set<Long> s = r.getSessions("path1");
-        assertEquals(2, s.size());
-        assertTrue(s.contains(101L));
-        assertTrue(s.contains(102L));
-        s = r.getSessions("path2");
-        assertEquals(1, s.size());
-        assertTrue(s.contains(201L));
-        assertNull(r.getSessions("path3"));
-    }
-    @Test public void testToMap() {
-        assertEquals(m, r.toMap());
-    }
-}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/WatchesReportTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/WatchesReportTest.java b/src/java/test/org/apache/zookeeper/server/WatchesReportTest.java
deleted file mode 100644
index 7f0343b..0000000
--- a/src/java/test/org/apache/zookeeper/server/WatchesReportTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import org.apache.zookeeper.ZKTestCase;
-import org.junit.Before;
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-public class WatchesReportTest extends ZKTestCase {
-    private Map<Long, Set<String>> m;
-    private WatchesReport r;
-    @Before public void setUp() {
-        m = new HashMap<Long, Set<String>>();
-        Set<String> s = new HashSet<String>();
-        s.add("path1a");
-        s.add("path1b");
-        m.put(1L, s);
-        s = new HashSet<String>();
-        s.add("path2a");
-        m.put(2L, s);
-        r = new WatchesReport(m);
-    }
-    @Test public void testHasPaths() {
-        assertTrue(r.hasPaths(1L));
-        assertTrue(r.hasPaths(2L));
-        assertFalse(r.hasPaths(3L));
-    }
-    @Test public void testGetPaths() {
-        Set<String> s = r.getPaths(1L);
-        assertEquals(2, s.size());
-        assertTrue(s.contains("path1a"));
-        assertTrue(s.contains("path1b"));
-        s = r.getPaths(2L);
-        assertEquals(1, s.size());
-        assertTrue(s.contains("path2a"));
-        assertNull(r.getPaths(3L));
-    }
-    @Test public void testToMap() {
-        assertEquals(m, r.toMap());
-    }
-}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/WatchesSummaryTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/WatchesSummaryTest.java b/src/java/test/org/apache/zookeeper/server/WatchesSummaryTest.java
deleted file mode 100644
index d679065..0000000
--- a/src/java/test/org/apache/zookeeper/server/WatchesSummaryTest.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.zookeeper.server;
-
-import java.util.Map;
-import org.apache.zookeeper.ZKTestCase;
-import org.junit.Before;
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-public class WatchesSummaryTest extends ZKTestCase {
-    private WatchesSummary s;
-    @Before public void setUp() {
-        s = new WatchesSummary(1, 2, 3);
-    }
-    @Test public void testGetters() {
-        assertEquals(1, s.getNumConnections());
-        assertEquals(2, s.getNumPaths());
-        assertEquals(3, s.getTotalWatches());
-    }
-    @Test public void testToMap() {
-        Map<String, Object> m = s.toMap();
-        assertEquals(3, m.size());
-        assertEquals(Integer.valueOf(1), m.get(WatchesSummary.KEY_NUM_CONNECTIONS));
-        assertEquals(Integer.valueOf(2), m.get(WatchesSummary.KEY_NUM_PATHS));
-        assertEquals(Integer.valueOf(3), m.get(WatchesSummary.KEY_NUM_TOTAL_WATCHES));
-    }
-}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/util/BitHashSetTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/util/BitHashSetTest.java b/src/java/test/org/apache/zookeeper/server/util/BitHashSetTest.java
new file mode 100644
index 0000000..a70eaa5
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/util/BitHashSetTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.util;
+
+import java.util.Set;
+import java.util.HashSet;
+import java.util.List;
+import java.util.ArrayList;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Test;
+import org.junit.Assert;
+
+public class BitHashSetTest extends ZKTestCase {
+
+    @Test
+    public void testAddWatchBit() {
+        int watcherCacheSize = 1;
+        BitHashSet ws = new BitHashSet(watcherCacheSize);
+        Assert.assertTrue(ws.add(1));
+        Assert.assertEquals(1, ws.size());
+        Assert.assertEquals(1, ws.cachedSize());
+
+        List<Integer> actualBits = new ArrayList<Integer>();
+
+        for (int bit: ws) {
+            actualBits.add(bit);
+        }
+        Assert.assertArrayEquals(
+            new Integer[] {1},
+            actualBits.toArray(new Integer[actualBits.size()]));
+
+        // add the same bit again
+        Assert.assertFalse(ws.add(1));
+        Assert.assertEquals(1, ws.size());
+        Assert.assertEquals(1, ws.cachedSize());
+
+        // add another bit, make sure there there is only 1 bit cached
+        Assert.assertTrue(ws.add(2));
+        Assert.assertEquals(2, ws.size());
+        Assert.assertEquals(1, ws.cachedSize());
+
+        Assert.assertTrue(ws.contains(1));
+
+        actualBits.clear();
+        for (int bit: ws) {
+            actualBits.add(bit);
+        }
+        Assert.assertArrayEquals(
+            new Integer[] {1, 2},
+            actualBits.toArray(new Integer[actualBits.size()]));
+    }
+
+    @Test
+    public void testRemoveWatchBit() {
+        int watcherCacheSize = 1;
+        BitHashSet ws = new BitHashSet(watcherCacheSize);
+        ws.add(1);
+        ws.add(2);
+
+        Assert.assertTrue(ws.contains(1));
+        Assert.assertTrue(ws.contains(2));
+
+        ws.remove(1);
+        Assert.assertFalse(ws.contains(1));
+        Assert.assertEquals(1, ws.size());
+        Assert.assertEquals(0, ws.cachedSize());
+
+        List<Integer> actualBits = new ArrayList<Integer>();
+
+        for (int bit: ws) {
+            actualBits.add(bit);
+        }
+        Assert.assertArrayEquals(
+            new Integer[] {2},
+            actualBits.toArray(new Integer[actualBits.size()]));
+
+        ws.add(3);
+        Assert.assertEquals(2, ws.size());
+        Assert.assertEquals(1, ws.cachedSize());
+
+        actualBits.clear();
+        for (int bit: ws) {
+            actualBits.add(bit);
+        }
+        Assert.assertArrayEquals(
+            new Integer[] {2, 3},
+            actualBits.toArray(new Integer[actualBits.size()]));
+
+        ws.remove(2);
+        ws.remove(3);
+
+        Assert.assertEquals(0, ws.size());
+        Assert.assertEquals(0, ws.cachedSize());
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/util/BitMapTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/util/BitMapTest.java b/src/java/test/org/apache/zookeeper/server/util/BitMapTest.java
new file mode 100644
index 0000000..eca0f2d
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/util/BitMapTest.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.server.util;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Test;
+import org.junit.Assert;
+
+public class BitMapTest extends ZKTestCase {
+
+    @Test
+    public void testAddAndRemove() {
+        BitMap<String> bitMap = new BitMap<String>();
+        String v1 = new String("v1");
+        Integer bit = bitMap.add(v1);
+
+        Assert.assertEquals(1, bitMap.size());
+        Assert.assertTrue(bit >= 0);
+        Assert.assertEquals(v1, bitMap.get(bit));
+        Assert.assertEquals(bit, bitMap.getBit(v1));
+
+        // add the same value again
+        Integer newBit = bitMap.add(v1);
+        Assert.assertEquals(bit, newBit);
+        Assert.assertEquals(1, bitMap.size());
+
+        String v2 = new String("v2");
+        Integer v2Bit = bitMap.add(v2);
+        Assert.assertEquals(2, bitMap.size());
+        Assert.assertNotEquals(v2Bit, bit);
+
+        // remove by value
+        bitMap.remove(v1);
+        Assert.assertEquals(1, bitMap.size());
+        Assert.assertNull(bitMap.get(bit));
+        Assert.assertNull(bitMap.getBit(v1));
+
+        // remove by bit
+        bitMap.remove(v2Bit);
+        Assert.assertEquals(0, bitMap.size());
+        Assert.assertNull(bitMap.get(v2Bit));
+        Assert.assertNull(bitMap.getBit(v2));
+    }
+
+    @Test
+    public void testBitReuse() {
+        BitMap<String> bitMap = new BitMap<String>();
+        int v1Bit = bitMap.add("v1");
+        int v2Bit = bitMap.add("v2");
+        int v3Bit = bitMap.add("v3");
+        bitMap.remove(v2Bit);
+
+        int v4Bit = bitMap.add("v4");
+
+        Assert.assertEquals(v4Bit, v2Bit);
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatchManagerTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatchManagerTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatchManagerTest.java
new file mode 100644
index 0000000..f6a229b
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatchManagerTest.java
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.server.DumbWatcher;
+import org.apache.zookeeper.server.ServerCnxn;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(Parameterized.class)
+public class WatchManagerTest extends ZKTestCase {
+    protected static final Logger LOG = LoggerFactory.getLogger(WatchManagerTest.class);
+
+    private static final String PATH_PREFIX = "path";
+
+    private ConcurrentHashMap<Integer, DumbWatcher> watchers;
+    private Random r;
+    private String className;
+
+    public WatchManagerTest(String className) {
+        this.className = className;
+    }
+
+    @Parameterized.Parameters
+    public static List<Object []> data() {
+        return Arrays.asList(new Object [][] {
+            {WatchManager.class.getName()},
+            {WatchManagerOptimized.class.getName()}
+        });
+    }
+
+    @Before
+    public void setUp() {
+        watchers = new ConcurrentHashMap<Integer, DumbWatcher>();
+        r = new Random(System.nanoTime());
+    }
+
+    public IWatchManager getWatchManager() throws IOException {
+        System.setProperty(WatchManagerFactory.ZOOKEEPER_WATCH_MANAGER_NAME, className);
+        return WatchManagerFactory.createWatchManager();
+    }
+
+    public DumbWatcher createOrGetWatcher(int watcherId) {
+        if (!watchers.containsKey(watcherId)) {
+            DumbWatcher watcher = new DumbWatcher(watcherId);
+            watchers.putIfAbsent(watcherId, watcher);
+        }
+        return watchers.get(watcherId);
+    }
+
+    public class AddWatcherWorker extends Thread {
+
+        private final IWatchManager manager;
+        private final int paths;
+        private final int watchers;
+        private final AtomicInteger watchesAdded;
+        private volatile boolean stopped = false;
+
+        public AddWatcherWorker(IWatchManager manager,
+                int paths, int watchers, AtomicInteger watchesAdded) {
+            this.manager = manager;
+            this.paths = paths;
+            this.watchers = watchers;
+            this.watchesAdded = watchesAdded;
+        }
+
+        @Override
+        public void run() {
+            while (!stopped) {
+                String path = PATH_PREFIX + r.nextInt(paths);
+                Watcher watcher = createOrGetWatcher(r.nextInt(watchers));
+                if (manager.addWatch(path, watcher)) {
+                    watchesAdded.addAndGet(1);
+                }
+            }
+        }
+
+        public void shutdown() {
+            stopped = true;
+        }
+    }
+
+    public class WatcherTriggerWorker extends Thread {
+
+        private final IWatchManager manager;
+        private final int paths;
+        private final AtomicInteger triggeredCount;
+        private volatile boolean stopped = false;
+
+        public WatcherTriggerWorker(IWatchManager manager,
+                int paths, AtomicInteger triggeredCount) {
+            this.manager = manager;
+            this.paths = paths;
+            this.triggeredCount = triggeredCount;
+        }
+
+        @Override
+        public void run() {
+            while (!stopped) {
+                String path = PATH_PREFIX + r.nextInt(paths);
+                WatcherOrBitSet s = manager.triggerWatch(
+                        path, EventType.NodeDeleted);
+                if (s != null) {
+                    triggeredCount.addAndGet(s.size());
+                }
+                try {
+                    Thread.sleep(r.nextInt(10));
+                } catch (InterruptedException e) {}
+            }
+        }
+
+        public void shutdown() {
+            stopped = true;
+        }
+    }
+
+    public class RemoveWatcherWorker extends Thread {
+
+        private final IWatchManager manager;
+        private final int paths;
+        private final int watchers;
+        private final AtomicInteger watchesRemoved;
+        private volatile boolean stopped = false;
+
+        public RemoveWatcherWorker(IWatchManager manager,
+                int paths, int watchers, AtomicInteger watchesRemoved) {
+            this.manager = manager;
+            this.paths = paths;
+            this.watchers = watchers;
+            this.watchesRemoved = watchesRemoved;
+        }
+
+        @Override
+        public void run() {
+            while (!stopped) {
+                String path = PATH_PREFIX + r.nextInt(paths);
+                Watcher watcher = createOrGetWatcher(r.nextInt(watchers));
+                if (manager.removeWatcher(path, watcher)) {
+                    watchesRemoved.addAndGet(1);
+                }
+                try {
+                    Thread.sleep(r.nextInt(10));
+                } catch (InterruptedException e) {}
+            }
+        }
+
+        public void shutdown() {
+            stopped = true;
+        }
+
+    }
+
+    public class CreateDeadWatchersWorker extends Thread {
+
+        private final IWatchManager manager;
+        private final int watchers;
+        private final Set<Watcher> removedWatchers;
+        private volatile boolean stopped = false;
+
+        public CreateDeadWatchersWorker(IWatchManager manager,
+                int watchers, Set<Watcher> removedWatchers) {
+            this.manager = manager;
+            this.watchers = watchers;
+            this.removedWatchers = removedWatchers;
+        }
+
+        @Override
+        public void run() {
+            while (!stopped) {
+                DumbWatcher watcher = createOrGetWatcher(r.nextInt(watchers));
+                watcher.setStale();
+                manager.removeWatcher(watcher);
+                synchronized (removedWatchers) {
+                    removedWatchers.add(watcher);
+                }
+                try {
+                    Thread.sleep(r.nextInt(10));
+                } catch (InterruptedException e) {}
+            }
+        }
+
+        public void shutdown() {
+            stopped = true;
+        }
+
+    }
+
+    /**
+     * Concurrently add and trigger watch, make sure the watches triggered
+     * are the same as the number added.
+     */
+    @Test(timeout = 90000)
+    public void testAddAndTriggerWatcher() throws IOException {
+        IWatchManager manager = getWatchManager();
+        int paths = 1;
+        int watchers = 10000;
+
+        // 1. start 5 workers to trigger watchers on that path
+        //    count all the watchers have been fired
+        AtomicInteger watchTriggered = new AtomicInteger();
+        List<WatcherTriggerWorker> triggerWorkers =
+                new ArrayList<WatcherTriggerWorker>();
+        for (int i = 0; i < 5; i++) {
+            WatcherTriggerWorker worker =
+                    new WatcherTriggerWorker(manager, paths, watchTriggered);
+            triggerWorkers.add(worker);
+            worker.start();
+        }
+
+        // 2. start 5 workers to add different watchers on the same path
+        //    count all the watchers being added
+        AtomicInteger watchesAdded = new AtomicInteger();
+        List<AddWatcherWorker> addWorkers = new ArrayList<AddWatcherWorker>();
+        for (int i = 0; i < 5; i++) {
+            AddWatcherWorker worker = new AddWatcherWorker(
+                    manager, paths, watchers, watchesAdded);
+            addWorkers.add(worker);
+            worker.start();
+        }
+
+        while (watchesAdded.get() < 100000) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {}
+        }
+
+        // 3. stop all the addWorkers
+        for (AddWatcherWorker worker: addWorkers) {
+            worker.shutdown();
+        }
+
+        // 4. running the trigger worker a bit longer to make sure
+        //    all watchers added are fired
+        try {
+            Thread.sleep(500);
+        } catch (InterruptedException e) {}
+
+        // 5. stop all triggerWorkers
+        for (WatcherTriggerWorker worker: triggerWorkers) {
+            worker.shutdown();
+        }
+
+        // 6. make sure the total watch triggered is same as added
+        Assert.assertTrue(watchesAdded.get() > 0);
+        Assert.assertEquals(watchesAdded.get(), watchTriggered.get());
+    }
+
+    /**
+     * Concurrently add and remove watch, make sure the watches left +
+     * the watches removed are equal to the total added watches.
+     */
+    @Test(timeout = 90000)
+    public void testRemoveWatcherOnPath() throws IOException {
+        IWatchManager manager = getWatchManager();
+        int paths = 10;
+        int watchers = 10000;
+
+        // 1. start 5 workers to remove watchers on those path
+        //    record the watchers have been removed
+        AtomicInteger watchesRemoved = new AtomicInteger();
+        List<RemoveWatcherWorker> removeWorkers =
+                new ArrayList<RemoveWatcherWorker>();
+        for (int i = 0; i < 5; i++) {
+            RemoveWatcherWorker worker =
+                    new RemoveWatcherWorker(manager, paths, watchers, watchesRemoved);
+            removeWorkers.add(worker);
+            worker.start();
+        }
+
+        // 2. start 5 workers to add different watchers on different path
+        //    record the watchers have been added
+        AtomicInteger watchesAdded = new AtomicInteger();
+        List<AddWatcherWorker> addWorkers = new ArrayList<AddWatcherWorker>();
+        for (int i = 0; i < 5; i++) {
+            AddWatcherWorker worker = new AddWatcherWorker(
+                    manager, paths, watchers, watchesAdded);
+            addWorkers.add(worker);
+            worker.start();
+        }
+
+        while (watchesAdded.get() < 100000) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {}
+        }
+
+        // 3. stop all workers
+        for (RemoveWatcherWorker worker: removeWorkers) {
+            worker.shutdown();
+        }
+        for (AddWatcherWorker worker: addWorkers) {
+            worker.shutdown();
+        }
+
+        // 4. sleep for a while to make sure all the thread exited
+        try {
+            Thread.sleep(500);
+        } catch (InterruptedException e) {}
+
+        // 5. make sure left watches + removed watches = added watches
+        Assert.assertTrue(watchesAdded.get() > 0);
+        Assert.assertTrue(watchesRemoved.get() > 0);
+        Assert.assertTrue(manager.size() > 0);
+        Assert.assertEquals(
+                watchesAdded.get(), watchesRemoved.get() + manager.size());
+    }
+
+    /**
+     * Concurrently add watch while close the watcher to simulate the
+     * client connections closed on prod.
+     */
+    @Test(timeout = 90000)
+    public void testDeadWatchers() throws IOException {
+        System.setProperty("zookeeper.watcherCleanThreshold", "10");
+        System.setProperty("zookeeper.watcherCleanIntervalInSeconds", "1");
+
+        IWatchManager manager = getWatchManager();
+        int paths = 1;
+        int watchers = 100000;
+
+        // 1. start 5 workers to randomly mark those watcher as dead
+        //    and remove them from watch manager
+        Set<Watcher> deadWatchers = new HashSet<Watcher>();
+        List<CreateDeadWatchersWorker> deadWorkers =
+                new ArrayList<CreateDeadWatchersWorker>();
+        for (int i = 0; i < 5; i++) {
+            CreateDeadWatchersWorker worker = new CreateDeadWatchersWorker(
+                    manager, watchers, deadWatchers);
+            deadWorkers.add(worker);
+            worker.start();
+        }
+
+        // 2. start 5 workers to add different watchers on the same path
+        AtomicInteger watchesAdded = new AtomicInteger();
+        List<AddWatcherWorker> addWorkers = new ArrayList<AddWatcherWorker>();
+        for (int i = 0; i < 5; i++) {
+            AddWatcherWorker worker = new AddWatcherWorker(
+                    manager, paths, watchers, watchesAdded);
+            addWorkers.add(worker);
+            worker.start();
+        }
+
+        while (watchesAdded.get() < 50000) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {}
+        }
+
+        // 3. stop all workers
+        for (CreateDeadWatchersWorker worker: deadWorkers) {
+            worker.shutdown();
+        }
+        for (AddWatcherWorker worker: addWorkers) {
+            worker.shutdown();
+        }
+
+        // 4. sleep for a while to make sure all the thread exited
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {}
+
+        // 5. make sure the dead watchers are not in the existing watchers
+        WatchesReport existingWatchers = manager.getWatches();
+        for (Watcher w: deadWatchers) {
+            Assert.assertFalse(
+                    existingWatchers.hasPaths(((ServerCnxn) w).getSessionId()));
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatcherCleanerTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatcherCleanerTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatcherCleanerTest.java
new file mode 100644
index 0000000..d315232
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatcherCleanerTest.java
@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.util.Set;
+import java.util.HashSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.common.Time;
+import org.junit.Test;
+import org.junit.Assert;
+
+public class WatcherCleanerTest extends ZKTestCase {
+
+    public static class MyDeadWatcherListener implements IDeadWatcherListener {
+
+        private CountDownLatch latch;
+        private int delayMs;
+        private Set<Integer> deadWatchers = new HashSet<Integer>();
+
+        public void setCountDownLatch(CountDownLatch latch) {
+            this.latch = latch;
+        }
+
+        public void setDelayMs(int delayMs) {
+            this.delayMs = delayMs;
+        }
+
+        @Override
+        public void processDeadWatchers(Set<Integer> deadWatchers) {
+            if (delayMs > 0) {
+                try {
+                    Thread.sleep(delayMs);
+                } catch (InterruptedException e) {}
+            }
+            this.deadWatchers.clear();
+            this.deadWatchers.addAll(deadWatchers);
+            latch.countDown();
+        }
+
+        public Set<Integer> getDeadWatchers() {
+            return deadWatchers;
+        }
+
+        public boolean wait(int maxWaitMs) {
+            try {
+                return latch.await(maxWaitMs, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {}
+            return false;
+        }
+    }
+
+    @Test
+    public void testProcessDeadWatchersBasedOnThreshold() {
+        MyDeadWatcherListener listener = new MyDeadWatcherListener();
+        int threshold = 3;
+        WatcherCleaner cleaner = new WatcherCleaner(listener, threshold, 60, 1, 10);
+        cleaner.start();
+
+        int i = 0;
+        while (i++ < threshold - 1) {
+            cleaner.addDeadWatcher(i);
+        }
+        // not trigger processDeadWatchers yet
+        Assert.assertEquals(0, listener.getDeadWatchers().size());
+
+        listener.setCountDownLatch(new CountDownLatch(1));
+        // add another dead watcher to trigger the process
+        cleaner.addDeadWatcher(i);
+        Assert.assertTrue(listener.wait(1000));
+        Assert.assertEquals(threshold, listener.getDeadWatchers().size());
+    }
+
+    @Test
+    public void testProcessDeadWatchersBasedOnTime() {
+        MyDeadWatcherListener listener = new MyDeadWatcherListener();
+        WatcherCleaner cleaner = new WatcherCleaner(listener, 10, 1, 1, 10);
+        cleaner.start();
+
+        cleaner.addDeadWatcher(1);
+        // not trigger processDeadWatchers yet
+        Assert.assertEquals(0, listener.getDeadWatchers().size());
+
+        listener.setCountDownLatch(new CountDownLatch(1));
+        Assert.assertTrue(listener.wait(2000));
+        Assert.assertEquals(1, listener.getDeadWatchers().size());
+
+        // won't trigger event if there is no dead watchers
+        listener.setCountDownLatch(new CountDownLatch(1));
+        Assert.assertFalse(listener.wait(2000));
+    }
+
+    @Test
+    public void testMaxInProcessingDeadWatchers() {
+        MyDeadWatcherListener listener = new MyDeadWatcherListener();
+        int delayMs = 1000;
+        listener.setDelayMs(delayMs);
+        WatcherCleaner cleaner = new WatcherCleaner(listener, 1, 60, 1, 1);
+        cleaner.start();
+
+        listener.setCountDownLatch(new CountDownLatch(2));
+
+        long startTime = Time.currentElapsedTime();
+        cleaner.addDeadWatcher(1);
+        cleaner.addDeadWatcher(2);
+        long time = Time.currentElapsedTime() - startTime;
+        System.out.println("time used " + time);
+        Assert.assertTrue(Time.currentElapsedTime() - startTime >= delayMs);
+        Assert.assertTrue(listener.wait(5000));
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatcherOrBitSetTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatcherOrBitSetTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatcherOrBitSetTest.java
new file mode 100644
index 0000000..4b7fbd5
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatcherOrBitSetTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.util.Set;
+import java.util.HashSet;
+
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.server.DumbWatcher;
+import org.apache.zookeeper.server.util.BitHashSet;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Test;
+import org.junit.Assert;
+
+public class WatcherOrBitSetTest extends ZKTestCase {
+
+    @Test
+    public void testWatcherSet() {
+        Set<Watcher> wset = new HashSet<Watcher>();
+        WatcherOrBitSet hashSet = new WatcherOrBitSet(wset);
+        Assert.assertEquals(0, hashSet.size());
+
+        DumbWatcher w1 = new DumbWatcher();
+        Assert.assertFalse(hashSet.contains(w1));
+        wset.add(w1);
+        Assert.assertTrue(hashSet.contains(w1));
+        Assert.assertEquals(1, hashSet.size());
+        Assert.assertFalse(hashSet.contains(1));
+    }
+
+    @Test
+    public void testBitSet() {
+        BitHashSet bset = new BitHashSet(0);
+        WatcherOrBitSet bitSet = new WatcherOrBitSet(bset);
+        Assert.assertEquals(0, bitSet.size());
+
+        Integer bit = new Integer(1);
+        Assert.assertFalse(bitSet.contains(1));
+        Assert.assertFalse(bitSet.contains(bit));
+
+        bset.add(bit);
+        Assert.assertTrue(bitSet.contains(1));
+        Assert.assertTrue(bitSet.contains(bit));
+        Assert.assertEquals(1, bitSet.size());
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatchesPathReportTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatchesPathReportTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatchesPathReportTest.java
new file mode 100644
index 0000000..34e3789
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatchesPathReportTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class WatchesPathReportTest extends ZKTestCase {
+    private Map<String, Set<Long>> m;
+    private WatchesPathReport r;
+    @Before public void setUp() {
+        m = new HashMap<String, Set<Long>>();
+        Set<Long> s = new HashSet<Long>();
+        s.add(101L);
+        s.add(102L);
+        m.put("path1", s);
+        s = new HashSet<Long>();
+        s.add(201L);
+        m.put("path2", s);
+        r = new WatchesPathReport(m);
+    }
+    @Test public void testHasSessions() {
+        assertTrue(r.hasSessions("path1"));
+        assertTrue(r.hasSessions("path2"));
+        assertFalse(r.hasSessions("path3"));
+    }
+    @Test public void testGetSessions() {
+        Set<Long> s = r.getSessions("path1");
+        assertEquals(2, s.size());
+        assertTrue(s.contains(101L));
+        assertTrue(s.contains(102L));
+        s = r.getSessions("path2");
+        assertEquals(1, s.size());
+        assertTrue(s.contains(201L));
+        assertNull(r.getSessions("path3"));
+    }
+    @Test public void testToMap() {
+        assertEquals(m, r.toMap());
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatchesReportTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatchesReportTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatchesReportTest.java
new file mode 100644
index 0000000..237583a
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatchesReportTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class WatchesReportTest extends ZKTestCase {
+    private Map<Long, Set<String>> m;
+    private WatchesReport r;
+    @Before public void setUp() {
+        m = new HashMap<Long, Set<String>>();
+        Set<String> s = new HashSet<String>();
+        s.add("path1a");
+        s.add("path1b");
+        m.put(1L, s);
+        s = new HashSet<String>();
+        s.add("path2a");
+        m.put(2L, s);
+        r = new WatchesReport(m);
+    }
+    @Test public void testHasPaths() {
+        assertTrue(r.hasPaths(1L));
+        assertTrue(r.hasPaths(2L));
+        assertFalse(r.hasPaths(3L));
+    }
+    @Test public void testGetPaths() {
+        Set<String> s = r.getPaths(1L);
+        assertEquals(2, s.size());
+        assertTrue(s.contains("path1a"));
+        assertTrue(s.contains("path1b"));
+        s = r.getPaths(2L);
+        assertEquals(1, s.size());
+        assertTrue(s.contains("path2a"));
+        assertNull(r.getPaths(3L));
+    }
+    @Test public void testToMap() {
+        assertEquals(m, r.toMap());
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/java/test/org/apache/zookeeper/server/watch/WatchesSummaryTest.java
----------------------------------------------------------------------
diff --git a/src/java/test/org/apache/zookeeper/server/watch/WatchesSummaryTest.java b/src/java/test/org/apache/zookeeper/server/watch/WatchesSummaryTest.java
new file mode 100644
index 0000000..35956f1
--- /dev/null
+++ b/src/java/test/org/apache/zookeeper/server/watch/WatchesSummaryTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.watch;
+
+import java.util.Map;
+import org.apache.zookeeper.ZKTestCase;
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class WatchesSummaryTest extends ZKTestCase {
+    private WatchesSummary s;
+    @Before public void setUp() {
+        s = new WatchesSummary(1, 2, 3);
+    }
+    @Test public void testGetters() {
+        assertEquals(1, s.getNumConnections());
+        assertEquals(2, s.getNumPaths());
+        assertEquals(3, s.getTotalWatches());
+    }
+    @Test public void testToMap() {
+        Map<String, Object> m = s.toMap();
+        assertEquals(3, m.size());
+        assertEquals(Integer.valueOf(1), m.get(WatchesSummary.KEY_NUM_CONNECTIONS));
+        assertEquals(Integer.valueOf(2), m.get(WatchesSummary.KEY_NUM_PATHS));
+        assertEquals(Integer.valueOf(3), m.get(WatchesSummary.KEY_NUM_TOTAL_WATCHES));
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/test/java/bench/org/apache/zookeeper/BenchMain.java
----------------------------------------------------------------------
diff --git a/src/test/java/bench/org/apache/zookeeper/BenchMain.java b/src/test/java/bench/org/apache/zookeeper/BenchMain.java
new file mode 100644
index 0000000..8e370c0
--- /dev/null
+++ b/src/test/java/bench/org/apache/zookeeper/BenchMain.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;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+
+public class BenchMain {
+    public static void main(String args[]) throws Exception {
+        org.openjdk.jmh.Main.main(args);
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/src/test/java/bench/org/apache/zookeeper/server/watch/WatchBench.java
----------------------------------------------------------------------
diff --git a/src/test/java/bench/org/apache/zookeeper/server/watch/WatchBench.java b/src/test/java/bench/org/apache/zookeeper/server/watch/WatchBench.java
new file mode 100644
index 0000000..0510df7
--- /dev/null
+++ b/src/test/java/bench/org/apache/zookeeper/server/watch/WatchBench.java
@@ -0,0 +1,300 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.watch;
+
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.server.watch.IWatchManager;
+import org.apache.zookeeper.server.DumbWatcher;
+
+import org.openjdk.jmh.annotations.*;
+
+import java.util.concurrent.TimeUnit;
+
+@Fork(3)
+public class WatchBench {
+
+    static final String pathPrefix = "/reasonably/long/path/";
+    static final EventType event = EventType.NodeDataChanged;
+
+    static IWatchManager createWatchManager(String className) throws Exception {
+        Class clazz = Class.forName(
+                "org.apache.zookeeper.server.watch." + className);
+        return (IWatchManager) clazz.newInstance();
+    }
+
+    static void forceGC() {
+        int gcTimes = 3;
+        for (int i = 0; i < gcTimes; i++) {
+            try {
+                System.gc();
+                Thread.currentThread().sleep(1000);
+
+                System.runFinalization();
+                Thread.currentThread().sleep(1000);
+            } catch (InterruptedException ex) { /* ignore */ }
+        }
+    }
+
+    static long getMemoryUse() {
+        forceGC();
+        long totalMem = Runtime.getRuntime().totalMemory();
+
+        forceGC();
+        long freeMem = Runtime.getRuntime().freeMemory();
+        return totalMem - freeMem;
+    }
+
+    @State(Scope.Benchmark)
+    public static class IterationState {
+
+        @Param({"WatchManager", "WatchManagerOptimized"})
+        public String watchManagerClass;
+
+        @Param({"10000"})
+        public int pathCount;
+
+        String[] paths;
+
+        long watchesAdded = 0;
+        IWatchManager watchManager;
+
+        long memWhenSetup = 0;
+
+        @Setup(Level.Iteration)
+        public void setup() throws Exception {
+            paths = new String[pathCount];
+            for (int i = 0; i < paths.length; i++) {
+                paths[i] = pathPrefix + i;
+            }
+
+            watchesAdded = 0;
+            watchManager = createWatchManager(watchManagerClass);
+
+            memWhenSetup = getMemoryUse();
+        }
+
+        @TearDown(Level.Iteration)
+        public void tearDown() {
+            long memUsed = getMemoryUse() - memWhenSetup;
+            System.out.println("Memory used: " + watchesAdded + " " + memUsed);
+
+            double memPerMillionWatchesMB = memUsed * 1.0 / watchesAdded ;
+            System.out.println(
+                    "Memory used per million watches " +
+                    String.format("%.2f", memPerMillionWatchesMB) + "MB");
+        }
+    }
+
+    /**
+     * Test concenrate watch case where the watcher watches all paths.
+     *
+     * The output of this test will be the average time used to add the
+     * watch to all paths.
+     */
+    @Benchmark
+    @BenchmarkMode(Mode.AverageTime)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    @Warmup(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
+    @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS)
+    public void testAddConcentrateWatch(IterationState state) throws Exception {
+        Watcher watcher = new DumbWatcher();
+
+        // watch all paths
+        for (String path : state.paths) {
+            if (state.watchManager.addWatch(path, watcher)) {
+                state.watchesAdded++;
+            }
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class InvocationState {
+
+        @Param({"WatchManager", "WatchManagerOptimized"})
+        public String watchManagerClass;
+
+        @Param({"1", "1000"})
+        public int pathCount;
+
+        @Param({"1", "1000"})
+        public int watcherCount;
+
+        String[] paths;
+        Watcher[] watchers;
+
+        IWatchManager watchManager;
+
+        @Setup(Level.Invocation)
+        public void setup() throws Exception {
+            initialize();
+            prepare();
+        }
+
+        void initialize() throws Exception {
+            if (paths == null || paths.length != pathCount) {
+                paths = new String[pathCount];
+                for (int i = 0; i < pathCount; i++) {
+                    paths[i] = pathPrefix + i;
+                }
+            }
+
+            if (watchers == null || watchers.length != watcherCount) {
+                watchers = new Watcher[watcherCount];
+                for (int i = 0; i < watcherCount; i++) {
+                    watchers[i] = new DumbWatcher();
+                }
+            }
+            if (watchManager == null ||
+                    !watchManager.getClass().getSimpleName().contains(
+                            watchManagerClass)) {
+                watchManager = createWatchManager(watchManagerClass);
+            }
+        }
+
+        void prepare() {
+            for (String path : paths) {
+                for (Watcher watcher : watchers) {
+                    watchManager.addWatch(path, watcher);
+                }
+            }
+        }
+    }
+
+    /**
+     * Test trigger watches in concenrate case.
+     *
+     * The output of this test is the time used to trigger those watches on
+     * all paths.
+     */
+    @Benchmark
+    @BenchmarkMode(Mode.AverageTime)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    @Warmup(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
+    @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS)
+    public void testTriggerConcentrateWatch(InvocationState state) throws Exception {
+        for (String path : state.paths) {
+            state.watchManager.triggerWatch(path, event);
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class AddSparseWatchState extends InvocationState {
+
+        @Param({"10000"})
+        public int pathCount;
+
+        @Param({"10000"})
+        public int watcherCount;
+
+        long watchesAdded = 0;
+        long memWhenSetup = 0;
+
+        @Override
+        public void prepare() {
+            watchesAdded = 0;
+            memWhenSetup = getMemoryUse();
+        }
+
+        @TearDown(Level.Invocation)
+        public void tearDown() {
+            long memUsed = getMemoryUse() - memWhenSetup;
+            System.out.println("Memory used: " + watchesAdded + " " + memUsed);
+
+            double memPerMillionWatchesMB = memUsed * 1.0 / watchesAdded ;
+            System.out.println(
+                    "Memory used per million sparse watches " +
+                    String.format("%.2f", memPerMillionWatchesMB) + "MB");
+
+            // clear all the watches
+            for (String path : paths) {
+                watchManager.triggerWatch(path, event);
+            }
+        }
+    }
+
+    /**
+     * Test sparse watch case where only one watcher watches all paths, and
+     * only one path being watched by all watchers.
+     *
+     * The output of this test will be the average time used to add those
+     * sparse watches.
+     */
+    @Benchmark
+    @BenchmarkMode(Mode.AverageTime)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    @Warmup(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
+    @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS)
+    public void testAddSparseWatch(AddSparseWatchState state) throws Exception {
+        // All watchers are watching the 1st path
+        for (Watcher watcher : state.watchers) {
+            if (state.watchManager.addWatch(state.paths[0], watcher)) {
+                state.watchesAdded++;
+            }
+        }
+        // The 1st watcher is watching all paths
+        for (String path : state.paths) {
+            if (state.watchManager.addWatch(path, state.watchers[0])) {
+                state.watchesAdded++;
+            }
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class TriggerSparseWatchState extends InvocationState {
+
+        @Param({"10000"})
+        public int pathCount;
+
+        @Param({"10000"})
+        public int watcherCount;
+
+        @Override
+        public void prepare() {
+            // All watchers are watching the 1st path
+            for (Watcher watcher : watchers) {
+                watchManager.addWatch(paths[0], watcher);
+            }
+
+            // The 1st watcher is watching all paths
+            for (String path : paths) {
+                watchManager.addWatch(path, watchers[0]);
+            }
+        }
+    }
+
+
+    /**
+     * Test trigger watches in sparse case.
+     *
+     * The output of this test is the time used to trigger those watches on
+     * all paths.
+     */
+    @Benchmark
+    @BenchmarkMode(Mode.AverageTime)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    @Warmup(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
+    @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS)
+    public void testTriggerSparseWatch(TriggerSparseWatchState state) throws Exception {
+        for (String path : state.paths) {
+            state.watchManager.triggerWatch(path, event);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/zookeeper-contrib/zookeeper-contrib-fatjar/src/main/resources/mainClasses
----------------------------------------------------------------------
diff --git a/zookeeper-contrib/zookeeper-contrib-fatjar/src/main/resources/mainClasses b/zookeeper-contrib/zookeeper-contrib-fatjar/src/main/resources/mainClasses
index 2b0fc83..ba29e89 100644
--- a/zookeeper-contrib/zookeeper-contrib-fatjar/src/main/resources/mainClasses
+++ b/zookeeper-contrib/zookeeper-contrib-fatjar/src/main/resources/mainClasses
@@ -8,3 +8,4 @@ quorumBench:org.apache.zookeeper.server.QuorumBenchmark:A benchmark of just the
 abBench:org.apache.zookeeper.server.quorum.AtomicBroadcastBenchmark:A benchmark of just the atomic broadcast
 ic:org.apache.zookeeper.test.system.InstanceContainer:A container that will instantiate classes as directed by an instance manager
 systest:org.apache.zookeeper.test.system.BaseSysTest:Start system test
+jmh:org.apache.zookeeper.BenchMain:Run jmh micro benchmarks

http://git-wip-us.apache.org/repos/asf/zookeeper/blob/fdde8b00/zookeeper-docs/src/documentation/content/xdocs/zookeeperAdmin.xml
----------------------------------------------------------------------
diff --git a/zookeeper-docs/src/documentation/content/xdocs/zookeeperAdmin.xml b/zookeeper-docs/src/documentation/content/xdocs/zookeeperAdmin.xml
index 50143d4..b7cd21b 100644
--- a/zookeeper-docs/src/documentation/content/xdocs/zookeeperAdmin.xml
+++ b/zookeeper-docs/src/documentation/content/xdocs/zookeeperAdmin.xml
@@ -1025,6 +1025,102 @@ server.3=zoo3:2888:3888</programlisting>
             </listitem>
           </varlistentry>
 
+
+          <varlistentry>
+            <term>watchManaggerName</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.watchManagerName</emphasis>)</para>
+
+              <para><emphasis role="bold">New in 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> New watcher
+                manager WatchManagerOptimized is added to optimize the memory overhead in heavy watch use cases. This
+                config is used to define which watcher manager to be used. Currently, we only support WatchManager and
+                WatchManagerOptimized.</para>
+            </listitem>
+          </varlistentry>
+
+          <varlistentry>
+            <term>watcherCleanThreadsNum</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.watcherCleanThreadsNum</emphasis>)</para>
+
+              <para><emphasis role="bold">New in 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> The new watcher
+                manager WatchManagerOptimized will clean up the dead watchers lazily, this config is used to decide how
+                many thread is used in the WatcherCleaner. More thread usually means larger clean up throughput. The
+                default value is 2, which is good enough even for heavy and continuous session closing/recreating cases.</para>
+            </listitem>
+          </varlistentry>
+
+          <varlistentry>
+            <term>watcherCleanThreshold</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.watcherCleanThreshold</emphasis>)</para>
+
+              <para><emphasis role="bold">New in 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> The new watcher
+                manager WatchManagerOptimized will clean up the dead watchers lazily, the clean up process is relatively
+                heavy, batch processing will reduce the cost and improve the performance. This setting is used to decide
+                the batch size. The default one is 1000, we don't need to change it if there is no memory or clean up
+                speed issue.</para>
+            </listitem>
+          </varlistentry>
+
+          <varlistentry>
+            <term>watcherCleanIntervalInSeconds</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.watcherCleanIntervalInSeconds</emphasis>)</para>
+
+              <para><emphasis role="bold">New in 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> The new watcher
+                manager WatchManagerOptimized will clean up the dead watchers lazily, the clean up process is relatively
+                heavy, batch processing will reduce the cost and improve the performance. Besides watcherCleanThreshold,
+                this setting is used to clean up the dead watchers after certain time even the dead watchers are not larger
+                than watcherCleanThreshold, so that we won't leave the dead watchers there for too long. The default setting
+                is 10 minutes, which usually don't need to be changed.</para>
+            </listitem>
+          </varlistentry>
+
+          <varlistentry>
+            <term>maxInProcessingDeadWatchers</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.maxInProcessingDeadWatchers</emphasis>)</para>
+
+              <para><emphasis role="bold">New in 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> This is used
+                to control how many backlog can we have in the WatcherCleaner, when it reaches this number, it will
+                slow down adding the dead watcher to WatcherCleaner, which will in turn slow down adding and closing
+                watchers, so that we can avoid OOM issue. By default there is no limit, you can set it to values like
+                watcherCleanThreshold * 1000.</para>
+            </listitem>
+          </varlistentry>
+
+          <varlistentry>
+            <term>bitHashCacheSize</term>
+
+            <listitem>
+              <para>(Java system property only: <emphasis
+                    role="bold">zookeeper.bitHashCacheSize</emphasis>)</para>
+
+              <para><emphasis role="bold">New 3.6.0:</emphasis> Added in
+                <ulink url="https://issues.apache.org/jira/browse/ZOOKEEPER-1179">ZOOKEEPER-1179</ulink> This is the
+                setting used to decide the HashSet cache size in the BitHashSet implementation. Without HashSet, we
+                need to use O(N) time to get the elements, N is the bit numbers in elementBits. But we need to
+                keep the size small to make sure it doesn't cost too much in memory, there is a trade off between memory
+                and time complexity. The default value is 10, which seems a relatively reasonable cache size.</para>
+            </listitem>
+          </varlistentry>
+
         </variablelist>
       </section>
 


Mime
View raw message