From 459e9a92d37fe3536a782e1c7783e473bfc23459 Mon Sep 17 00:00:00 2001 From: randgalt Date: Wed, 2 Oct 2019 15:54:16 -0500 Subject: [PATCH 1/5] ZOOKEEPER-1416 Implementation for a persistent, recursive watch addition for ZK. These watches are set via a new method, addPersistentWatch() and are removed via the existing watcher removal methods. Persistent, recursive watches have these characteristics: a) Once set, they do not auto-remove when triggered; b) they trigger for all event types (child, data, etc.) on the node they are registered for and any child znode recursively; c) they are efficiently implemented by using the existing watch internals. A new class PathIterator walks up the path parent-by-parent when checking if a watcher applies. Persistent watcher specific tests are in PersistentWatcherTest.java. --- .../main/resources/markdown/zookeeperOver.md | 5 + .../markdown/zookeeperProgrammers.md | 27 ++- .../src/main/resources/zookeeper.jute | 12 ++ .../java/org/apache/zookeeper/ClientCnxn.java | 25 ++- .../java/org/apache/zookeeper/Watcher.java | 5 +- .../java/org/apache/zookeeper/ZooDefs.java | 4 + .../java/org/apache/zookeeper/ZooKeeper.java | 184 ++++++++++++++++++ .../org/apache/zookeeper/server/DataTree.java | 18 +- .../server/FinalRequestProcessor.java | 34 +++- .../server/PrepRequestProcessor.java | 2 + .../org/apache/zookeeper/server/Request.java | 5 + .../apache/zookeeper/server/ZKDatabase.java | 21 +- .../util/RequestPathMetricsCollector.java | 2 + .../zookeeper/server/watch/IWatchManager.java | 24 +++ .../server/watch/PathParentIterator.java | 106 ++++++++++ .../zookeeper/server/watch/WatchManager.java | 142 ++++++++++---- .../zookeeper/server/watch/WatcherMode.java | 44 +++++ .../server/watch/WatcherModeManager.java | 91 +++++++++ .../server/watch/PathParentIteratorTest.java | 84 ++++++++ .../server/watch/WatchManagerTest.java | 2 +- .../test/PersistentRecursiveWatcherTest.java | 173 ++++++++++++++++ .../zookeeper/test/PersistentWatcherTest.java | 169 ++++++++++++++++ .../zookeeper/test/RecursiveWatchQtyTest.java | 108 ++++++++++ 23 files changed, 1233 insertions(+), 54 deletions(-) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/PathParentIterator.java create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/PathParentIteratorTest.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperOver.md b/zookeeper-docs/src/main/resources/markdown/zookeeperOver.md index 1c00609e6d4..4c60a3de7e2 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperOver.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperOver.md @@ -146,6 +146,11 @@ receives a packet saying that the znode has changed. If the connection between the client and one of the ZooKeeper servers is broken, the client will receive a local notification. +**New in 3.6.0:** Clients can also set +permanent, recursive watches on a znode that are not removed when triggered +and that trigger for changes on the registered znode as well as any children +znodes recursively. + ### Guarantees diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md b/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md index 26e4aa530bf..8d45ca6c650 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md @@ -32,6 +32,7 @@ limitations under the License. * [ZooKeeper Sessions](#ch_zkSessions) * [ZooKeeper Watches](#ch_zkWatches) * [Semantics of Watches](#sc_WatchSemantics) + * [Persistent, Recursive Watches](#sc_WatchPersistentRecursive) * [Remove Watches](#sc_WatchRemoval) * [What ZooKeeper Guarantees about Watches](#sc_WatchGuarantees) * [Things to Remember about Watches](#sc_WatchRememberThese) @@ -640,6 +641,11 @@ general this all occurs transparently. There is one case where a watch may be missed: a watch for the existence of a znode not yet created will be missed if the znode is created and deleted while disconnected. +**New in 3.6.0:** Clients can also set +permanent, recursive watches on a znode that are not removed when triggered +and that trigger for changes on the registered znode as well as any children +znodes recursively. + ### Semantics of Watches @@ -657,6 +663,21 @@ the events that a watch can trigger and the calls that enable them: * **Child event:** Enabled with a call to getChildren. + + +### Persistent, Recursive Watches + +**New in 3.6.0:** There is now a variation on the standard +watch described above whereby you can set a watch that does not get removed when triggered. +Additionally, these watches trigger the event types *NodeCreated*, *NodeDeleted*, and *NodeDataChanged* +and, optionally, recursively for all znodes starting at the znode that the watch is registered for. Note +that *NodeChildrenChanged* events are not triggered for persistent recursive watches as it would be redundant. + +Persistent watches are set using the method *addPersistentWatch()*. The triggering semantics and guarantees +(other than one-time triggering) are the same as standard watches. The only exception regarding events is that +recursive persistent watchers never trigger child changed events as they are redundant. +Persistent watches are removed using *removeWatches()* with watcher type *WatcherType.Any*. + ### Remove Watches @@ -671,6 +692,8 @@ successful watch removal. Watcher which was added with a call to getChildren. * **Data Remove event:** Watcher which was added with a call to exists or getData. +* **Persistent Remove event:** + Watcher which was added with a call to add a persistent watch. @@ -693,11 +716,11 @@ guarantees: ### Things to Remember about Watches -* Watches are one time triggers; if you get a watch event and +* Standard watches are one time triggers; if you get a watch event and you want to get notified of future changes, you must set another watch. -* Because watches are one time triggers and there is latency +* Because standard watches are one time triggers and there is latency between getting the event and sending a new request to get a watch you cannot reliably see every change that happens to a node in ZooKeeper. Be prepared to handle the case where the znode changes diff --git a/zookeeper-jute/src/main/resources/zookeeper.jute b/zookeeper-jute/src/main/resources/zookeeper.jute index 8310664a6e5..aab3c3f2d22 100644 --- a/zookeeper-jute/src/main/resources/zookeeper.jute +++ b/zookeeper-jute/src/main/resources/zookeeper.jute @@ -73,6 +73,14 @@ module org.apache.zookeeper.proto { vectorexistWatches; vectorchildWatches; } + class SetWatches2 { + long relativeZxid; + vectordataWatches; + vectorexistWatches; + vectorchildWatches; + vectorpersistentWatches; + vectorpersistentRecursiveWatches; + } class RequestHeader { int xid; int type; @@ -180,6 +188,10 @@ module org.apache.zookeeper.proto { class SetACLResponse { org.apache.zookeeper.data.Stat stat; } + class AddPersistentWatcherRequest { + ustring path; + boolean recursive; + } class WatcherEvent { int type; // event type int state; // state of the Keeper client runtime diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java index b0f7b0710e1..c61f6be1568 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -85,7 +85,7 @@ import org.apache.zookeeper.proto.RequestHeader; import org.apache.zookeeper.proto.SetACLResponse; import org.apache.zookeeper.proto.SetDataResponse; -import org.apache.zookeeper.proto.SetWatches; +import org.apache.zookeeper.proto.SetWatches2; import org.apache.zookeeper.proto.WatcherEvent; import org.apache.zookeeper.server.ByteBufferInputStream; import org.apache.zookeeper.server.ZooKeeperThread; @@ -990,16 +990,24 @@ void primeConnection() throws IOException { List dataWatches = zooKeeper.getDataWatches(); List existWatches = zooKeeper.getExistWatches(); List childWatches = zooKeeper.getChildWatches(); - if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty()) { + List persistentWatches = zooKeeper.getPersistentWatches(); + List persistentRecursiveWatches = zooKeeper.getPersistentRecursiveWatches(); + if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty() + || !persistentWatches.isEmpty() || !persistentRecursiveWatches.isEmpty()) { Iterator dataWatchesIter = prependChroot(dataWatches).iterator(); Iterator existWatchesIter = prependChroot(existWatches).iterator(); Iterator childWatchesIter = prependChroot(childWatches).iterator(); + Iterator persistentWatchesIter = prependChroot(persistentWatches).iterator(); + Iterator persistentRecursiveWatchesIter = prependChroot(persistentRecursiveWatches).iterator(); long setWatchesLastZxid = lastZxid; - while (dataWatchesIter.hasNext() || existWatchesIter.hasNext() || childWatchesIter.hasNext()) { + while (dataWatchesIter.hasNext() || existWatchesIter.hasNext() || childWatchesIter.hasNext() + || persistentWatchesIter.hasNext() || persistentRecursiveWatchesIter.hasNext()) { List dataWatchesBatch = new ArrayList(); List existWatchesBatch = new ArrayList(); List childWatchesBatch = new ArrayList(); + List persistentWatchesBatch = new ArrayList(); + List persistentRecursiveWatchesBatch = new ArrayList(); int batchLength = 0; // Note, we may exceed our max length by a bit when we add the last @@ -1015,14 +1023,21 @@ void primeConnection() throws IOException { } else if (childWatchesIter.hasNext()) { watch = childWatchesIter.next(); childWatchesBatch.add(watch); + } else if (persistentWatchesIter.hasNext()) { + watch = persistentWatchesIter.next(); + persistentWatchesBatch.add(watch); + } else if (persistentRecursiveWatchesIter.hasNext()) { + watch = persistentRecursiveWatchesIter.next(); + persistentRecursiveWatchesBatch.add(watch); } else { break; } batchLength += watch.length(); } - SetWatches sw = new SetWatches(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, childWatchesBatch); - RequestHeader header = new RequestHeader(-8, OpCode.setWatches); + SetWatches2 sw = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, + childWatchesBatch, persistentWatchesBatch, persistentRecursiveWatchesBatch); + RequestHeader header = new RequestHeader(-8, OpCode.setWatches2); Packet packet = new Packet(header, new ReplyHeader(), sw, null, null); outgoingQueue.addFirst(packet); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/Watcher.java b/zookeeper-server/src/main/java/org/apache/zookeeper/Watcher.java index db81fdf2668..ab4b654880e 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/Watcher.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/Watcher.java @@ -143,7 +143,8 @@ enum EventType { NodeDataChanged(3), NodeChildrenChanged(4), DataWatchRemoved(5), - ChildWatchRemoved(6); + ChildWatchRemoved(6), + PersistentWatchRemoved (7); private final int intValue; // Integer representation of value // for sending over wire @@ -172,6 +173,8 @@ public static EventType fromInt(int intValue) { return EventType.DataWatchRemoved; case 6: return EventType.ChildWatchRemoved; + case 7: + return EventType.PersistentWatchRemoved; default: throw new RuntimeException("Invalid integer value for conversion to EventType"); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java index db176f5b76e..5917a14ec5e 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java @@ -89,6 +89,10 @@ public interface OpCode { int getAllChildrenNumber = 104; + int setWatches2 = 105; + + int addPersistentWatch = 106; + int createSession = -10; int closeSession = -11; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java index 7f3b8483572..dd4e18c6026 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java @@ -53,12 +53,14 @@ import org.apache.zookeeper.common.PathUtils; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.proto.AddPersistentWatcherRequest; import org.apache.zookeeper.proto.CheckWatchesRequest; import org.apache.zookeeper.proto.Create2Response; import org.apache.zookeeper.proto.CreateRequest; import org.apache.zookeeper.proto.CreateResponse; import org.apache.zookeeper.proto.CreateTTLRequest; import org.apache.zookeeper.proto.DeleteRequest; +import org.apache.zookeeper.proto.ErrorResponse; import org.apache.zookeeper.proto.ExistsRequest; import org.apache.zookeeper.proto.GetACLRequest; import org.apache.zookeeper.proto.GetACLResponse; @@ -83,6 +85,7 @@ import org.apache.zookeeper.proto.SyncResponse; import org.apache.zookeeper.server.DataTree; import org.apache.zookeeper.server.EphemeralType; +import org.apache.zookeeper.server.watch.PathParentIterator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -254,6 +257,18 @@ protected List getChildWatches() { return rc; } } + protected List getPersistentWatches() { + synchronized (watchManager.persistentWatches) { + List rc = new ArrayList(watchManager.persistentWatches.keySet()); + return rc; + } + } + protected List getPersistentRecursiveWatches() { + synchronized (watchManager.persistentRecursiveWatches) { + List rc = new ArrayList(watchManager.persistentRecursiveWatches.keySet()); + return rc; + } + } /** * Manage watchers and handle events generated by the ClientCnxn object. @@ -267,6 +282,8 @@ static class ZKWatchManager implements ClientWatchManager { private final Map> dataWatches = new HashMap>(); private final Map> existWatches = new HashMap>(); private final Map> childWatches = new HashMap>(); + private final Map> persistentWatches = new HashMap>(); + private final Map> persistentRecursiveWatches = new HashMap>(); private boolean disableAutoWatchReset; ZKWatchManager(boolean disableAutoWatchReset) { @@ -296,6 +313,8 @@ public Map> removeWatcher( removedWatchers.put(EventType.ChildWatchRemoved, childWatchersToRem); HashSet dataWatchersToRem = new HashSet<>(); removedWatchers.put(EventType.DataWatchRemoved, dataWatchersToRem); + HashSet persistentWatchersToRem = new HashSet<>(); + removedWatchers.put(EventType.PersistentWatchRemoved, persistentWatchersToRem); boolean removedWatcher = false; switch (watcherType) { case Children: { @@ -324,10 +343,23 @@ public Map> removeWatcher( boolean removedDataWatcher = removeWatches(dataWatches, watcher, clientPath, local, rc, dataWatchersToRem); removedWatcher |= removedDataWatcher; } + synchronized (existWatches) { boolean removedDataWatcher = removeWatches(existWatches, watcher, clientPath, local, rc, dataWatchersToRem); removedWatcher |= removedDataWatcher; } + + synchronized (persistentWatches) { + boolean removedPersistentWatcher = removeWatches(persistentWatches, + watcher, clientPath, local, rc, persistentWatchersToRem); + removedWatcher |= removedPersistentWatcher; + } + + synchronized (persistentRecursiveWatches) { + boolean removedPersistentRecursiveWatcher = removeWatches(persistentRecursiveWatches, + watcher, clientPath, local, rc, persistentWatchersToRem); + removedWatcher |= removedPersistentRecursiveWatcher; + } } } // Watcher function doesn't exists for the specified params @@ -373,6 +405,18 @@ void containsWatcher(String path, Watcher watcher, WatcherType watcherType) thro synchronized (childWatches) { containsWatcher = contains(path, watcher, childWatches); } + + synchronized (persistentWatches) { + boolean contains_temp = contains(path, watcher, + persistentWatches); + containsWatcher |= contains_temp; + } + + synchronized (persistentRecursiveWatches) { + boolean contains_temp = contains(path, watcher, + persistentRecursiveWatches); + containsWatcher |= contains_temp; + } break; } case Data: { @@ -384,6 +428,18 @@ void containsWatcher(String path, Watcher watcher, WatcherType watcherType) thro boolean contains_temp = contains(path, watcher, existWatches); containsWatcher |= contains_temp; } + + synchronized (persistentWatches) { + boolean contains_temp = contains(path, watcher, + persistentWatches); + containsWatcher |= contains_temp; + } + + synchronized (persistentRecursiveWatches) { + boolean contains_temp = contains(path, watcher, + persistentRecursiveWatches); + containsWatcher |= contains_temp; + } break; } case Any: { @@ -395,10 +451,23 @@ void containsWatcher(String path, Watcher watcher, WatcherType watcherType) thro boolean contains_temp = contains(path, watcher, dataWatches); containsWatcher |= contains_temp; } + synchronized (existWatches) { boolean contains_temp = contains(path, watcher, existWatches); containsWatcher |= contains_temp; } + + synchronized (persistentWatches) { + boolean contains_temp = contains(path, watcher, + persistentWatches); + containsWatcher |= contains_temp; + } + + synchronized (persistentRecursiveWatches) { + boolean contains_temp = contains(path, watcher, + persistentRecursiveWatches); + containsWatcher |= contains_temp; + } } } // Watcher function doesn't exists for the specified params @@ -490,6 +559,18 @@ public Set materialize( } } + synchronized (persistentWatches) { + for (Set ws: persistentWatches.values()) { + result.addAll(ws); + } + } + + synchronized (persistentRecursiveWatches) { + for (Set ws: persistentRecursiveWatches.values()) { + result.addAll(ws); + } + } + return result; case NodeDataChanged: case NodeCreated: @@ -499,11 +580,13 @@ public Set materialize( synchronized (existWatches) { addTo(existWatches.remove(clientPath), result); } + addPersistentWatches(clientPath, result); break; case NodeChildrenChanged: synchronized (childWatches) { addTo(childWatches.remove(clientPath), result); } + addPersistentWatches(clientPath, result); break; case NodeDeleted: synchronized (dataWatches) { @@ -520,6 +603,7 @@ public Set materialize( synchronized (childWatches) { addTo(childWatches.remove(clientPath), result); } + addPersistentWatches(clientPath, result); break; default: String errorMsg = String.format( @@ -534,6 +618,16 @@ public Set materialize( return result; } + private void addPersistentWatches(String clientPath, Set result) { + synchronized (persistentWatches) { + addTo(persistentWatches.get(clientPath), result); + } + synchronized (persistentRecursiveWatches) { + for (String path : PathParentIterator.forAll(clientPath).asIterable()) { + addTo(persistentRecursiveWatches.get(path), result); + } + } + } } /** @@ -627,6 +721,25 @@ protected Map> getWatches(int rc) { } + class PersistentWatchRegistration extends WatchRegistration { + private final boolean recursive; + + public PersistentWatchRegistration(Watcher watcher, String clientPath, boolean recursive) { + super(watcher, clientPath); + this.recursive = recursive; + } + + @Override + protected Map> getWatches(int rc) { + return recursive ? watchManager.persistentRecursiveWatches : watchManager.persistentWatches; + } + + @Override + protected boolean shouldAddWatch(int rc) { + return rc == 0 || rc == KeeperException.Code.NONODE.intValue(); + } + } + @InterfaceAudience.Public public enum States { CONNECTING, @@ -3035,6 +3148,77 @@ public void removeAllWatches(String path, WatcherType watcherType, boolean local removeWatches(ZooDefs.OpCode.removeWatches, path, null, watcherType, local, cb, ctx); } + /** + *

