You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2026/09/23 10:58:11 UTC
(accumulo) branch main updated (01bb841eee -> ad3c94ccad)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 01bb841eee Merge branch '2.1' into main
add 5c8c202723 Reverted setting interrupted flag in ZooCache (#6552)
new ad3c94ccad Merge branch '2.1'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/accumulo/core/zookeeper/ZooCache.java | 70 ++++++++++------------
.../accumulo/core/zookeeper/ZooCacheTest.java | 6 +-
2 files changed, 37 insertions(+), 39 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by dl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit ad3c94ccad468b9147be63cf3d132cab4f229f22
Merge: 01bb841eee 5c8c202723
Author: Dave Marion <dl...@apache.org>
AuthorDate: Wed Sep 23 10:57:53 2026 +0000
Merge branch '2.1'
.../apache/accumulo/core/zookeeper/ZooCache.java | 70 ++++++++++------------
.../accumulo/core/zookeeper/ZooCacheTest.java | 6 +-
2 files changed, 37 insertions(+), 39 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
index 1d85c358df,0000000000..a5d0911c15
mode 100644,000000..100644
--- a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
@@@ -1,605 -1,0 +1,601 @@@
+/*
+ * 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
+ *
+ * https://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.accumulo.core.zookeeper;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Objects.requireNonNull;
+import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLockData;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.util.cache.Caches;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Ticker;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+/**
+ * A cache for values stored in ZooKeeper. Values are kept up to date as they change.
+ */
+@SuppressFBWarnings(value = "CT_CONSTRUCTOR_THROW",
+ justification = "Constructor validation is required for proper initialization")
+public class ZooCache {
+
+ public interface ZooCacheWatcher extends Consumer<WatchedEvent> {}
+
+ private static final Logger log = LoggerFactory.getLogger(ZooCache.class);
+
+ private final NavigableSet<String> watchedPaths;
+
+ // visible for tests
+ public final ZCacheWatcher watcher = new ZCacheWatcher();
+ private final List<ZooCacheWatcher> externalWatchers =
+ Collections.synchronizedList(new ArrayList<>());
+
+ private static final AtomicLong nextCacheId = new AtomicLong(0);
+ private final String cacheId = "ZC" + nextCacheId.incrementAndGet();
+
+ public static final Duration CACHE_DURATION = Duration.ofMinutes(30);
+
+ private final Cache<String,ZcNode> cache;
+
+ private final ConcurrentMap<String,ZcNode> nodeCache;
+
+ private final ZooSession zk;
+
+ private volatile boolean closed = false;
+
+ private final AtomicLong updateCount = new AtomicLong();
+
+ private final AtomicLong zkClientTracker = new AtomicLong();
+
+ public class ZCacheWatcher implements Watcher {
+ @Override
+ public void process(WatchedEvent event) {
+ if (log.isTraceEnabled()) {
+ log.trace("{}: {}", cacheId, event);
+ }
+
+ switch (event.getType()) {
+ case NodeChildrenChanged:
+ // According to documentation we should not receive this event.
+ // According to https://issues.apache.org/jira/browse/ZOOKEEPER-4475 we
+ // may receive this event (Fixed in 3.9.0)
+ break;
+ case ChildWatchRemoved:
+ case DataWatchRemoved:
+ // We don't need to do anything with the cache on these events.
+ break;
+ case NodeDataChanged:
+ log.trace("{} node data changed; clearing {}", cacheId, event.getPath());
+ clear(path -> path.equals(event.getPath()));
+ break;
+ case NodeCreated:
+ case NodeDeleted:
+ // With the Watcher being set at a higher level we need to remove
+ // the parent of the affected node and all of its children from the cache
+ // so that the parent and children node can be re-cached. If we only remove the
+ // affected node, then the cached children in the parent could be incorrect.
+ int lastSlash = event.getPath().lastIndexOf('/');
+ String parent = lastSlash == 0 ? "/" : event.getPath().substring(0, lastSlash);
+ log.trace("{} node created or deleted {}; clearing {}", cacheId, event.getPath(), parent);
+ clear((path) -> path.startsWith(parent));
+ break;
+ case PersistentWatchRemoved:
+ log.warn(
+ "{} persistent watch removed {} which is only done in ZooSession.addPersistentRecursiveWatchers; ignoring;",
+ cacheId, event.getPath());
+ break;
+ case None:
+ switch (event.getState()) {
+ case Closed:
+ log.trace("{} ZooKeeper connection closed, clearing cache; {}", cacheId, event);
+ clear();
+ break;
+ case Disconnected:
+ log.trace("{} ZooKeeper connection disconnected, clearing cache; {}", cacheId, event);
+ clear();
+ break;
+ case SyncConnected:
+ log.trace("{} ZooKeeper connection established, ignoring; {}", cacheId, event);
+ break;
+ case Expired:
+ log.trace("{} ZooKeeper connection expired, clearing cache; {}", cacheId, event);
+ clear();
+ break;
+ default:
+ log.warn("{} Unhandled state {}", cacheId, event);
+ break;
+ }
+ break;
+ default:
+ log.warn("{} Unhandled event type {}", cacheId, event);
+ break;
+ }
+
+ externalWatchers.forEach(ew -> {
+ try {
+ ew.accept(event);
+ } catch (Exception e) {
+ log.error(
+ "Exception calling external watcher. This is a bug and could impact proper operation.",
+ e);
+ }
+ });
+ }
+ }
+
+ /**
+ * Creates a ZooCache instance that uses the supplied ZooSession for communicating with the
+ * instance's ZooKeeper servers. The ZooCache will create persistent watchers at the given
+ * pathsToWatch, if any, to be updated when changes are made in ZooKeeper for nodes at or below in
+ * the tree. If ZooCacheWatcher's are added via {@code addZooCacheWatcher}, then they will be
+ * notified when this object is notified of changes via the PersistentWatcher callback.
+ *
+ * @param zk ZooSession for this instance
+ * @param pathsToWatch Paths in ZooKeeper to watch
+ */
+ public ZooCache(ZooSession zk, Set<String> pathsToWatch) {
+ this(zk, pathsToWatch, Ticker.systemTicker());
+ }
+
+ // visible for tests that use a Ticker
+ public ZooCache(ZooSession zk, Set<String> pathsToWatch, Ticker ticker) {
+ this.zk = requireNonNull(zk);
+ // this initial value is meant to indicate watchers were never setup
+ this.zkClientTracker.set(-1);
+ this.cache = Caches.getInstance().createNewBuilder(Caches.CacheName.ZOO_CACHE, false)
+ .ticker(requireNonNull(ticker)).expireAfterAccess(CACHE_DURATION).build();
+ // The concurrent map returned by Caffeine will only allow one thread to run at a time for a
+ // given key and ZooCache relies on that. Not all concurrent map implementations have this
+ // behavior for their compute functions.
+ this.nodeCache = cache.asMap();
+ this.watchedPaths = Collections.unmodifiableNavigableSet(new TreeSet<>(pathsToWatch));
+ setupWatchers();
+ log.trace("{} created new cache watching {}", cacheId, pathsToWatch, new Exception());
+ }
+
+ public void addZooCacheWatcher(ZooCacheWatcher watcher) {
+ externalWatchers.add(requireNonNull(watcher));
+ }
+
+ // visible for tests
+ long getZKClientObjectVersion() {
+ long counter = zk.getConnectionCounter();
+ // -1 is used to signify ZK has not been setup in this code and this code assume ZooSession will
+ // always return something >= 0.
+ Preconditions.checkState(counter >= 0);
+ return counter;
+ }
+
+ /**
+ * @return true if ZK has changed; false otherwise
+ */
+ private boolean handleZKConnectionChange() {
+ final long currentCount = getZKClientObjectVersion();
+ final long oldCount = zkClientTracker.get();
+ if (oldCount != currentCount) {
+ return setupWatchers();
+ }
+ return false;
+ }
+
+ // Called on construction and when ZooKeeper connection changes
+ synchronized boolean setupWatchers() {
+
+ final long currentCount = getZKClientObjectVersion();
+ final long oldCount = zkClientTracker.get();
+
+ if (currentCount == oldCount) {
+ return false;
+ }
+
+ for (String left : watchedPaths) {
+ for (String right : watchedPaths) {
+ if (!right.startsWith("/")) {
+ throw new IllegalArgumentException("Watched path must start with slash: '" + right + "'");
+ }
+ if (right.length() > 1 && right.endsWith("/")) {
+ throw new IllegalArgumentException(
+ "Watched path must not end with slash: '" + right + "'");
+ }
+ if ((left.equals("/") && right.length() > 1) || right.startsWith(left + "/")) {
+ throw new IllegalArgumentException(
+ "Overlapping paths found in paths to watch. '" + left + "' contains '" + right + "'");
+ }
+ }
+ }
+
+ try {
+ long zkId = zk.addPersistentRecursiveWatchers(watchedPaths, watcher);
+ zkClientTracker.set(zkId);
+ clear();
+ log.trace("{} Reinitialized persistent watchers and cleared cache {}", cacheId,
+ zkClientTracker.get());
+ return true;
+ } catch (KeeperException | InterruptedException e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Error setting up persistent recursive watcher", e);
+ }
+
+ }
+
+ private boolean isWatchedPath(String path) {
+ // Check that the path is equal to, or a descendant of, a watched path
+ var floor = watchedPaths.floor(path);
+ return floor != null
+ && (floor.equals("/") || floor.equals(path) || path.startsWith(floor + "/"));
+ }
+
+ private void ensureWatched(String path) {
+ if (!isWatchedPath(path)) {
+ throw new IllegalStateException("Supplied path " + path + " is not watched by this ZooCache");
+ }
+ }
+
+ private abstract class ZooRunnable<T> {
+ /**
+ * Runs an operation against ZooKeeper. Retries are performed by the retry method when
+ * KeeperExceptions occur.
+ *
+ * @return T the result of the runnable
+ */
+ abstract T run() throws KeeperException, InterruptedException;
+
+ /**
+ * Retry will attempt to call the run method.
+ *
+ * @return result of the runnable access success ( i.e. no exceptions ).
+ */
+ public T retry() {
+
+ int sleepTime = 100;
+
- while (true) {
-
- try {
- T result = run();
- if (handleZKConnectionChange()) {
- continue;
- }
- return result;
- } catch (KeeperException | ZcException e) {
- KeeperException ke;
- if (e instanceof ZcException) {
- ke = ((ZcException) e).getZKException();
- } else {
- ke = ((KeeperException) e);
- }
- final Code code = ke.code();
- if (code == Code.NONODE) {
- log.error("Looked up non-existent node in cache " + ke.getPath(), e);
- } else if (code == Code.CONNECTIONLOSS || code == Code.OPERATIONTIMEOUT
- || code == Code.SESSIONEXPIRED) {
- log.warn("Saw (possibly) transient exception communicating with ZooKeeper, will retry",
- e);
- } else {
- log.warn("Zookeeper error, will retry", e);
- }
- } catch (InterruptedException | ZcInterruptedException e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
++ try {
++ while (true) {
++
++ try {
++ T result = run();
++ if (handleZKConnectionChange()) {
++ continue;
++ }
++ return result;
++ } catch (KeeperException | ZcException e) {
++ KeeperException ke;
++ if (e instanceof ZcException) {
++ ke = ((ZcException) e).getZKException();
++ } else {
++ ke = ((KeeperException) e);
++ }
++ final Code code = ke.code();
++ if (code == Code.NONODE) {
++ log.error("Looked up non-existent node in cache " + ke.getPath(), e);
++ } else if (code == Code.CONNECTIONLOSS || code == Code.OPERATIONTIMEOUT
++ || code == Code.SESSIONEXPIRED) {
++ log.warn(
++ "Saw (possibly) transient exception communicating with ZooKeeper, will retry", e);
++ } else {
++ log.warn("Zookeeper error, will retry", e);
++ }
++ } catch (ZcInterruptedException e) {
++ throw (InterruptedException) e.getCause();
+ }
- log.info("Zookeeper error, will retry", e);
- }
+
- try {
- // do not hold lock while sleeping
+ Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.debug("Wait in retry() was interrupted.", e);
- }
- if (sleepTime < 10_000) {
- sleepTime = (int) (sleepTime + sleepTime * RANDOM.get().nextDouble());
++ if (sleepTime < 10_000) {
++ sleepTime = (int) (sleepTime + sleepTime * RANDOM.get().nextDouble());
++ }
+ }
++ } catch (InterruptedException e) {
++ Thread.currentThread().interrupt();
++ throw new IllegalStateException("ZooCache thread interrupted.", e);
+ }
+ }
+
+ }
+
+ private static class ZcException extends RuntimeException {
+ private static final long serialVersionUID = 1;
+
+ private ZcException(KeeperException e) {
+ super(e);
+ }
+
+ public KeeperException getZKException() {
+ return (KeeperException) getCause();
+ }
+ }
+
+ private static class ZcInterruptedException extends RuntimeException {
+ private static final long serialVersionUID = 1;
+
+ private ZcInterruptedException(InterruptedException e) {
+ super(e);
+ }
+ }
+
+ /**
+ * Gets the children of the given node.
+ *
+ * @param zPath path of node
+ * @return children list, or null if node has no children or does not exist
+ */
+ public List<String> getChildren(final String zPath) {
+ Preconditions.checkState(!closed, "Operation not allowed: ZooCache is already closed.");
+ ensureWatched(zPath);
+ ZooRunnable<List<String>> zr = new ZooRunnable<>() {
+
+ @Override
+ public List<String> run() throws KeeperException, InterruptedException {
+
+ var zcNode = nodeCache.get(zPath);
+ if (zcNode != null && zcNode.cachedChildren()) {
+ return zcNode.getChildren();
+ }
+
+ log.trace("{} {} was not in children cache, looking up in zookeeper", cacheId, zPath);
+
+ zcNode = nodeCache.compute(zPath, (zp, zcn) -> {
+ // recheck the children now that lock is held on key
+ if (zcn != null && zcn.cachedChildren()) {
+ return zcn;
+ }
+ try {
+ List<String> children = zk.getChildren(zPath, null);
+ log.trace("{} adding {} children of {} to cache", cacheId, children.size(), zPath);
+ return new ZcNode(children, zcn);
+ } catch (KeeperException.NoNodeException nne) {
+ log.trace("{} getChildren saw that {} does not exists", cacheId, zPath);
+ return ZcNode.NON_EXISTENT;
+ } catch (KeeperException e) {
+ throw new ZcException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ZcInterruptedException(e);
+ }
+ });
+ // increment this after compute call completes when the change is visible
+ updateCount.incrementAndGet();
+ return zcNode.getChildren();
+ }
+ };
+
+ return zr.retry();
+ }
+
+ /**
+ * Gets data at the given path. Status information is not returned.
+ *
+ * @param zPath path to get
+ * @return path data, or null if non-existent
+ */
+ public byte[] get(final String zPath) {
+ return get(zPath, null);
+ }
+
+ /**
+ * Gets data at the given path, filling status information into the given <code>Stat</code>
+ * object.
+ *
+ * @param zPath path to get
+ * @param status status object to populate
+ * @return path data, or null if non-existent
+ */
+ public byte[] get(final String zPath, final ZcStat status) {
+ Preconditions.checkState(!closed, "Operation not allowed: ZooCache is already closed.");
+ ensureWatched(zPath);
+ ZooRunnable<byte[]> zr = new ZooRunnable<>() {
+
+ @Override
+ public byte[] run() throws KeeperException, InterruptedException {
+
+ var zcNode = nodeCache.get(zPath);
+ if (zcNode != null && zcNode.cachedData()) {
+ if (status != null) {
+ copyStats(status, zcNode.getStat());
+ }
+ return zcNode.getData();
+ }
+
+ log.trace("{} {} was not in data cache, looking up in zookeeper", cacheId, zPath);
+
+ zcNode = nodeCache.compute(zPath, (zp, zcn) -> {
+ // recheck the now that lock is held on key, it may be present now. Could have been
+ // computed while waiting for lock.
+ if (zcn != null && zcn.cachedData()) {
+ return zcn;
+ }
+ try {
+ byte[] data = null;
+ Stat stat = new Stat();
+ ZcStat zstat = null;
+ try {
+ data = zk.getData(zPath, null, stat);
+ zstat = new ZcStat(stat);
+ } catch (KeeperException.NoNodeException e1) {
+ log.trace("{} zookeeper did not contain {}", cacheId, zPath);
+ return ZcNode.NON_EXISTENT;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ZcInterruptedException(e);
+ }
+ if (log.isTraceEnabled()) {
+ log.trace("{} zookeeper contained {} {}", cacheId, zPath,
+ (data == null ? null : new String(data, UTF_8)));
+ }
+ return new ZcNode(data, zstat, zcn);
+ } catch (KeeperException ke) {
+ throw new ZcException(ke);
+ }
+ });
+
+ // update this after the compute call completes when the change is visible
+ updateCount.incrementAndGet();
+ if (status != null) {
+ copyStats(status, zcNode.getStat());
+ }
+ return zcNode.getData();
+ }
+ };
+
+ return zr.retry();
+ }
+
+ /**
+ * Helper method to copy stats from the cached stat into userStat
+ *
+ * @param userStat user Stat object
+ * @param cachedStat cached statistic, that is or will be cached
+ */
+ protected void copyStats(ZcStat userStat, ZcStat cachedStat) {
+ Preconditions.checkState(!closed, "Operation not allowed: ZooCache is already closed.");
+ if (userStat != null && cachedStat != null) {
+ userStat.set(cachedStat);
+ }
+ }
+
+ /**
+ * Clears this cache.
+ */
+ protected void clear() {
+ if (closed) {
+ return;
+ }
+ nodeCache.clear();
+ updateCount.incrementAndGet();
+ log.trace("{} cleared all from cache", cacheId);
+ }
+
+ public void close() {
+ if (closed) {
+ return;
+ }
+ clear();
+ closed = true;
+ }
+
+ /**
+ * Returns a monotonically increasing count of the number of time the cache was updated. If the
+ * count is the same, then it means cache did not change.
+ */
+ public long getUpdateCount() {
+ Preconditions.checkState(!closed, "Operation not allowed: ZooCache is already closed.");
+ return updateCount.get();
+ }
+
+ /**
+ * Checks if a data value (or lack of one) is cached.
+ *
+ * @param zPath path of node
+ * @return true if data value is cached
+ */
+ @VisibleForTesting
+ public boolean dataCached(String zPath) {
+ ensureWatched(zPath);
+ var zcn = nodeCache.get(zPath);
+ return zcn != null && zcn.cachedData();
+ }
+
+ /**
+ * Checks if children of a node (or lack of them) are cached.
+ *
+ * @param zPath path of node
+ * @return true if children are cached
+ */
+ @VisibleForTesting
+ public boolean childrenCached(String zPath) {
+ ensureWatched(zPath);
+ var zcn = nodeCache.get(zPath);
+ return zcn != null && zcn.cachedChildren();
+ }
+
+ /**
+ * Removes all paths in the cache match the predicate.
+ */
+ public void clear(Predicate<String> pathPredicate) {
+ Preconditions.checkState(!closed, "Operation not allowed: ZooCache is already closed.");
+ Predicate<String> pathPredicateWrapper = path -> {
+ boolean testResult = pathPredicate.test(path);
+ if (testResult) {
+ updateCount.incrementAndGet();
+ log.trace("{} removing {} from cache", cacheId, path);
+ }
+ return testResult;
+ };
+ nodeCache.keySet().removeIf(pathPredicateWrapper);
+ }
+
+ /**
+ * Clears this cache of all information about nodes rooted at the given path.
+ *
+ * @param zPath path of top node
+ */
+ public void clear(String zPath) {
+ ensureWatched(zPath);
+ clear(path -> path.startsWith(zPath));
+ }
+
+ /**
+ * Gets the lock data from the node in the cache at the specified path
+ */
+ public Optional<ServiceLockData> getLockData(ServiceLockPath path) {
+ ensureWatched(path.toString());
+ List<String> children = ServiceLock.validateAndSort(path, getChildren(path.toString()));
+ if (children == null || children.isEmpty()) {
+ return Optional.empty();
+ }
+ String lockNode = children.get(0);
+
+ byte[] lockData = get(path + "/" + lockNode);
+ if (log.isTraceEnabled()) {
+ log.trace("{} Data from lockNode {} is {}", cacheId, lockNode, new String(lockData, UTF_8));
+ }
+ if (lockData == null) {
+ lockData = new byte[0];
+ }
+ return ServiceLockData.parse(lockData);
+ }
+
+}
diff --cc core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
index def7ac6625,0000000000..ddbec44951
mode 100644,000000..100644
--- a/core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
@@@ -1,450 -1,0 +1,452 @@@
+/*
+ * 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
+ *
+ * https://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.accumulo.core.zookeeper;
+
+import static org.easymock.EasyMock.createStrictMock;
+import static org.easymock.EasyMock.eq;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.isA;
+import static org.easymock.EasyMock.isNull;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.zookeeper.ZooCache.ZooCacheWatcher;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.easymock.EasyMock;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class ZooCacheTest {
+
+ /**
+ * Test class that extends ZooCache to suppress the creation of the persistent recursive watchers
+ * that are created in the constructor and to provide access to the watcher.
+ */
+ private static class TestZooCache extends ZooCache {
+
+ public TestZooCache(ZooSession zk, Set<String> pathsToWatch) {
+ super(zk, pathsToWatch);
+ }
+
+ @Override
+ protected boolean setupWatchers() {
+ return false;
+ }
+
+ public void executeWatcher(WatchedEvent event) {
+ // simulate ZooKeeper calling our Watcher
+ watcher.process(event);
+ }
+
+ @Override
+ long getZKClientObjectVersion() {
+ return 1L;
+ }
+
+ }
+
+ private static final String root = Constants.ZROOT + "/" + UUID.randomUUID().toString();
+ private static final String ZPATH = root + "/testPath";
+ private static final byte[] DATA = {(byte) 1, (byte) 2, (byte) 3, (byte) 4};
+ private static final List<String> CHILDREN = java.util.Arrays.asList("huey", "dewey", "louie");
+
+ private ZooSession zk;
+ private TestZooCache zc;
+
+ @BeforeEach
+ public void setUp() {
+ zk = createStrictMock(ZooSession.class);
+ zc = new TestZooCache(zk, Set.of(root));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOverlappingPaths() throws Exception {
+ expect(zk.getConnectionCounter()).andReturn(2L).times(7);
+ expect(zk.addPersistentRecursiveWatchers(isA(Set.class), isA(Watcher.class))).andReturn(3L);
+ replay(zk);
+ var e = assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of("/test/localhost:9995", "/test")));
+ assertEquals(
+ "Overlapping paths found in paths to watch. '/test' contains '/test/localhost:9995'",
+ e.getMessage());
+ e = assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of("/test/localhost:9995", "/test")));
+ assertEquals(
+ "Overlapping paths found in paths to watch. '/test' contains '/test/localhost:9995'",
+ e.getMessage());
+ e = assertThrows(IllegalArgumentException.class, () -> new ZooCache(zk, Set.of("/test", "/")));
+ assertEquals("Overlapping paths found in paths to watch. '/' contains '/test'", e.getMessage());
+ e = assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of("/test/", "/test")));
+ assertEquals("Watched path must not end with slash: '/test/'", e.getMessage());
+ e = assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of("/test", "/test/")));
+ assertEquals("Watched path must not end with slash: '/test/'", e.getMessage());
+ e = assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of("/test", "test")));
+ assertEquals("Watched path must start with slash: 'test'", e.getMessage());
+
+ // "/co" is added because it is a prefix of /compactors, but being a prefix is not enough to
+ // overlap; it needs to actually be a child of the other, not just have a similar name
+ Set<String> goodPaths = Set.of("/co", "/compactors", "/dead/tservers", "/gc/lock",
+ "/managers/lock", "/namespaces", "/recovery", "/root_tablet", "/sservers", "/tables",
+ "/tservers", "/users", "/mini", "/monitor/lock");
+ new ZooCache(zk, goodPaths);
+ verify(zk);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testUnwatchedPaths() throws Exception {
+ expect(zk.getConnectionCounter()).andReturn(2L).anyTimes();
+ expect(zk.addPersistentRecursiveWatchers(isA(Set.class), isA(Watcher.class))).andReturn(3L);
+ replay(zk);
+ ZooCache cache = new ZooCache(zk, Set.of(root));
+ assertThrows(IllegalStateException.class, () -> cache.get("/some/unknown/path"));
+ assertThrows(IllegalStateException.class, () -> cache.getChildren("/some/unknown/path"));
+ verify(zk);
+ }
+
+ @Test
+ public void testGet() throws Exception {
+
+ expect(zk.getData(eq(ZPATH), isNull(), isA(Stat.class))).andAnswer(() -> {
+ ((Stat) EasyMock.getCurrentArguments()[2]).setEphemeralOwner(123456789);
+ return DATA;
+ });
+ replay(zk);
+
+ assertFalse(zc.dataCached(ZPATH));
+ ZcStat zcStat = new ZcStat();
+ assertSame(DATA, zc.get(ZPATH, zcStat));
+ assertEquals(123456789, zcStat.getEphemeralOwner());
+ assertSame(DATA, zc.get(ZPATH));
+ verify(zk);
+ }
+
+ @Test
+ public void testGet_NonExistent() throws Exception {
+ expect(zk.getData(ZPATH, null, new Stat()))
+ .andThrow(new KeeperException.NoNodeException(ZPATH));
+ replay(zk);
+
+ assertNull(zc.get(ZPATH));
+ verify(zk);
+ }
+
+ @Test
+ public void testGet_Retry_NoNode() throws Exception {
+ testGet_Retry(new KeeperException.NoNodeException(ZPATH));
+ }
+
+ @Test
+ public void testGet_Retry_ConnectionLoss() throws Exception {
+ testGet_Retry(new KeeperException.ConnectionLossException());
+ }
+
+ @Test
+ public void testGet_Retry_Interrupted() throws Exception {
+ testGet_Retry(new InterruptedException());
+ }
+
+ private void testGet_Retry(Exception e) throws Exception {
+ Stat existsStat = new Stat();
+ expect(zk.getData(ZPATH, null, existsStat)).andThrow(e);
- if (!(e instanceof KeeperException.NoNodeException)) {
- // Retry will not happen on NoNodeException or BadVersionException
++ if (!(e instanceof KeeperException.NoNodeException) && !(e instanceof InterruptedException)) {
++ // Retry will not happen on NoNodeException, BadVersionException, or InterruptedException
+ expect(zk.getData(ZPATH, null, existsStat)).andReturn(DATA);
+ }
+ replay(zk);
+
+ if (e instanceof KeeperException.NoNodeException
+ || e instanceof KeeperException.BadVersionException) {
+ assertNull(zc.get(ZPATH));
++ } else if (e instanceof InterruptedException) {
++ assertThrows(IllegalStateException.class, () -> zc.get(ZPATH));
+ } else {
+ assertArrayEquals(DATA, zc.get(ZPATH));
+ }
+ verify(zk);
+ }
+
+ @Test
+ public void testGetChildren() throws Exception {
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
+ replay(zk);
+
+ assertFalse(zc.childrenCached(ZPATH));
+ assertEquals(CHILDREN, zc.getChildren(ZPATH));
+ verify(zk);
+
+ assertTrue(zc.childrenCached(ZPATH));
+ // cannot check for sameness, return value is wrapped each time
+ assertEquals(CHILDREN, zc.getChildren(ZPATH)); // cache hit
+ }
+
+ @Test
+ public void testGetChildren_NoKids() throws Exception {
+ expect(zk.getChildren(ZPATH, null)).andReturn(List.of());
+ replay(zk);
+
+ assertEquals(List.of(), zc.getChildren(ZPATH));
+ verify(zk);
+
+ assertEquals(List.of(), zc.getChildren(ZPATH)); // cache hit
+ }
+
+ @Test
+ public void testGetChildren_Retry() throws Exception {
+ expect(zk.getChildren(ZPATH, null)).andThrow(new KeeperException.BadVersionException(ZPATH));
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
+ replay(zk);
+
+ assertEquals(CHILDREN, zc.getChildren(ZPATH));
+ verify(zk);
+ assertEquals(CHILDREN, zc.getChildren(ZPATH));
+ }
+
+ @Test
+ public void testGetChildren_NoNode() throws Exception {
+ assertFalse(zc.childrenCached(ZPATH));
+ assertFalse(zc.dataCached(ZPATH));
+ expect(zk.getChildren(ZPATH, null)).andThrow(new KeeperException.NoNodeException(ZPATH));
+ replay(zk);
+
+ assertNull(zc.getChildren(ZPATH));
+ verify(zk);
+ assertNull(zc.getChildren(ZPATH));
+ // when its discovered a node does not exists in getChildren then its also known it does not
+ // exists for getData
+ assertNull(zc.get(ZPATH));
+ assertTrue(zc.childrenCached(ZPATH));
+ assertTrue(zc.dataCached(ZPATH));
+ }
+
+ private static class TestWatcher implements ZooCacheWatcher {
+ private final WatchedEvent expectedEvent;
+ private boolean wasCalled;
+
+ TestWatcher(WatchedEvent event) {
+ expectedEvent = event;
+ wasCalled = false;
+ }
+
+ @Override
+ public void accept(WatchedEvent event) {
+ assertSame(expectedEvent, event);
+ wasCalled = true;
+ }
+
+ boolean wasCalled() {
+ return wasCalled;
+ }
+ }
+
+ @Test
+ public void testWatchDataNode_Deleted() throws Exception {
+ testWatchDataNode(DATA, Watcher.Event.EventType.NodeDeleted, false);
+ }
+
+ @Test
+ public void testWatchDataNode_DataChanged() throws Exception {
+ testWatchDataNode(DATA, Watcher.Event.EventType.NodeDataChanged, false);
+ }
+
+ @Test
+ public void testWatchDataNode_Created() throws Exception {
+ testWatchDataNode(null, Watcher.Event.EventType.NodeCreated, false);
+ }
+
+ @Test
+ public void testWatchDataNode_NoneSyncConnected() throws Exception {
+ testWatchDataNode(null, Watcher.Event.EventType.None, true);
+ }
+
+ private void testWatchDataNode(byte[] initialData, Watcher.Event.EventType eventType,
+ boolean stillCached) throws Exception {
+ WatchedEvent event =
+ new WatchedEvent(eventType, Watcher.Event.KeeperState.SyncConnected, ZPATH);
+ TestWatcher exw = new TestWatcher(event);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
+
+ watchData(initialData, stillCached);
+ zc.executeWatcher(event);
+ assertTrue(exw.wasCalled());
+ assertEquals(stillCached, zc.dataCached(ZPATH));
+ }
+
+ private void watchData(byte[] initialData, boolean cached) throws Exception {
+ Stat existsStat = new Stat();
+ if (initialData != null) {
+ expect(zk.getData(ZPATH, null, existsStat)).andReturn(initialData);
+ } else {
+ expect(zk.getData(ZPATH, null, existsStat))
+ .andThrow(new KeeperException.NoNodeException(ZPATH));
+ }
+ replay(zk);
+ zc.get(ZPATH);
+ assertTrue(zc.dataCached(ZPATH));
+ }
+
+ @Test
+ public void testWatchDataNode_Disconnected() throws Exception {
+ testWatchDataNode_Clear(Watcher.Event.KeeperState.Disconnected);
+ }
+
+ @Test
+ public void testWatchDataNode_Expired() throws Exception {
+ testWatchDataNode_Clear(Watcher.Event.KeeperState.Expired);
+ }
+
+ @Test
+ public void testGetDataThenChildren() throws Exception {
+ testGetBoth(true);
+ }
+
+ @Test
+ public void testGetChildrenThenDate() throws Exception {
+ testGetBoth(false);
+ }
+
+ private void testGetBoth(boolean getDataFirst) throws Exception {
+ assertFalse(zc.childrenCached(ZPATH));
+ assertFalse(zc.dataCached(ZPATH));
+
+ var uc1 = zc.getUpdateCount();
+
+ if (getDataFirst) {
+ expect(zk.getData(ZPATH, null, new Stat())).andReturn(DATA);
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
+ } else {
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
+ expect(zk.getData(ZPATH, null, new Stat())).andReturn(DATA);
+ }
+
+ replay(zk);
+
+ if (getDataFirst) {
+ var zcStat = new ZcStat();
+ var data = zc.get(ZPATH, zcStat);
+ assertArrayEquals(DATA, data);
+ } else {
+ var children = zc.getChildren(ZPATH);
+ assertEquals(CHILDREN, children);
+ }
+ var uc2 = zc.getUpdateCount();
+ assertTrue(uc1 < uc2);
+
+ if (getDataFirst) {
+ var children = zc.getChildren(ZPATH);
+ assertEquals(CHILDREN, children);
+ } else {
+ var zcStat = new ZcStat();
+ var data = zc.get(ZPATH, zcStat);
+ assertArrayEquals(DATA, data);
+ }
+ var uc3 = zc.getUpdateCount();
+ assertTrue(uc2 < uc3);
+
+ verify(zk);
+
+ var zcStat = new ZcStat();
+ var data = zc.get(ZPATH, zcStat);
+ // the stat is associated with the data so should aways see the one returned by the call to get
+ // data and not get children
+ assertArrayEquals(DATA, data);
+ var children = zc.getChildren(ZPATH);
+ assertEquals(CHILDREN, children);
+ // everything is cached so the get calls on the cache should not change the update count
+ assertEquals(uc3, zc.getUpdateCount());
+ }
+
+ private void testWatchDataNode_Clear(Watcher.Event.KeeperState state) throws Exception {
+ WatchedEvent event = new WatchedEvent(Watcher.Event.EventType.None, state, null);
+ TestWatcher exw = new TestWatcher(event);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
+
+ watchData(DATA, true);
+ assertTrue(zc.dataCached(ZPATH));
+ zc.executeWatcher(event);
+ assertTrue(exw.wasCalled());
+ assertFalse(zc.dataCached(ZPATH));
+ }
+
+ @Test
+ public void testWatchChildrenNode_Deleted() throws Exception {
+ testWatchChildrenNode(CHILDREN, Watcher.Event.EventType.NodeDeleted, false);
+ }
+
+ @Test
+ public void testWatchChildrenNode_ChildrenChanged() throws Exception {
+ testWatchChildrenNode(CHILDREN, Watcher.Event.EventType.NodeChildrenChanged, true);
+ }
+
+ @Test
+ public void testWatchChildrenNode_Created() throws Exception {
+ testWatchChildrenNode(null, Watcher.Event.EventType.NodeCreated, false);
+ }
+
+ @Test
+ public void testWatchChildrenNode_NoneSyncConnected() throws Exception {
+ testWatchChildrenNode(CHILDREN, Watcher.Event.EventType.None, true);
+ }
+
+ private void testWatchChildrenNode(List<String> initialChildren,
+ Watcher.Event.EventType eventType, boolean stillCached) throws Exception {
+ WatchedEvent event =
+ new WatchedEvent(eventType, Watcher.Event.KeeperState.SyncConnected, ZPATH);
+ TestWatcher exw = new TestWatcher(event);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
+
+ watchChildren(initialChildren);
+ zc.executeWatcher(event);
+ assertTrue(exw.wasCalled());
+ assertEquals(stillCached, zc.childrenCached(ZPATH));
+ }
+
+ private void watchChildren(List<String> initialChildren) throws Exception {
+ if (initialChildren == null) {
+ expect(zk.getChildren(ZPATH, null)).andReturn(List.of());
+ } else {
+ expect(zk.getChildren(ZPATH, null)).andReturn(initialChildren);
+ }
+ replay(zk);
+ zc.getChildren(ZPATH);
+ assertTrue(zc.childrenCached(ZPATH));
+ }
+}