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 <[email protected]> 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)); + } +}