+ * Set a watcher on the given path that: a) does not get removed when triggered (i.e. it stays active + * until it is removed); b) optionally applies not only to the registered path but all child paths recursively. This watcher + * is triggered for both data and child events. To remove the watcher, use + * removeWatches() with WatcherType.Any + *

+ * + *

+ * If recursive is false, the watcher behaves as if you placed an exists() watch and + * a getData() watch on the ZNode at the given path. + *

+ * + *

+ * If recursive is true, the watcher behaves as if you placed an exists() watch and + * a getData() watch on the ZNode at the given path and any ZNodes that are children + * of the given path including children added later. + *

+ * + *

+ * NOTE: when there are active recursive watches there is a small performance decrease as all segments + * of ZNode paths must be checked for watch triggering. + *

+ * + * @param basePath the top path that the watcher applies to + * @param watcher the watcher + * @param recursive if true applies not only to the registered path but all child paths recursively including + * any child nodes added in the future + * @throws InterruptedException If the server transaction is interrupted. + * @throws KeeperException If the server signals an error with a non-zero + * error code. + */ + public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) + throws KeeperException, InterruptedException { + PathUtils.validatePath(basePath); + String serverPath = prependChroot(basePath); + + RequestHeader h = new RequestHeader(); + h.setType(ZooDefs.OpCode.addPersistentWatch); + AddPersistentWatcherRequest request = new AddPersistentWatcherRequest(serverPath, recursive); + ReplyHeader r = cnxn.submitRequest(h, request, new ErrorResponse(), + new PersistentWatchRegistration(watcher, basePath, recursive)); + if (r.getErr() != 0) { + throw KeeperException.create(KeeperException.Code.get(r.getErr()), + basePath); + } + } + + /** + * Async version of {@link #addPersistentWatch(String, Watcher, boolean)} (see it for details) + * + * @param basePath the top path that the watcher applies to + * @param watcher the watcher + * @param recursive if true applies not only to the registered path but all child paths recursively including + * any child nodes added in the future + * @param cb a handler for the callback + * @param ctx context to be provided to the callback + * @throws IllegalArgumentException if an invalid path is specified + */ + public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive, + VoidCallback cb, Object ctx) { + PathUtils.validatePath(basePath); + String serverPath = prependChroot(basePath); + + RequestHeader h = new RequestHeader(); + h.setType(ZooDefs.OpCode.addPersistentWatch); + AddPersistentWatcherRequest request = new AddPersistentWatcherRequest(serverPath, recursive); + cnxn.queuePacket(h, new ReplyHeader(), request, new ErrorResponse(), cb, + basePath, serverPath, ctx, new PersistentWatchRegistration(watcher, basePath, recursive)); + } + private void validateWatcher(Watcher watcher) { if (watcher == null) { throw new IllegalArgumentException("Invalid Watcher, shouldn't be null!"); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java index a9b08b0de56..3c3835af361 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java @@ -58,6 +58,7 @@ import org.apache.zookeeper.data.StatPersisted; import org.apache.zookeeper.server.watch.IWatchManager; import org.apache.zookeeper.server.watch.WatchManagerFactory; +import org.apache.zookeeper.server.watch.WatcherMode; import org.apache.zookeeper.server.watch.WatcherOrBitSet; import org.apache.zookeeper.server.watch.WatchesPathReport; import org.apache.zookeeper.server.watch.WatchesReport; @@ -701,6 +702,12 @@ public String getMaxPrefixWithQuota(String path) { } } + public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) { + WatcherMode mode = recursive ? WatcherMode.PERSISTENT_RECURSIVE : WatcherMode.PERSISTENT; + dataWatches.addWatch(basePath, watcher, mode); + childWatches.addWatch(basePath, watcher, mode); + } + public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException { DataNode n = nodes.get(path); byte[] data = null; @@ -1499,7 +1506,8 @@ public void removeCnxn(Watcher watcher) { childWatches.removeWatcher(watcher); } - public void setWatches(long relativeZxid, List dataWatches, List existWatches, List childWatches, Watcher watcher) { + public void setWatches(long relativeZxid, List dataWatches, List existWatches, List childWatches, + List persistentWatches, List persistentRecursiveWatches, Watcher watcher) { for (String path : dataWatches) { DataNode node = getNode(path); WatchedEvent e = null; @@ -1529,6 +1537,14 @@ public void setWatches(long relativeZxid, List dataWatches, List this.childWatches.addWatch(path, watcher); } } + for (String path : persistentWatches) { + this.childWatches.addWatch(path, watcher, WatcherMode.PERSISTENT); + this.dataWatches.addWatch(path, watcher, WatcherMode.PERSISTENT); + } + for (String path : persistentRecursiveWatches) { + this.childWatches.addWatch(path, watcher, WatcherMode.PERSISTENT_RECURSIVE); + this.dataWatches.addWatch(path, watcher, WatcherMode.PERSISTENT_RECURSIVE); + } } /** diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index 93d70d86943..360f886e4d8 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Locale; import java.util.Set; @@ -47,9 +48,11 @@ import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Id; import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.proto.AddPersistentWatcherRequest; import org.apache.zookeeper.proto.CheckWatchesRequest; import org.apache.zookeeper.proto.Create2Response; import org.apache.zookeeper.proto.CreateResponse; +import org.apache.zookeeper.proto.ErrorResponse; import org.apache.zookeeper.proto.ExistsRequest; import org.apache.zookeeper.proto.ExistsResponse; import org.apache.zookeeper.proto.GetACLRequest; @@ -69,6 +72,7 @@ import org.apache.zookeeper.proto.SetACLResponse; import org.apache.zookeeper.proto.SetDataResponse; import org.apache.zookeeper.proto.SetWatches; +import org.apache.zookeeper.proto.SetWatches2; import org.apache.zookeeper.proto.SyncRequest; import org.apache.zookeeper.proto.SyncResponse; import org.apache.zookeeper.server.DataTree.ProcessTxnResult; @@ -365,7 +369,7 @@ public void processRequest(Request request) { case OpCode.setWatches: { lastOp = "SETW"; SetWatches setWatches = new SetWatches(); - // TODO We really should NOT need this!!!! + // TODO we really should not need this request.request.rewind(); ByteBufferInputStream.byteBuffer2Record(request.request, setWatches); long relativeZxid = setWatches.getRelativeZxid(); @@ -375,9 +379,37 @@ public void processRequest(Request request) { setWatches.getDataWatches(), setWatches.getExistWatches(), setWatches.getChildWatches(), + Collections.emptyList(), + Collections.emptyList(), cnxn); break; } + case OpCode.setWatches2: { + lastOp = "STW2"; + SetWatches2 setWatches = new SetWatches2(); + // TODO we really should not need this + request.request.rewind(); + ByteBufferInputStream.byteBuffer2Record(request.request, setWatches); + long relativeZxid = setWatches.getRelativeZxid(); + zks.getZKDatabase().setWatches(relativeZxid, + setWatches.getDataWatches(), + setWatches.getExistWatches(), + setWatches.getChildWatches(), + setWatches.getPersistentWatches(), + setWatches.getPersistentRecursiveWatches(), + cnxn); + break; + } + case OpCode.addPersistentWatch: { + lastOp = "APEW"; + AddPersistentWatcherRequest addPersistentWatcherRequest = new AddPersistentWatcherRequest(); + ByteBufferInputStream.byteBuffer2Record(request.request, + addPersistentWatcherRequest); + zks.getZKDatabase().addPersistentWatch(addPersistentWatcherRequest.getPath(), cnxn, + addPersistentWatcherRequest.getRecursive()); + rsp = new ErrorResponse(0); + break; + } case OpCode.getACL: { lastOp = "GETA"; GetACLRequest getACLRequest = new GetACLRequest(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java index c393684c4d8..d5357b5c79c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java @@ -797,10 +797,12 @@ protected void pRequest(Request request) throws RequestProcessorException { case OpCode.getChildren2: case OpCode.ping: case OpCode.setWatches: + case OpCode.setWatches2: case OpCode.checkWatches: case OpCode.removeWatches: case OpCode.getEphemerals: case OpCode.multiRead: + case OpCode.addPersistentWatch: zks.sessionTracker.checkSession(request.sessionId, request.getOwner()); break; default: diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java index bab21944628..355bfaa0461 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java @@ -243,9 +243,11 @@ static boolean isValid(int type) { case OpCode.setACL: case OpCode.setData: case OpCode.setWatches: + case OpCode.setWatches2: case OpCode.sync: case OpCode.checkWatches: case OpCode.removeWatches: + case OpCode.addPersistentWatch: return true; default: return false; @@ -334,6 +336,8 @@ public static String op2String(int op) { return "auth"; case OpCode.setWatches: return "setWatches"; + case OpCode.setWatches2: + return "setWatches2"; case OpCode.sasl: return "sasl"; case OpCode.getEphemerals: @@ -364,6 +368,7 @@ public String toString() { String path = "n/a"; if (type != OpCode.createSession && type != OpCode.setWatches + && type != OpCode.setWatches2 && type != OpCode.closeSession && request != null && request.remaining() >= 4) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java index a753b8a8da4..72705148b1f 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java @@ -516,10 +516,27 @@ public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperExce * @param dataWatches the data watches the client wants to reset * @param existWatches the exists watches the client wants to reset * @param childWatches the child watches the client wants to reset + * @param persistentWatches the persistent watches the client wants to reset + * @param persistentRecursiveWatches the persistent recursive watches the client wants to reset * @param watcher the watcher function */ - public void setWatches(long relativeZxid, List dataWatches, List existWatches, List childWatches, Watcher watcher) { - dataTree.setWatches(relativeZxid, dataWatches, existWatches, childWatches, watcher); + public void setWatches(long relativeZxid, List dataWatches, List existWatches, List childWatches, + List persistentWatches, List persistentRecursiveWatches, Watcher watcher) { + dataTree.setWatches(relativeZxid, dataWatches, existWatches, childWatches, persistentWatches, persistentRecursiveWatches, watcher); + } + + /** + * Add a persistent watch + * + * @param basePath + * watch base + * @param watcher + * the watcher + * @param recursive + * true if recursive + */ + public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) { + dataTree.addPersistentWatch(basePath, watcher, recursive); } /** diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/RequestPathMetricsCollector.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/RequestPathMetricsCollector.java index 9ef430cef9f..f3ec1fcea92 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/RequestPathMetricsCollector.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/RequestPathMetricsCollector.java @@ -32,6 +32,7 @@ import static org.apache.zookeeper.ZooDefs.OpCode.removeWatches; import static org.apache.zookeeper.ZooDefs.OpCode.setACL; import static org.apache.zookeeper.ZooDefs.OpCode.setData; +import static org.apache.zookeeper.ZooDefs.OpCode.setWatches2; import static org.apache.zookeeper.ZooDefs.OpCode.sync; import java.io.PrintWriter; import java.util.Arrays; @@ -131,6 +132,7 @@ public RequestPathMetricsCollector(boolean accurateMode) { requestsMap.put(Request.op2String(getChildren2), new PathStatsQueue(getChildren2)); requestsMap.put(Request.op2String(checkWatches), new PathStatsQueue(checkWatches)); requestsMap.put(Request.op2String(removeWatches), new PathStatsQueue(removeWatches)); + requestsMap.put(Request.op2String(setWatches2), new PathStatsQueue(setWatches2)); requestsMap.put(Request.op2String(sync), new PathStatsQueue(sync)); this.immutableRequestsMap = java.util.Collections.unmodifiableMap(requestsMap); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java index 286c7db1f0b..1bc44c805a0 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/IWatchManager.java @@ -34,6 +34,22 @@ public interface IWatchManager { */ boolean addWatch(String path, Watcher watcher); + /** + * Add watch to specific path. + * + * @param path znode path + * @param watcher watcher object reference + * @param watcherMode the watcher mode to use + * + * @return true if the watcher added is not already present + */ + default boolean addWatch(String path, Watcher watcher, WatcherMode watcherMode) { + if (watcherMode == WatcherMode.DEFAULT_WATCHER_MODE) { + return addWatch(path, watcher); + } + throw new UnsupportedOperationException(); // custom implementations must defeat this + } + /** * Checks the specified watcher exists for the given path. * @@ -129,4 +145,12 @@ public interface IWatchManager { */ void dumpWatches(PrintWriter pwriter, boolean byPath); + /** + * Return the current number of recursive watchers + * + * @return qty + */ + default int getRecursiveWatchQty() { + return 0; + } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/PathParentIterator.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/PathParentIterator.java new file mode 100644 index 00000000000..a6aa8cd2542 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/PathParentIterator.java @@ -0,0 +1,106 @@ +/** + * 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.Iterator; +import java.util.NoSuchElementException; + +/** + * Iterates over a ZooKeeper path. Each iteration goes up one parent path. Thus, the + * effect of the iterator is to iterate over the initial path and then all of its parents. + */ +public class PathParentIterator implements Iterator { + private String path; + private final int maxLevel; + private int level = -1; + + /** + * Return a new PathParentIterator that iterates from the + * given path to all parents. + * + * @param path initial path + */ + public static PathParentIterator forAll(String path) { + return new PathParentIterator(path, Integer.MAX_VALUE); + } + + /** + * Return a new PathParentIterator that only returns the given path - i.e. + * does not iterate to parent paths. + * + * @param path initial path + */ + public static PathParentIterator forPathOnly(String path) { + return new PathParentIterator(path, 0); + } + + private PathParentIterator(String path, int maxLevel) { + // NOTE: asserts that the path has already been validated + this.path = path; + this.maxLevel = maxLevel; + } + + /** + * Return an Iterable view so that this Iterator can be used in for each + * statements. IMPORTANT: the returned Iterable is single use only + * @return Iterable + */ + public Iterable asIterable() { + return () -> PathParentIterator.this; + } + + @Override + public boolean hasNext() { + return !path.isEmpty() && (level < maxLevel); + } + + /** + * Returns true if this iterator is currently at a parent path as opposed + * to the initial path given to the constructor + * + * @return true/false + */ + public boolean atParentPath() { + return level > 0; + } + + @Override + public String next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + + String localPath = path; + ++level; + if (path.equals("/")) { + path = ""; + } else { + path = path.substring(0, path.lastIndexOf('/')); + if (path.length() == 0) { + path = "/"; + } + } + return localPath; + } + + @Override + public void remove() { + throw new UnsupportedOperationException(); + } +} \ No newline at end of file diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java index 39327fcaa7a..c5b133059b2 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java @@ -21,6 +21,7 @@ import java.io.PrintWriter; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.Set; @@ -42,9 +43,11 @@ public class WatchManager implements IWatchManager { private static final Logger LOG = LoggerFactory.getLogger(WatchManager.class); - private final Map> watchTable = new HashMap>(); + private final Map> watchTable = new HashMap<>(); - private final Map> watch2Paths = new HashMap>(); + private final Map> watch2Paths = new HashMap<>(); + + private final WatcherModeManager watcherModeManager = new WatcherModeManager(); @Override public synchronized int size() { @@ -55,12 +58,17 @@ public synchronized int size() { return result; } - boolean isDeadWatcher(Watcher watcher) { + private boolean isDeadWatcher(Watcher watcher) { return watcher instanceof ServerCnxn && ((ServerCnxn) watcher).isStale(); } @Override - public synchronized boolean addWatch(String path, Watcher watcher) { + public boolean addWatch(String path, Watcher watcher) { + return addWatch(path, watcher, WatcherMode.DEFAULT_WATCHER_MODE); + } + + @Override + public synchronized boolean addWatch(String path, Watcher watcher, WatcherMode watcherMode) { if (isDeadWatcher(watcher)) { LOG.debug("Ignoring addWatch with closed cnxn"); return false; @@ -71,7 +79,7 @@ public synchronized boolean addWatch(String path, Watcher watcher) { // don't waste memory if there are few watches on a node // rehash when the 4th entry is added, doubling size thereafter // seems like a good compromise - list = new HashSet(4); + list = new HashSet<>(4); watchTable.put(path, list); } list.add(watcher); @@ -79,9 +87,12 @@ public synchronized boolean addWatch(String path, Watcher watcher) { Set paths = watch2Paths.get(watcher); if (paths == null) { // cnxns typically have many watches, so use default cap here - paths = new HashSet(); + paths = new HashSet<>(); watch2Paths.put(watcher, paths); } + + watcherModeManager.setWatcherMode(watcher, path, watcherMode); + return paths.add(path); } @@ -99,6 +110,7 @@ public synchronized void removeWatcher(Watcher watcher) { watchTable.remove(p); } } + watcherModeManager.removeWatcher(watcher, p); } } @@ -110,22 +122,45 @@ public WatcherOrBitSet triggerWatch(String path, EventType type) { @Override public WatcherOrBitSet triggerWatch(String path, EventType type, WatcherOrBitSet supress) { WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path); - Set watchers; + Set watchers = new HashSet<>(); + PathParentIterator pathParentIterator = getPathParentIterator(path); synchronized (this) { - watchers = watchTable.remove(path); - if (watchers == null || watchers.isEmpty()) { - if (LOG.isTraceEnabled()) { - ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK, "No watchers for " + path); + for (String localPath : pathParentIterator.asIterable()) { + Set thisWatchers = watchTable.get(localPath); + if (thisWatchers == null || thisWatchers.isEmpty()) { + continue; } - return null; - } - for (Watcher w : watchers) { - Set paths = watch2Paths.get(w); - if (paths != null) { - paths.remove(path); + Iterator iterator = thisWatchers.iterator(); + while (iterator.hasNext()) { + Watcher watcher = iterator.next(); + WatcherMode watcherMode = watcherModeManager.getWatcherMode(watcher, localPath); + if (watcherMode.isRecursive()) { + if (type != EventType.NodeChildrenChanged) { + watchers.add(watcher); + } + } else if (!pathParentIterator.atParentPath()) { + watchers.add(watcher); + if (!watcherMode.isPersistent()) { + iterator.remove(); + Set paths = watch2Paths.get(watcher); + if (paths != null) { + paths.remove(localPath); + } + } + } + } + if (thisWatchers.isEmpty()) { + watchTable.remove(localPath); } } } + if (watchers.isEmpty()) { + if (LOG.isTraceEnabled()) { + ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK, "No watchers for " + path); + } + return null; + } + for (Watcher w : watchers) { if (supress != null && supress.contains(w)) { continue; @@ -134,24 +169,24 @@ public WatcherOrBitSet triggerWatch(String path, EventType type, WatcherOrBitSet } switch (type) { - case NodeCreated: - ServerMetrics.getMetrics().NODE_CREATED_WATCHER.add(watchers.size()); - break; - - case NodeDeleted: - ServerMetrics.getMetrics().NODE_DELETED_WATCHER.add(watchers.size()); - break; - - case NodeDataChanged: - ServerMetrics.getMetrics().NODE_CHANGED_WATCHER.add(watchers.size()); - break; - - case NodeChildrenChanged: - ServerMetrics.getMetrics().NODE_CHILDREN_WATCHER.add(watchers.size()); - break; - default: - // Other types not logged. - break; + case NodeCreated: + ServerMetrics.getMetrics().NODE_CREATED_WATCHER.add(watchers.size()); + break; + + case NodeDeleted: + ServerMetrics.getMetrics().NODE_DELETED_WATCHER.add(watchers.size()); + break; + + case NodeDataChanged: + ServerMetrics.getMetrics().NODE_CHANGED_WATCHER.add(watchers.size()); + break; + + case NodeChildrenChanged: + ServerMetrics.getMetrics().NODE_CHILDREN_WATCHER.add(watchers.size()); + break; + default: + // Other types not logged. + break; } return new WatcherOrBitSet(watchers); @@ -197,8 +232,20 @@ public synchronized void dumpWatches(PrintWriter pwriter, boolean byPath) { @Override public synchronized boolean containsWatcher(String path, Watcher watcher) { - Set paths = watch2Paths.get(watcher); - return paths != null && paths.contains(path); + WatcherMode watcherMode = watcherModeManager.getWatcherMode(watcher, path); + PathParentIterator pathParentIterator = getPathParentIterator(path); + for (String localPath : pathParentIterator.asIterable()) { + Set watchers = watchTable.get(localPath); + if (!pathParentIterator.atParentPath()) { + if (watchers != null) { + return true; // at the leaf node, all watcher types match + } + } + if (watcherMode.isRecursive()) { + return true; + } + } + return false; } @Override @@ -217,15 +264,17 @@ public synchronized boolean removeWatcher(String path, Watcher watcher) { watchTable.remove(path); } + watcherModeManager.removeWatcher(watcher, path); + return true; } @Override public synchronized WatchesReport getWatches() { - Map> id2paths = new HashMap>(); + Map> id2paths = new HashMap<>(); for (Entry> e : watch2Paths.entrySet()) { Long id = ((ServerCnxn) e.getKey()).getSessionId(); - Set paths = new HashSet(e.getValue()); + Set paths = new HashSet<>(e.getValue()); id2paths.put(id, paths); } return new WatchesReport(id2paths); @@ -233,9 +282,9 @@ public synchronized WatchesReport getWatches() { @Override public synchronized WatchesPathReport getWatchesByPath() { - Map> path2ids = new HashMap>(); + Map> path2ids = new HashMap<>(); for (Entry> e : watchTable.entrySet()) { - Set ids = new HashSet(e.getValue().size()); + Set ids = new HashSet<>(e.getValue().size()); path2ids.put(e.getKey(), ids); for (Watcher watcher : e.getValue()) { ids.add(((ServerCnxn) watcher).getSessionId()); @@ -256,4 +305,15 @@ public synchronized WatchesSummary getWatchesSummary() { @Override public void shutdown() { /* do nothing */ } + @Override + public int getRecursiveWatchQty() { + return watcherModeManager.getRecursiveQty(); + } + + private PathParentIterator getPathParentIterator(String path) { + if (watcherModeManager.getRecursiveQty() == 0) { + return PathParentIterator.forPathOnly(path); + } + return PathParentIterator.forAll(path); + } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java new file mode 100644 index 00000000000..cfad6cbe72c --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java @@ -0,0 +1,44 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.server.watch; + +public enum WatcherMode { + STANDARD(false, false), + PERSISTENT(true, false), + PERSISTENT_RECURSIVE(true, true) + ; + + public static final WatcherMode DEFAULT_WATCHER_MODE = WatcherMode.STANDARD; + + private final boolean isPersistent; + private final boolean isRecursive; + + WatcherMode(boolean isPersistent, boolean isRecursive) { + this.isPersistent = isPersistent; + this.isRecursive = isRecursive; + } + + public boolean isPersistent() { + return isPersistent; + } + + public boolean isRecursive() { + return isRecursive; + } +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java new file mode 100644 index 00000000000..ca260f98737 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.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.server.watch; + +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.zookeeper.Watcher; + +class WatcherModeManager { + private final Map watcherModes = new ConcurrentHashMap<>(); + private final AtomicInteger recursiveQty = new AtomicInteger(0); + + private static class Key { + private final Watcher watcher; + private final String path; + + Key(Watcher watcher, String path) { + this.watcher = watcher; + this.path = path; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Key key = (Key) o; + return watcher.equals(key.watcher) && path.equals(key.path); + } + + @Override + public int hashCode() { + return Objects.hash(watcher, path); + } + } + + void setWatcherMode(Watcher watcher, String path, WatcherMode mode) { + if (mode == WatcherMode.DEFAULT_WATCHER_MODE) { + removeWatcher(watcher, path); + } else { + adjustRecursiveQty(watcherModes.put(new Key(watcher, path), mode), mode); + } + } + + WatcherMode getWatcherMode(Watcher watcher, String path) { + return watcherModes.getOrDefault(new Key(watcher, path), WatcherMode.DEFAULT_WATCHER_MODE); + } + + void removeWatcher(Watcher watcher, String path) { + adjustRecursiveQty(watcherModes.remove(new Key(watcher, path)), WatcherMode.DEFAULT_WATCHER_MODE); + } + + // visible for testing + int getRecursiveQty() { + return recursiveQty.get(); + } + + private void adjustRecursiveQty(WatcherMode oldMode, WatcherMode newMode) { + if (oldMode == null) { + oldMode = WatcherMode.DEFAULT_WATCHER_MODE; + } + if (oldMode.isRecursive() != newMode.isRecursive()) { + if (newMode.isRecursive()) { + recursiveQty.incrementAndGet(); + } else { + recursiveQty.decrementAndGet(); + } + } + } +} diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/PathParentIteratorTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/PathParentIteratorTest.java new file mode 100644 index 00000000000..59bb17adaa3 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/PathParentIteratorTest.java @@ -0,0 +1,84 @@ +/** + * 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.junit.Assert; +import org.junit.Test; + +public class PathParentIteratorTest { + @Test + public void testRoot() { + PathParentIterator pathParentIterator = PathParentIterator.forAll("/"); + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertFalse(pathParentIterator.atParentPath()); + Assert.assertEquals(pathParentIterator.next(), "/"); + Assert.assertFalse(pathParentIterator.hasNext()); + } + + @Test + public void test1Level() { + PathParentIterator pathParentIterator = PathParentIterator.forAll("/a"); + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertFalse(pathParentIterator.atParentPath()); + Assert.assertEquals(pathParentIterator.next(), "/a"); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/"); + Assert.assertTrue(pathParentIterator.atParentPath()); + + Assert.assertFalse(pathParentIterator.hasNext()); + } + + @Test + public void testLong() { + PathParentIterator pathParentIterator = PathParentIterator.forAll("/a/b/c/d"); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/a/b/c/d"); + Assert.assertFalse(pathParentIterator.atParentPath()); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/a/b/c"); + Assert.assertTrue(pathParentIterator.atParentPath()); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/a/b"); + Assert.assertTrue(pathParentIterator.atParentPath()); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/a"); + Assert.assertTrue(pathParentIterator.atParentPath()); + + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/"); + Assert.assertTrue(pathParentIterator.atParentPath()); + + Assert.assertFalse(pathParentIterator.hasNext()); + } + + @Test + public void testForPathOnly() { + PathParentIterator pathParentIterator = PathParentIterator.forPathOnly("/a/b/c/d"); + Assert.assertTrue(pathParentIterator.hasNext()); + Assert.assertEquals(pathParentIterator.next(), "/a/b/c/d"); + Assert.assertFalse(pathParentIterator.atParentPath()); + + Assert.assertFalse(pathParentIterator.hasNext()); + } +} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java index 0ce0a59a0da..e29dab90649 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java @@ -49,7 +49,7 @@ public class WatchManagerTest extends ZKTestCase { protected static final Logger LOG = LoggerFactory.getLogger(WatchManagerTest.class); - private static final String PATH_PREFIX = "path"; + private static final String PATH_PREFIX = "/path"; private ConcurrentHashMap watchers; private Random r; diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java new file mode 100644 index 00000000000..4ce8eac4899 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java @@ -0,0 +1,173 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.test; + +import java.io.IOException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class PersistentRecursiveWatcherTest extends ClientBase { + private static final Logger LOG = LoggerFactory.getLogger(PersistentRecursiveWatcherTest.class); + private BlockingQueue events; + private Watcher persistentWatcher; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + events = new LinkedBlockingQueue<>(); + persistentWatcher = event -> events.add(event); + } + + @Test + public void testBasic() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, true); + internalTestBasic(zk); + } + } + + @Test + public void testBasicAsync() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + final CountDownLatch latch = new CountDownLatch(1); + AsyncCallback.VoidCallback cb = (rc, path, ctx) -> { + if (rc == 0) { + latch.countDown(); + } + }; + zk.addPersistentWatch("/a/b", persistentWatcher, true, cb, null); + Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); + internalTestBasic(zk); + } + } + + private void internalTestBasic(ZooKeeper zk) throws KeeperException, InterruptedException { + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c/d", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c/d/e", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.setData("/a/b/c/d/e", new byte[0], -1); + zk.delete("/a/b/c/d/e", -1); + zk.create("/a/b/c/d/e", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b/c"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b/c/d"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b/c/d/e"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b/c/d/e"); + assertEvent(events, Watcher.Event.EventType.NodeDeleted, "/a/b/c/d/e"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b/c/d/e"); + } + + @Test + public void testRemoval() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, true); + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b/c"); + + zk.removeWatches("/a/b", persistentWatcher, Watcher.WatcherType.Any, false); + zk.create("/a/b/c/d", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + assertEvent(events, Watcher.Event.EventType.PersistentWatchRemoved, "/a/b"); + } + } + + @Test + public void testDisconnect() throws Exception { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, true); + stopServer(); + assertEvent(events, Watcher.Event.EventType.None, null); + startServer(); + assertEvent(events, Watcher.Event.EventType.None, null); + internalTestBasic(zk); + } + } + + @Test + public void testMultiClient() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk1 = createClient(new CountdownWatcher(), hostPort); ZooKeeper zk2 = createClient(new CountdownWatcher(), hostPort)) { + + zk1.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk1.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk1.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + + zk1.addPersistentWatch("/a/b", persistentWatcher, true); + zk1.setData("/a/b/c", "one".getBytes(), -1); + Thread.sleep(1000); // give some time for the event to arrive + + zk2.setData("/a/b/c", "two".getBytes(), -1); + zk2.setData("/a/b/c", "three".getBytes(), -1); + zk2.setData("/a/b/c", "four".getBytes(), -1); + + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b/c"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b/c"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b/c"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b/c"); + } + } + + @Test + public void testRootWatcher() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/", persistentWatcher, true); + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/b"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/b/c"); + } + } + + private void assertEvent(BlockingQueue events, Watcher.Event.EventType eventType, String path) + throws InterruptedException { + WatchedEvent event = events.poll(5, TimeUnit.SECONDS); + Assert.assertNotNull(event); + Assert.assertEquals(eventType, event.getType()); + Assert.assertEquals(path, event.getPath()); + } +} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java new file mode 100644 index 00000000000..49c926932a6 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java @@ -0,0 +1,169 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.test; + +import java.io.IOException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class PersistentWatcherTest extends ClientBase { + private static final Logger LOG = LoggerFactory.getLogger(PersistentWatcherTest.class); + private BlockingQueue events; + private Watcher persistentWatcher; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + events = new LinkedBlockingQueue<>(); + persistentWatcher = event -> events.add(event); + } + + @Test + public void testBasic() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, false); + internalTestBasic(zk); + } + } + + @Test + public void testBasicAsync() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + final CountDownLatch latch = new CountDownLatch(1); + AsyncCallback.VoidCallback cb = (rc, path, ctx) -> { + if (rc == 0) { + latch.countDown(); + } + }; + zk.addPersistentWatch("/a/b", persistentWatcher, false, cb, null); + Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); + internalTestBasic(zk); + } + } + + private void internalTestBasic(ZooKeeper zk) throws KeeperException, InterruptedException { + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.setData("/a/b", new byte[0], -1); + zk.delete("/a/b/c", -1); + zk.delete("/a/b", -1); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeChildrenChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeChildrenChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeDeleted, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + } + + @Test + public void testRemoval() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, false); + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + assertEvent(events, Watcher.Event.EventType.NodeCreated, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeChildrenChanged, "/a/b"); + + zk.removeWatches("/a/b", persistentWatcher, Watcher.WatcherType.Any, false); + zk.delete("/a/b/c", -1); + zk.delete("/a/b", -1); + assertEvent(events, Watcher.Event.EventType.PersistentWatchRemoved, "/a/b"); + } + } + + @Test + public void testDisconnect() throws Exception { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/a/b", persistentWatcher, false); + stopServer(); + assertEvent(events, Watcher.Event.EventType.None, null); + startServer(); + assertEvent(events, Watcher.Event.EventType.None, null); + internalTestBasic(zk); + } + } + + @Test + public void testMultiClient() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk1 = createClient(new CountdownWatcher(), hostPort); + ZooKeeper zk2 = createClient(new CountdownWatcher(), hostPort)) { + + zk1.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk1.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + + zk1.addPersistentWatch("/a/b", persistentWatcher, false); + zk1.setData("/a/b", "one".getBytes(), -1); + Thread.sleep(1000); // give some time for the event to arrive + + zk2.setData("/a/b", "two".getBytes(), -1); + zk2.setData("/a/b", "three".getBytes(), -1); + zk2.setData("/a/b", "four".getBytes(), -1); + + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b"); + assertEvent(events, Watcher.Event.EventType.NodeDataChanged, "/a/b"); + } + } + + @Test + public void testRootWatcher() + throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { + zk.addPersistentWatch("/", persistentWatcher, false); + zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.setData("/a", new byte[0], -1); + zk.create("/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + assertEvent(events, Watcher.Event.EventType.NodeChildrenChanged, "/"); + assertEvent(events, Watcher.Event.EventType.NodeChildrenChanged, "/"); + } + } + + private void assertEvent(BlockingQueue events, Watcher.Event.EventType eventType, String path) + throws InterruptedException { + WatchedEvent event = events.poll(5, TimeUnit.SECONDS); + Assert.assertNotNull(event); + Assert.assertEquals(eventType, event.getType()); + Assert.assertEquals(path, event.getPath()); + } +} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java new file mode 100644 index 00000000000..2959a4a1d41 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java @@ -0,0 +1,108 @@ +/** + * 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.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.server.watch.WatchManager; +import org.apache.zookeeper.server.watch.WatcherMode; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class RecursiveWatchQtyTest { + private WatchManager watchManager; + + private static class DummyWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + // NOP + } + } + + @Before + public void setup() { + watchManager = new WatchManager(); + } + + @Test + public void testAddRemove() { + Watcher watcher1 = new DummyWatcher(); + Watcher watcher2 = new DummyWatcher(); + + watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); + Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); + Assert.assertTrue(watchManager.removeWatcher("/a", watcher1)); + Assert.assertTrue(watchManager.removeWatcher("/b", watcher2)); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testAddRemoveAlt() { + Watcher watcher1 = new DummyWatcher(); + Watcher watcher2 = new DummyWatcher(); + + watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); + Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher1); + watchManager.removeWatcher(watcher2); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testDoubleAdd() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + Assert.assertEquals(1, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testSameWatcherMultiPath() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a/b", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a/b/c", watcher, WatcherMode.PERSISTENT_RECURSIVE); + Assert.assertEquals(3, watchManager.getRecursiveWatchQty()); + Assert.assertTrue(watchManager.removeWatcher("/a/b", watcher)); + Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testChangeType() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + Assert.assertEquals(1, watchManager.getRecursiveWatchQty()); + watchManager.addWatch("/a", watcher, WatcherMode.STANDARD); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + Assert.assertTrue(watchManager.removeWatcher("/a", watcher)); + Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); + } +} \ No newline at end of file From ee93f7569e1b2d81d750a1e459f13abba55b7476 Mon Sep 17 00:00:00 2001 From: randgalt Date: Thu, 3 Oct 2019 08:16:54 -0500 Subject: [PATCH 2/5] ZOOKEEPER-1416 maintain compatibility with older servers - if no persistent/recursive watchers are used, use the old version of SetWatches --- .../java/org/apache/zookeeper/ClientCnxn.java | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java index c61f6be1568..3713646d631 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -85,6 +85,7 @@ import org.apache.zookeeper.proto.RequestHeader; import org.apache.zookeeper.proto.SetACLResponse; import org.apache.zookeeper.proto.SetDataResponse; +import org.apache.zookeeper.proto.SetWatches; import org.apache.zookeeper.proto.SetWatches2; import org.apache.zookeeper.proto.WatcherEvent; import org.apache.zookeeper.server.ByteBufferInputStream; @@ -1035,10 +1036,20 @@ void primeConnection() throws IOException { batchLength += watch.length(); } - SetWatches2 sw = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, - childWatchesBatch, persistentWatchesBatch, persistentRecursiveWatchesBatch); - RequestHeader header = new RequestHeader(-8, OpCode.setWatches2); - Packet packet = new Packet(header, new ReplyHeader(), sw, null, null); + Record record; + int opcode; + if (persistentWatchesBatch.isEmpty() && persistentRecursiveWatchesBatch.isEmpty()) { + // maintain compatibility with older servers - if no persistent/recursive watchers + // are used, use the old version of SetWatches + record = new SetWatches(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, childWatchesBatch); + opcode = OpCode.setWatches; + } else { + record = new SetWatches2(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, + childWatchesBatch, persistentWatchesBatch, persistentRecursiveWatchesBatch); + opcode = OpCode.setWatches2; + } + RequestHeader header = new RequestHeader(-8, opcode); + Packet packet = new Packet(header, new ReplyHeader(), record, null, null); outgoingQueue.addFirst(packet); } } From 52d2caa5ee27e4ab1171c2bcda4cd92e99d6e85f Mon Sep 17 00:00:00 2001 From: randgalt Date: Thu, 3 Oct 2019 09:26:29 -0500 Subject: [PATCH 3/5] ZOOKEEPER-1416 Generalized this feature in the hopes of making it easier to change in the future. The APIs are no longer specific to persistent/recursive. Now, they are generically "add watch" and there is an additional mode that details the behavior of the watch. For now, the only choices are persistent or recursive but that can change in the future. --- .../markdown/zookeeperProgrammers.md | 2 +- .../src/main/resources/zookeeper.jute | 4 +- .../org/apache/zookeeper/AddWatchMode.java | 67 ++++++ .../java/org/apache/zookeeper/ZooDefs.java | 9 +- .../java/org/apache/zookeeper/ZooKeeper.java | 73 +++---- .../org/apache/zookeeper/server/DataTree.java | 8 +- .../server/FinalRequestProcessor.java | 13 +- .../server/PrepRequestProcessor.java | 2 +- .../org/apache/zookeeper/server/Request.java | 2 +- .../apache/zookeeper/server/ZKDatabase.java | 10 +- .../zookeeper/server/watch/WatcherMode.java | 12 ++ .../server/watch/WatcherModeManager.java | 7 +- .../server/watch/RecursiveWatchQtyTest.java | 197 ++++++++++++++++++ .../test/PersistentRecursiveWatcherTest.java | 13 +- .../zookeeper/test/PersistentWatcherTest.java | 13 +- .../zookeeper/test/RecursiveWatchQtyTest.java | 108 ---------- .../test/UnsupportedAddWatcherTest.java | 124 +++++++++++ 17 files changed, 477 insertions(+), 187 deletions(-) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/AddWatchMode.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/RecursiveWatchQtyTest.java delete mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/UnsupportedAddWatcherTest.java diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md b/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md index 8d45ca6c650..d31bbca00ab 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperProgrammers.md @@ -673,7 +673,7 @@ Additionally, these watches trigger the event types *NodeCreated*, *NodeDeleted* and, optionally, recursively for all znodes starting at the znode that the watch is registered for. Note that *NodeChildrenChanged* events are not triggered for persistent recursive watches as it would be redundant. -Persistent watches are set using the method *addPersistentWatch()*. The triggering semantics and guarantees +Persistent watches are set using the method *addWatch()*. The triggering semantics and guarantees (other than one-time triggering) are the same as standard watches. The only exception regarding events is that recursive persistent watchers never trigger child changed events as they are redundant. Persistent watches are removed using *removeWatches()* with watcher type *WatcherType.Any*. diff --git a/zookeeper-jute/src/main/resources/zookeeper.jute b/zookeeper-jute/src/main/resources/zookeeper.jute index aab3c3f2d22..6d553657540 100644 --- a/zookeeper-jute/src/main/resources/zookeeper.jute +++ b/zookeeper-jute/src/main/resources/zookeeper.jute @@ -188,9 +188,9 @@ module org.apache.zookeeper.proto { class SetACLResponse { org.apache.zookeeper.data.Stat stat; } - class AddPersistentWatcherRequest { + class AddWatchRequest { ustring path; - boolean recursive; + int mode; } class WatcherEvent { int type; // event type diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/AddWatchMode.java b/zookeeper-server/src/main/java/org/apache/zookeeper/AddWatchMode.java new file mode 100644 index 00000000000..0f339c1197a --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/AddWatchMode.java @@ -0,0 +1,67 @@ +/* + * 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; + +/** + * Modes available to {@link ZooKeeper#addWatch(String, Watcher, AddWatchMode)} + */ +public enum AddWatchMode { + /** + *

+ * Set a watcher on the given path that does not get removed when triggered (i.e. it stays active + * until it is removed). This watcher + * is triggered for both data and child events. To remove the watcher, use + * removeWatches() with WatcherType.Any. The watcher behaves as if you placed an exists() watch and + * a getData() watch on the ZNode at the given path. + *

+ */ + PERSISTENT(ZooDefs.AddWatchModes.persistent), + + /** + *

+ * Set a watcher on the given path that: a) does not get removed when triggered (i.e. it stays active + * until it is removed); b) applies not only to the registered path but all child paths recursively. This watcher + * is triggered for both data and child events. To remove the watcher, use + * removeWatches() with WatcherType.Any + *

+ * + *

+ * The watcher behaves as if you placed an exists() watch and + * a getData() watch on the ZNode at the given path and any ZNodes that are children + * of the given path including children added later. + *

+ * + *

+ * NOTE: when there are active recursive watches there is a small performance decrease as all segments + * of ZNode paths must be checked for watch triggering. + *

+ */ + PERSISTENT_RECURSIVE(ZooDefs.AddWatchModes.persistentRecursive) + ; + + public int getMode() { + return mode; + } + + private final int mode; + + AddWatchMode(int mode) { + this.mode = mode; + } +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java index 5917a14ec5e..a12e5803c27 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java @@ -91,7 +91,7 @@ public interface OpCode { int setWatches2 = 105; - int addPersistentWatch = 106; + int addWatch = 106; int createSession = -10; @@ -152,6 +152,13 @@ public interface Ids { } + @InterfaceAudience.Public + public interface AddWatchModes { + int persistent = 0; // matches AddWatchMode.PERSISTENT + + int persistentRecursive = 1; // matches AddWatchMode.PERSISTENT_RECURSIVE + } + public static final String[] opNames = {"notification", "create", "delete", "exists", "getData", "setData", "getACL", "setACL", "getChildren", "getChildren2", "getMaxChildren", "setMaxChildren", "ping", "reconfig", "getConfig"}; } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java index dd4e18c6026..f2e52c9503f 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java @@ -53,7 +53,7 @@ import org.apache.zookeeper.common.PathUtils; import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Stat; -import org.apache.zookeeper.proto.AddPersistentWatcherRequest; +import org.apache.zookeeper.proto.AddWatchRequest; import org.apache.zookeeper.proto.CheckWatchesRequest; import org.apache.zookeeper.proto.Create2Response; import org.apache.zookeeper.proto.CreateRequest; @@ -721,17 +721,23 @@ protected Map> getWatches(int rc) { } - class PersistentWatchRegistration extends WatchRegistration { - private final boolean recursive; + class AddWatchRegistration extends WatchRegistration { + private final AddWatchMode mode; - public PersistentWatchRegistration(Watcher watcher, String clientPath, boolean recursive) { + public AddWatchRegistration(Watcher watcher, String clientPath, AddWatchMode mode) { super(watcher, clientPath); - this.recursive = recursive; + this.mode = mode; } @Override protected Map> getWatches(int rc) { - return recursive ? watchManager.persistentRecursiveWatches : watchManager.persistentWatches; + switch (mode) { + case PERSISTENT: + return watchManager.persistentWatches; + case PERSISTENT_RECURSIVE: + return watchManager.persistentRecursiveWatches; + } + throw new IllegalArgumentException("Mode not supported: " + mode); } @Override @@ -3149,47 +3155,27 @@ public void removeAllWatches(String path, WatcherType watcherType, boolean local } /** - *

- * Set a watcher on the given path that: a) does not get removed when triggered (i.e. it stays active - * until it is removed); b) optionally applies not only to the registered path but all child paths recursively. This watcher - * is triggered for both data and child events. To remove the watcher, use - * removeWatches() with WatcherType.Any - *

- * - *

- * If recursive is false, the watcher behaves as if you placed an exists() watch and - * a getData() watch on the ZNode at the given path. - *

- * - *

- * If recursive is true, the watcher behaves as if you placed an exists() watch and - * a getData() watch on the ZNode at the given path and any ZNodes that are children - * of the given path including children added later. - *

- * - *

- * NOTE: when there are active recursive watches there is a small performance decrease as all segments - * of ZNode paths must be checked for watch triggering. - *

+ * Add a watch to the given znode using the given mode. Note: not all + * watch types can be set with this method. Only the modes available + * in {@link AddWatchMode} can be set with this method. * - * @param basePath the top path that the watcher applies to + * @param basePath the path that the watcher applies to * @param watcher the watcher - * @param recursive if true applies not only to the registered path but all child paths recursively including - * any child nodes added in the future + * @param mode type of watcher to add * @throws InterruptedException If the server transaction is interrupted. * @throws KeeperException If the server signals an error with a non-zero * error code. */ - public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) + public void addWatch(String basePath, Watcher watcher, AddWatchMode mode) throws KeeperException, InterruptedException { PathUtils.validatePath(basePath); String serverPath = prependChroot(basePath); RequestHeader h = new RequestHeader(); - h.setType(ZooDefs.OpCode.addPersistentWatch); - AddPersistentWatcherRequest request = new AddPersistentWatcherRequest(serverPath, recursive); + h.setType(ZooDefs.OpCode.addWatch); + AddWatchRequest request = new AddWatchRequest(serverPath, mode.getMode()); ReplyHeader r = cnxn.submitRequest(h, request, new ErrorResponse(), - new PersistentWatchRegistration(watcher, basePath, recursive)); + new AddWatchRegistration(watcher, basePath, mode)); if (r.getErr() != 0) { throw KeeperException.create(KeeperException.Code.get(r.getErr()), basePath); @@ -3197,26 +3183,25 @@ public void addPersistentWatch(String basePath, Watcher watcher, boolean recursi } /** - * Async version of {@link #addPersistentWatch(String, Watcher, boolean)} (see it for details) + * Async version of {@link #addWatch(String, Watcher, AddWatchMode)} (see it for details) * - * @param basePath the top path that the watcher applies to + * @param basePath the path that the watcher applies to * @param watcher the watcher - * @param recursive if true applies not only to the registered path but all child paths recursively including - * any child nodes added in the future + * @param mode type of watcher to add * @param cb a handler for the callback * @param ctx context to be provided to the callback * @throws IllegalArgumentException if an invalid path is specified */ - public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive, - VoidCallback cb, Object ctx) { + public void addWatch(String basePath, Watcher watcher, AddWatchMode mode, + VoidCallback cb, Object ctx) { PathUtils.validatePath(basePath); String serverPath = prependChroot(basePath); RequestHeader h = new RequestHeader(); - h.setType(ZooDefs.OpCode.addPersistentWatch); - AddPersistentWatcherRequest request = new AddPersistentWatcherRequest(serverPath, recursive); + h.setType(ZooDefs.OpCode.addWatch); + AddWatchRequest request = new AddWatchRequest(serverPath, mode.getMode()); cnxn.queuePacket(h, new ReplyHeader(), request, new ErrorResponse(), cb, - basePath, serverPath, ctx, new PersistentWatchRegistration(watcher, basePath, recursive)); + basePath, serverPath, ctx, new AddWatchRegistration(watcher, basePath, mode)); } private void validateWatcher(Watcher watcher) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java index 3c3835af361..766949ece5a 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java @@ -702,10 +702,10 @@ public String getMaxPrefixWithQuota(String path) { } } - public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) { - WatcherMode mode = recursive ? WatcherMode.PERSISTENT_RECURSIVE : WatcherMode.PERSISTENT; - dataWatches.addWatch(basePath, watcher, mode); - childWatches.addWatch(basePath, watcher, mode); + public void addWatch(String basePath, Watcher watcher, int mode) { + WatcherMode watcherMode = WatcherMode.fromZooDef(mode); + dataWatches.addWatch(basePath, watcher, watcherMode); + childWatches.addWatch(basePath, watcher, watcherMode); } public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index 360f886e4d8..79a69771805 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -48,7 +48,7 @@ import org.apache.zookeeper.data.ACL; import org.apache.zookeeper.data.Id; import org.apache.zookeeper.data.Stat; -import org.apache.zookeeper.proto.AddPersistentWatcherRequest; +import org.apache.zookeeper.proto.AddWatchRequest; import org.apache.zookeeper.proto.CheckWatchesRequest; import org.apache.zookeeper.proto.Create2Response; import org.apache.zookeeper.proto.CreateResponse; @@ -400,13 +400,12 @@ public void processRequest(Request request) { cnxn); break; } - case OpCode.addPersistentWatch: { - lastOp = "APEW"; - AddPersistentWatcherRequest addPersistentWatcherRequest = new AddPersistentWatcherRequest(); + case OpCode.addWatch: { + lastOp = "ADDW"; + AddWatchRequest addWatcherRequest = new AddWatchRequest(); ByteBufferInputStream.byteBuffer2Record(request.request, - addPersistentWatcherRequest); - zks.getZKDatabase().addPersistentWatch(addPersistentWatcherRequest.getPath(), cnxn, - addPersistentWatcherRequest.getRecursive()); + addWatcherRequest); + zks.getZKDatabase().addWatch(addWatcherRequest.getPath(), cnxn, addWatcherRequest.getMode()); rsp = new ErrorResponse(0); break; } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java index d5357b5c79c..70d989a34ac 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java @@ -802,7 +802,7 @@ protected void pRequest(Request request) throws RequestProcessorException { case OpCode.removeWatches: case OpCode.getEphemerals: case OpCode.multiRead: - case OpCode.addPersistentWatch: + case OpCode.addWatch: zks.sessionTracker.checkSession(request.sessionId, request.getOwner()); break; default: diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java index 355bfaa0461..122f0cab097 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java @@ -247,7 +247,7 @@ static boolean isValid(int type) { case OpCode.sync: case OpCode.checkWatches: case OpCode.removeWatches: - case OpCode.addPersistentWatch: + case OpCode.addWatch: return true; default: return false; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java index 72705148b1f..c9c6d54cdb8 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java @@ -526,17 +526,17 @@ public void setWatches(long relativeZxid, List dataWatches, List } /** - * Add a persistent watch + * Add a watch * * @param basePath * watch base * @param watcher * the watcher - * @param recursive - * true if recursive + * @param mode + * a mode from ZooDefs.AddWatchModes */ - public void addPersistentWatch(String basePath, Watcher watcher, boolean recursive) { - dataTree.addPersistentWatch(basePath, watcher, recursive); + public void addWatch(String basePath, Watcher watcher, int mode) { + dataTree.addWatch(basePath, watcher, mode); } /** diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java index cfad6cbe72c..b8a1dda7408 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherMode.java @@ -18,6 +18,8 @@ package org.apache.zookeeper.server.watch; +import org.apache.zookeeper.ZooDefs; + public enum WatcherMode { STANDARD(false, false), PERSISTENT(true, false), @@ -26,6 +28,16 @@ public enum WatcherMode { public static final WatcherMode DEFAULT_WATCHER_MODE = WatcherMode.STANDARD; + public static WatcherMode fromZooDef(int mode) { + switch (mode) { + case ZooDefs.AddWatchModes.persistent: + return PERSISTENT; + case ZooDefs.AddWatchModes.persistentRecursive: + return PERSISTENT_RECURSIVE; + } + throw new IllegalArgumentException("Unsupported mode: " + mode); + } + private final boolean isPersistent; private final boolean isRecursive; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java index ca260f98737..c1a8225f8ae 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherModeManager.java @@ -55,6 +55,11 @@ public int hashCode() { } } + // VisibleForTesting + Map getWatcherModes() { + return watcherModes; + } + void setWatcherMode(Watcher watcher, String path, WatcherMode mode) { if (mode == WatcherMode.DEFAULT_WATCHER_MODE) { removeWatcher(watcher, path); @@ -71,11 +76,11 @@ void removeWatcher(Watcher watcher, String path) { adjustRecursiveQty(watcherModes.remove(new Key(watcher, path)), WatcherMode.DEFAULT_WATCHER_MODE); } - // visible for testing int getRecursiveQty() { return recursiveQty.get(); } + // recursiveQty is an optimization to avoid having to walk the map every time this value is needed private void adjustRecursiveQty(WatcherMode oldMode, WatcherMode newMode) { if (oldMode == null) { oldMode = WatcherMode.DEFAULT_WATCHER_MODE; diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/RecursiveWatchQtyTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/RecursiveWatchQtyTest.java new file mode 100644 index 00000000000..067cb2af94a --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/RecursiveWatchQtyTest.java @@ -0,0 +1,197 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadLocalRandom; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.junit.Before; +import org.junit.Test; + +public class RecursiveWatchQtyTest { + private WatchManager watchManager; + + private static final int clientQty = 25; + private static final int iterations = 1000; + + private static class DummyWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + // NOP + } + } + + @Before + public void setup() { + watchManager = new WatchManager(); + } + + @Test + public void testRecursiveQty() { + WatcherModeManager manager = new WatcherModeManager(); + DummyWatcher watcher = new DummyWatcher(); + manager.setWatcherMode(watcher, "/a", WatcherMode.DEFAULT_WATCHER_MODE); + assertEquals(0, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a", WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(1, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a/b", WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(2, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a", WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(2, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a/b", WatcherMode.PERSISTENT); + assertEquals(1, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a/b", WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(2, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a/b", WatcherMode.DEFAULT_WATCHER_MODE); + assertEquals(1, manager.getRecursiveQty()); + manager.setWatcherMode(watcher, "/a", WatcherMode.PERSISTENT); + assertEquals(0, manager.getRecursiveQty()); + } + + @Test + public void testAddRemove() { + Watcher watcher1 = new DummyWatcher(); + Watcher watcher2 = new DummyWatcher(); + + watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(2, watchManager.getRecursiveWatchQty()); + assertTrue(watchManager.removeWatcher("/a", watcher1)); + assertTrue(watchManager.removeWatcher("/b", watcher2)); + assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testAddRemoveAlt() { + Watcher watcher1 = new DummyWatcher(); + Watcher watcher2 = new DummyWatcher(); + + watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(2, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher1); + watchManager.removeWatcher(watcher2); + assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testDoubleAdd() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(1, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher); + assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testSameWatcherMultiPath() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a/b", watcher, WatcherMode.PERSISTENT_RECURSIVE); + watchManager.addWatch("/a/b/c", watcher, WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(3, watchManager.getRecursiveWatchQty()); + assertTrue(watchManager.removeWatcher("/a/b", watcher)); + assertEquals(2, watchManager.getRecursiveWatchQty()); + watchManager.removeWatcher(watcher); + assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testChangeType() { + Watcher watcher = new DummyWatcher(); + + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT); + assertEquals(0, watchManager.getRecursiveWatchQty()); + watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); + assertEquals(1, watchManager.getRecursiveWatchQty()); + watchManager.addWatch("/a", watcher, WatcherMode.STANDARD); + assertEquals(0, watchManager.getRecursiveWatchQty()); + assertTrue(watchManager.removeWatcher("/a", watcher)); + assertEquals(0, watchManager.getRecursiveWatchQty()); + } + + @Test + public void testRecursiveQtyConcurrency() { + ThreadLocalRandom random = ThreadLocalRandom.current(); + WatcherModeManager manager = new WatcherModeManager(); + ExecutorService threadPool = Executors.newFixedThreadPool(clientQty); + List> tasks = null; + CountDownLatch completedLatch = new CountDownLatch(clientQty); + try { + tasks = IntStream.range(0, clientQty) + .mapToObj(__ -> threadPool.submit(() -> iterate(manager, completedLatch))) + .collect(Collectors.toList()); + try { + completedLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } finally { + if (tasks != null) { + tasks.forEach(t -> t.cancel(true)); + } + threadPool.shutdownNow(); + } + + int expectedRecursiveQty = (int) manager.getWatcherModes().values() + .stream() + .filter(mode -> mode == WatcherMode.PERSISTENT_RECURSIVE) + .count(); + assertEquals(expectedRecursiveQty, manager.getRecursiveQty()); + } + + private void iterate(WatcherModeManager manager, CountDownLatch completedLatch) { + ThreadLocalRandom random = ThreadLocalRandom.current(); + try { + for (int i = 0; i < iterations; ++i) { + String path = "/" + random.nextInt(clientQty); + boolean doSet = random.nextInt(100) > 33; // 2/3 will be sets + if (doSet) { + WatcherMode mode = WatcherMode.values()[random.nextInt(WatcherMode.values().length)]; + manager.setWatcherMode(new DummyWatcher(), path, mode); + } else { + manager.removeWatcher(new DummyWatcher(), path); + } + + int sleepMillis = random.nextInt(2); + if (sleepMillis > 0) { + try { + Thread.sleep(sleepMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + } finally { + completedLatch.countDown(); + } + } +} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java index 4ce8eac4899..67f19dc0549 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentRecursiveWatcherTest.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.test; +import static org.apache.zookeeper.AddWatchMode.PERSISTENT_RECURSIVE; import java.io.IOException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -54,7 +55,7 @@ public void setUp() throws Exception { public void testBasic() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, true); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT_RECURSIVE); internalTestBasic(zk); } } @@ -69,7 +70,7 @@ public void testBasicAsync() latch.countDown(); } }; - zk.addPersistentWatch("/a/b", persistentWatcher, true, cb, null); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT_RECURSIVE, cb, null); Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); internalTestBasic(zk); } @@ -98,7 +99,7 @@ private void internalTestBasic(ZooKeeper zk) throws KeeperException, Interrupted public void testRemoval() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, true); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT_RECURSIVE); zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); @@ -114,7 +115,7 @@ public void testRemoval() @Test public void testDisconnect() throws Exception { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, true); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT_RECURSIVE); stopServer(); assertEvent(events, Watcher.Event.EventType.None, null); startServer(); @@ -132,7 +133,7 @@ public void testMultiClient() zk1.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk1.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - zk1.addPersistentWatch("/a/b", persistentWatcher, true); + zk1.addWatch("/a/b", persistentWatcher, PERSISTENT_RECURSIVE); zk1.setData("/a/b/c", "one".getBytes(), -1); Thread.sleep(1000); // give some time for the event to arrive @@ -151,7 +152,7 @@ public void testMultiClient() public void testRootWatcher() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/", persistentWatcher, true); + zk.addWatch("/", persistentWatcher, PERSISTENT_RECURSIVE); zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java index 49c926932a6..565d57da04d 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.test; +import static org.apache.zookeeper.AddWatchMode.PERSISTENT; import java.io.IOException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -54,7 +55,7 @@ public void setUp() throws Exception { public void testBasic() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, false); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT); internalTestBasic(zk); } } @@ -69,7 +70,7 @@ public void testBasicAsync() latch.countDown(); } }; - zk.addPersistentWatch("/a/b", persistentWatcher, false, cb, null); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT, cb, null); Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); internalTestBasic(zk); } @@ -96,7 +97,7 @@ private void internalTestBasic(ZooKeeper zk) throws KeeperException, Interrupted public void testRemoval() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, false); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT); zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.create("/a/b/c", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); @@ -113,7 +114,7 @@ public void testRemoval() @Test public void testDisconnect() throws Exception { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/a/b", persistentWatcher, false); + zk.addWatch("/a/b", persistentWatcher, PERSISTENT); stopServer(); assertEvent(events, Watcher.Event.EventType.None, null); startServer(); @@ -131,7 +132,7 @@ public void testMultiClient() zk1.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk1.create("/a/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - zk1.addPersistentWatch("/a/b", persistentWatcher, false); + zk1.addWatch("/a/b", persistentWatcher, PERSISTENT); zk1.setData("/a/b", "one".getBytes(), -1); Thread.sleep(1000); // give some time for the event to arrive @@ -150,7 +151,7 @@ public void testMultiClient() public void testRootWatcher() throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { - zk.addPersistentWatch("/", persistentWatcher, false); + zk.addWatch("/", persistentWatcher, PERSISTENT); zk.create("/a", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.setData("/a", new byte[0], -1); zk.create("/b", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java deleted file mode 100644 index 2959a4a1d41..00000000000 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/RecursiveWatchQtyTest.java +++ /dev/null @@ -1,108 +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.test; - -import org.apache.zookeeper.WatchedEvent; -import org.apache.zookeeper.Watcher; -import org.apache.zookeeper.server.watch.WatchManager; -import org.apache.zookeeper.server.watch.WatcherMode; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; - -public class RecursiveWatchQtyTest { - private WatchManager watchManager; - - private static class DummyWatcher implements Watcher { - @Override - public void process(WatchedEvent event) { - // NOP - } - } - - @Before - public void setup() { - watchManager = new WatchManager(); - } - - @Test - public void testAddRemove() { - Watcher watcher1 = new DummyWatcher(); - Watcher watcher2 = new DummyWatcher(); - - watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); - watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); - Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); - Assert.assertTrue(watchManager.removeWatcher("/a", watcher1)); - Assert.assertTrue(watchManager.removeWatcher("/b", watcher2)); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - } - - @Test - public void testAddRemoveAlt() { - Watcher watcher1 = new DummyWatcher(); - Watcher watcher2 = new DummyWatcher(); - - watchManager.addWatch("/a", watcher1, WatcherMode.PERSISTENT_RECURSIVE); - watchManager.addWatch("/b", watcher2, WatcherMode.PERSISTENT_RECURSIVE); - Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); - watchManager.removeWatcher(watcher1); - watchManager.removeWatcher(watcher2); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - } - - @Test - public void testDoubleAdd() { - Watcher watcher = new DummyWatcher(); - - watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); - watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); - Assert.assertEquals(1, watchManager.getRecursiveWatchQty()); - watchManager.removeWatcher(watcher); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - } - - @Test - public void testSameWatcherMultiPath() { - Watcher watcher = new DummyWatcher(); - - watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); - watchManager.addWatch("/a/b", watcher, WatcherMode.PERSISTENT_RECURSIVE); - watchManager.addWatch("/a/b/c", watcher, WatcherMode.PERSISTENT_RECURSIVE); - Assert.assertEquals(3, watchManager.getRecursiveWatchQty()); - Assert.assertTrue(watchManager.removeWatcher("/a/b", watcher)); - Assert.assertEquals(2, watchManager.getRecursiveWatchQty()); - watchManager.removeWatcher(watcher); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - } - - @Test - public void testChangeType() { - Watcher watcher = new DummyWatcher(); - - watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - watchManager.addWatch("/a", watcher, WatcherMode.PERSISTENT_RECURSIVE); - Assert.assertEquals(1, watchManager.getRecursiveWatchQty()); - watchManager.addWatch("/a", watcher, WatcherMode.STANDARD); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - Assert.assertTrue(watchManager.removeWatcher("/a", watcher)); - Assert.assertEquals(0, watchManager.getRecursiveWatchQty()); - } -} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/UnsupportedAddWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/UnsupportedAddWatcherTest.java new file mode 100644 index 00000000000..95b5569bb08 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/UnsupportedAddWatcherTest.java @@ -0,0 +1,124 @@ +/* + * 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.io.PrintWriter; +import java.util.Collections; +import org.apache.zookeeper.AddWatchMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.server.watch.IWatchManager; +import org.apache.zookeeper.server.watch.WatchManagerFactory; +import org.apache.zookeeper.server.watch.WatcherOrBitSet; +import org.apache.zookeeper.server.watch.WatchesPathReport; +import org.apache.zookeeper.server.watch.WatchesReport; +import org.apache.zookeeper.server.watch.WatchesSummary; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class UnsupportedAddWatcherTest extends ClientBase { + + public static class StubbedWatchManager implements IWatchManager { + @Override + public boolean addWatch(String path, Watcher watcher) { + return false; + } + + @Override + public boolean containsWatcher(String path, Watcher watcher) { + return false; + } + + @Override + public boolean removeWatcher(String path, Watcher watcher) { + return false; + } + + @Override + public void removeWatcher(Watcher watcher) { + // NOP + } + + @Override + public WatcherOrBitSet triggerWatch(String path, Watcher.Event.EventType type) { + return new WatcherOrBitSet(Collections.emptySet()); + } + + @Override + public WatcherOrBitSet triggerWatch(String path, Watcher.Event.EventType type, WatcherOrBitSet suppress) { + return new WatcherOrBitSet(Collections.emptySet()); + } + + @Override + public int size() { + return 0; + } + + @Override + public void shutdown() { + // NOP + } + + @Override + public WatchesSummary getWatchesSummary() { + return null; + } + + @Override + public WatchesReport getWatches() { + return null; + } + + @Override + public WatchesPathReport getWatchesByPath() { + return null; + } + + @Override + public void dumpWatches(PrintWriter pwriter, boolean byPath) { + // NOP + } + } + + @Before + public void setUp() throws Exception { + System.setProperty(WatchManagerFactory.ZOOKEEPER_WATCH_MANAGER_NAME, StubbedWatchManager.class.getName()); + super.setUp(); + } + + @After + public void tearDown() throws Exception { + try { + super.tearDown(); + } finally { + System.clearProperty(WatchManagerFactory.ZOOKEEPER_WATCH_MANAGER_NAME); + } + } + + @Test(expected = KeeperException.MarshallingErrorException.class) + public void testBehavior() throws IOException, InterruptedException, KeeperException { + try (ZooKeeper zk = createClient(hostPort)) { + // the server will generate an exception as our custom watch manager doesn't implement + // the new version of addWatch() + zk.addWatch("/foo", event -> {}, AddWatchMode.PERSISTENT_RECURSIVE); + } + } +} From 4da2834777eda9853e4ce2483fbaeab6b4b787b1 Mon Sep 17 00:00:00 2001 From: randgalt Date: Wed, 23 Oct 2019 13:04:05 +0300 Subject: [PATCH 4/5] ZOOKEEPER-1416 Added CLI command that corresponds to new addWatch() method in ZooKeeper --- .../org/apache/zookeeper/ZooKeeperMain.java | 11 +++ .../apache/zookeeper/cli/AddWatchCommand.java | 87 +++++++++++++++++++ .../org/apache/zookeeper/cli/CliCommand.java | 9 ++ .../org/apache/zookeeper/cli/Helpers.java | 34 ++++++++ 4 files changed, 141 insertions(+) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java index c9f49a16ec2..1df2d600a4e 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java @@ -38,6 +38,7 @@ import org.apache.yetus.audience.InterfaceAudience; import org.apache.zookeeper.admin.ZooKeeperAdmin; import org.apache.zookeeper.cli.AddAuthCommand; +import org.apache.zookeeper.cli.AddWatchCommand; import org.apache.zookeeper.cli.CliCommand; import org.apache.zookeeper.cli.CliException; import org.apache.zookeeper.cli.CloseCommand; @@ -51,6 +52,7 @@ import org.apache.zookeeper.cli.GetCommand; import org.apache.zookeeper.cli.GetConfigCommand; import org.apache.zookeeper.cli.GetEphemeralsCommand; +import org.apache.zookeeper.cli.Helpers; import org.apache.zookeeper.cli.ListQuotaCommand; import org.apache.zookeeper.cli.Ls2Command; import org.apache.zookeeper.cli.LsCommand; @@ -88,6 +90,13 @@ public class ZooKeeperMain { protected ZooKeeper zk; protected String host = ""; + private final Helpers helpers = new Helpers() { + @Override + public Watcher newWatcher() { + return new MyWatcher(); + } + }; + public boolean getPrintWatches() { return printWatches; } @@ -123,6 +132,7 @@ public boolean getPrintWatches() { new GetEphemeralsCommand().addToMap(commandMapCli); new GetAllChildrenNumberCommand().addToMap(commandMapCli); new VersionCommand().addToMap(commandMapCli); + new AddWatchCommand().addToMap(commandMapCli); // add all to commandMap for (Entry entry : commandMapCli.entrySet()) { @@ -437,6 +447,7 @@ protected boolean processZKCmd(MyCommandOptions co) throws CliException, IOExcep CliCommand cliCmd = commandMapCli.get(cmd); if (cliCmd != null) { cliCmd.setZk(zk); + cliCmd.setHelpers(helpers); watch = cliCmd.parse(args).exec(); } else if (!commandMap.containsKey(cmd)) { usage(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java new file mode 100644 index 00000000000..f609b45e175 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java @@ -0,0 +1,87 @@ +/* + * 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.cli; + +import java.util.Arrays; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.commons.cli.Parser; +import org.apache.commons.cli.PosixParser; +import org.apache.zookeeper.AddWatchMode; +import org.apache.zookeeper.KeeperException; + +/** + * addWatch command for cli. + * Matches the ZooKeeper API addWatch() + */ +public class AddWatchCommand extends CliCommand { + + private static final Options options = new Options(); + private static final AddWatchMode defaultMode = AddWatchMode.PERSISTENT_RECURSIVE; + + private CommandLine cl; + private AddWatchMode mode = defaultMode; + + static { + options.addOption("m", true, ""); + } + + public AddWatchCommand() { + super("addWatch", "[-m mode] path # optional mode is one of " + + Arrays.toString(AddWatchMode.values()) + " - default is " + defaultMode.name()); + } + + @Override + public CliCommand parse(String[] cmdArgs) throws CliParseException { + Parser parser = new PosixParser(); + try { + cl = parser.parse(options, cmdArgs); + } catch (ParseException ex) { + throw new CliParseException(ex); + } + if (cl.getArgs().length != 2) { + throw new CliParseException(getUsageStr()); + } + + if (cl.hasOption("m")) { + try { + mode = AddWatchMode.valueOf(cl.getOptionValue("m").toUpperCase()); + } catch (IllegalArgumentException e) { + throw new CliParseException(getUsageStr()); + } + } + + return this; + } + + @Override + public boolean exec() throws CliException { + String path = cl.getArgs()[1]; + try { + zk.addWatch(path, helpers.newWatcher(), mode); + } catch (KeeperException | InterruptedException ex) { + throw new CliWrapperException(ex); + } + + return false; + + } + +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java index 83e03a84175..d76cf4c1436 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java @@ -28,6 +28,7 @@ public abstract class CliCommand { protected ZooKeeper zk; + protected Helpers helpers; protected PrintStream out; protected PrintStream err; private String cmdStr; @@ -70,6 +71,14 @@ public void setZk(ZooKeeper zk) { this.zk = zk; } + /** + * Set the helpers instance + * @param helpers instance + */ + public void setHelpers(Helpers helpers) { + this.helpers = helpers; + } + /** * get the string used to call this command * @return diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java new file mode 100644 index 00000000000..89f68ae74f4 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java @@ -0,0 +1,34 @@ +/* + * 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.cli; + +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeperMain; + +/** + * Various helpers for CLI commands + */ +public interface Helpers { + /** + * Return a new watcher that respects settings such as {@link ZooKeeperMain#getPrintWatches()} + * + * @return new watcher + */ + Watcher newWatcher(); +} From a37149a531b5f1a343a72fec5c8ce372480f9161 Mon Sep 17 00:00:00 2001 From: randgalt Date: Sun, 3 Nov 2019 09:14:47 -0500 Subject: [PATCH 5/5] ZOOKEEPER-1416 Added versions of addWatch that use the default watcher. With this, the "Helpers" in the CLI is no longer needed. --- .../java/org/apache/zookeeper/ZooKeeper.java | 35 ++++++++++++++++ .../org/apache/zookeeper/ZooKeeperMain.java | 9 ---- .../apache/zookeeper/cli/AddWatchCommand.java | 2 +- .../org/apache/zookeeper/cli/CliCommand.java | 9 ---- .../org/apache/zookeeper/cli/Helpers.java | 34 --------------- .../zookeeper/test/PersistentWatcherTest.java | 41 +++++++++++++++++++ 6 files changed, 77 insertions(+), 53 deletions(-) delete mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java index f2e52c9503f..f6f165deaf4 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java @@ -3165,6 +3165,7 @@ public void removeAllWatches(String path, WatcherType watcherType, boolean local * @throws InterruptedException If the server transaction is interrupted. * @throws KeeperException If the server signals an error with a non-zero * error code. + * @since 3.6.0 */ public void addWatch(String basePath, Watcher watcher, AddWatchMode mode) throws KeeperException, InterruptedException { @@ -3182,6 +3183,24 @@ public void addWatch(String basePath, Watcher watcher, AddWatchMode mode) } } + /** + * Add a watch to the given znode using the given mode. Note: not all + * watch types can be set with this method. Only the modes available + * in {@link AddWatchMode} can be set with this method. In this version of the method, + * the default watcher is used + * + * @param basePath the path that the watcher applies to + * @param mode type of watcher to add + * @throws InterruptedException If the server transaction is interrupted. + * @throws KeeperException If the server signals an error with a non-zero + * error code. + * @since 3.6.0 + */ + public void addWatch(String basePath, AddWatchMode mode) + throws KeeperException, InterruptedException { + addWatch(basePath, watchManager.defaultWatcher, mode); + } + /** * Async version of {@link #addWatch(String, Watcher, AddWatchMode)} (see it for details) * @@ -3191,6 +3210,7 @@ public void addWatch(String basePath, Watcher watcher, AddWatchMode mode) * @param cb a handler for the callback * @param ctx context to be provided to the callback * @throws IllegalArgumentException if an invalid path is specified + * @since 3.6.0 */ public void addWatch(String basePath, Watcher watcher, AddWatchMode mode, VoidCallback cb, Object ctx) { @@ -3204,6 +3224,21 @@ public void addWatch(String basePath, Watcher watcher, AddWatchMode mode, basePath, serverPath, ctx, new AddWatchRegistration(watcher, basePath, mode)); } + /** + * Async version of {@link #addWatch(String, AddWatchMode)} (see it for details) + * + * @param basePath the path that the watcher applies to + * @param mode type of watcher to add + * @param cb a handler for the callback + * @param ctx context to be provided to the callback + * @throws IllegalArgumentException if an invalid path is specified + * @since 3.6.0 + */ + public void addWatch(String basePath, AddWatchMode mode, + VoidCallback cb, Object ctx) { + addWatch(basePath, watchManager.defaultWatcher, mode, cb, ctx); + } + private void validateWatcher(Watcher watcher) { if (watcher == null) { throw new IllegalArgumentException("Invalid Watcher, shouldn't be null!"); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java index 1df2d600a4e..857b16fcc70 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java @@ -52,7 +52,6 @@ import org.apache.zookeeper.cli.GetCommand; import org.apache.zookeeper.cli.GetConfigCommand; import org.apache.zookeeper.cli.GetEphemeralsCommand; -import org.apache.zookeeper.cli.Helpers; import org.apache.zookeeper.cli.ListQuotaCommand; import org.apache.zookeeper.cli.Ls2Command; import org.apache.zookeeper.cli.LsCommand; @@ -90,13 +89,6 @@ public class ZooKeeperMain { protected ZooKeeper zk; protected String host = ""; - private final Helpers helpers = new Helpers() { - @Override - public Watcher newWatcher() { - return new MyWatcher(); - } - }; - public boolean getPrintWatches() { return printWatches; } @@ -447,7 +439,6 @@ protected boolean processZKCmd(MyCommandOptions co) throws CliException, IOExcep CliCommand cliCmd = commandMapCli.get(cmd); if (cliCmd != null) { cliCmd.setZk(zk); - cliCmd.setHelpers(helpers); watch = cliCmd.parse(args).exec(); } else if (!commandMap.containsKey(cmd)) { usage(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java index f609b45e175..1e34b10159c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/AddWatchCommand.java @@ -75,7 +75,7 @@ public CliCommand parse(String[] cmdArgs) throws CliParseException { public boolean exec() throws CliException { String path = cl.getArgs()[1]; try { - zk.addWatch(path, helpers.newWatcher(), mode); + zk.addWatch(path, mode); } catch (KeeperException | InterruptedException ex) { throw new CliWrapperException(ex); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java index d76cf4c1436..83e03a84175 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/CliCommand.java @@ -28,7 +28,6 @@ public abstract class CliCommand { protected ZooKeeper zk; - protected Helpers helpers; protected PrintStream out; protected PrintStream err; private String cmdStr; @@ -71,14 +70,6 @@ public void setZk(ZooKeeper zk) { this.zk = zk; } - /** - * Set the helpers instance - * @param helpers instance - */ - public void setHelpers(Helpers helpers) { - this.helpers = helpers; - } - /** * get the string used to call this command * @return diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java deleted file mode 100644 index 89f68ae74f4..00000000000 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/Helpers.java +++ /dev/null @@ -1,34 +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.cli; - -import org.apache.zookeeper.Watcher; -import org.apache.zookeeper.ZooKeeperMain; - -/** - * Various helpers for CLI commands - */ -public interface Helpers { - /** - * Return a new watcher that respects settings such as {@link ZooKeeperMain#getPrintWatches()} - * - * @return new watcher - */ - Watcher newWatcher(); -} diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java index 565d57da04d..bffa8e0a2db 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/PersistentWatcherTest.java @@ -60,9 +60,50 @@ public void testBasic() } } + @Test + public void testDefaultWatcher() + throws IOException, InterruptedException, KeeperException { + CountdownWatcher watcher = new CountdownWatcher() { + @Override + public synchronized void process(WatchedEvent event) { + super.process(event); + events.add(event); + } + }; + try (ZooKeeper zk = createClient(watcher, hostPort)) { + zk.addWatch("/a/b", PERSISTENT); + events.clear(); // clear any events added during client connection + internalTestBasic(zk); + } + } + @Test public void testBasicAsync() throws IOException, InterruptedException, KeeperException { + CountdownWatcher watcher = new CountdownWatcher() { + @Override + public synchronized void process(WatchedEvent event) { + super.process(event); + events.add(event); + } + }; + try (ZooKeeper zk = createClient(watcher, hostPort)) { + final CountDownLatch latch = new CountDownLatch(1); + AsyncCallback.VoidCallback cb = (rc, path, ctx) -> { + if (rc == 0) { + latch.countDown(); + } + }; + zk.addWatch("/a/b", persistentWatcher, PERSISTENT, cb, null); + Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); + events.clear(); // clear any events added during client connection + internalTestBasic(zk); + } + } + + @Test + public void testAsyncDefaultWatcher() + throws IOException, InterruptedException, KeeperException { try (ZooKeeper zk = createClient(new CountdownWatcher(), hostPort)) { final CountDownLatch latch = new CountDownLatch(1); AsyncCallback.VoidCallback cb = (rc, path, ctx) -> {