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 2025/02/03 21:35:56 UTC
(accumulo) branch main updated: Reused ZooCache where possible, used persistent recursive watchers (#5256)
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
The following commit(s) were added to refs/heads/main by this push:
new e523deed4d Reused ZooCache where possible, used persistent recursive watchers (#5256)
e523deed4d is described below
commit e523deed4d52a0746244bc7d03431b0c0fe0ebe1
Author: Dave Marion <dl...@apache.org>
AuthorDate: Mon Feb 3 16:35:50 2025 -0500
Reused ZooCache where possible, used persistent recursive watchers (#5256)
This change modifies ZooCache to use Persistent Recursive watchers, a
ZooKeeper 3.6.0 feature, in an attempt to create a single shared ZooCache
instance in the codebase. This change get close, I think it get's it down to
2 instances, both created in ClientContext. This change bumps the minimum
ZooKeeper version from 3.5.9 to 3.6.0 for the main branch.
Related to #5134
Closes #5154, #5157
Co-authored-by: Christopher Tubbs <ct...@apache.org>
Co-authored-by: Keith Turner <kt...@apache.org>
---
.github/workflows/maven.yaml | 2 +-
.../accumulo/core/clientImpl/ClientContext.java | 33 +-
.../accumulo/core/clientImpl/NamespaceMapping.java | 5 +-
.../core/clientImpl/ZookeeperLockChecker.java | 2 +-
.../org/apache/accumulo/core/lock/ServiceLock.java | 8 +-
.../accumulo/core/lock/ServiceLockPaths.java | 4 +-
.../core/metadata/schema/TabletMetadata.java | 4 +-
.../accumulo/core/rpc/clients/TServerClient.java | 2 +-
.../util/compaction/ExternalCompactionUtil.java | 2 +-
.../apache/accumulo/core/util/tables/TableMap.java | 2 +-
.../accumulo/core/util/tables/TableZooHelper.java | 2 +-
.../accumulo/core/{fate => }/zookeeper/ZcNode.java | 9 +-
.../org/apache/accumulo/core/zookeeper/ZcStat.java | 53 ++++
.../core/{fate => }/zookeeper/ZooCache.java | 343 +++++++++++----------
.../apache/accumulo/core/zookeeper/ZooSession.java | 42 +++
.../core/clientImpl/ZookeeperLockCheckerTest.java | 2 +-
.../accumulo/core/lock/ServiceLockPathsTest.java | 4 +-
.../core/{fate => }/zookeeper/ZooCacheTest.java | 273 ++++++++--------
.../miniclusterImpl/MiniAccumuloClusterImpl.java | 28 +-
.../accumulo/server/manager/LiveTServerSet.java | 90 ++++--
.../security/handler/KerberosAuthenticator.java | 40 +--
.../server/security/handler/ZKAuthenticator.java | 58 ++--
.../server/security/handler/ZKAuthorizor.java | 23 +-
.../server/security/handler/ZKPermHandler.java | 129 ++++----
.../accumulo/server/tables/TableManager.java | 40 +--
.../org/apache/accumulo/server/util/Admin.java | 2 +-
.../accumulo/server/util/TabletServerLocks.java | 2 +-
.../server/manager/LiveTServerSetTest.java | 13 +-
.../security/handler/ZKAuthenticatorTest.java | 17 +-
.../org/apache/accumulo/server/util/AdminTest.java | 4 +-
.../accumulo/server/util/ServiceStatusCmdTest.java | 4 +-
.../java/org/apache/accumulo/manager/Manager.java | 2 +-
.../accumulo/manager/recovery/RecoveryManager.java | 5 +-
.../org/apache/accumulo/tserver/ScanServer.java | 10 -
.../accumulo/tserver/TabletClientHandler.java | 8 +-
.../accumulo/tserver/TabletHostingServer.java | 3 -
.../org/apache/accumulo/tserver/TabletServer.java | 9 -
.../accumulo/test/BadDeleteMarkersCreatedIT.java | 3 +-
.../accumulo/test/functional/CacheTestReader.java | 73 +++--
.../test/functional/MemoryStarvedScanIT.java | 2 +-
.../accumulo/test/functional/ReadWriteIT.java | 2 +-
.../apache/accumulo/test/functional/RestartIT.java | 7 +-
.../accumulo/test/functional/ZooCacheIT.java | 164 ++++++++++
.../apache/accumulo/test/zookeeper/ZooCacheIT.java | 175 ++++++-----
44 files changed, 995 insertions(+), 710 deletions(-)
diff --git a/.github/workflows/maven.yaml b/.github/workflows/maven.yaml
index ad7c810418..7bdcbc91cc 100644
--- a/.github/workflows/maven.yaml
+++ b/.github/workflows/maven.yaml
@@ -63,7 +63,7 @@ jobs:
profile:
- {name: 'unit-tests', javaver: 17, args: 'verify -PskipQA -DskipTests=false'}
- {name: 'qa-checks', javaver: 17, args: 'verify javadoc:jar -Psec-bugs -DskipTests -Dspotbugs.timeout=3600000'}
- - {name: 'compat', javaver: 17, args: 'package -DskipTests -Dversion.hadoop=3.0.3 -Dversion.zookeeper=3.5.10'}
+ - {name: 'compat', javaver: 17, args: 'package -DskipTests -Dversion.hadoop=3.0.3 -Dversion.zookeeper=3.6.4'}
- {name: 'errorprone', javaver: 17, args: 'verify -Perrorprone,skipQA'}
fail-fast: false
runs-on: ubuntu-latest
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
index 34871de647..c8db8b5c1f 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
@@ -32,6 +32,7 @@ import java.nio.file.Path;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -48,6 +49,7 @@ import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
@@ -76,8 +78,6 @@ import org.apache.accumulo.core.data.InstanceId;
import org.apache.accumulo.core.data.KeyValue;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
@@ -86,6 +86,7 @@ import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.AmpleImpl;
import org.apache.accumulo.core.rpc.SaslConnectionParams;
@@ -102,6 +103,8 @@ import org.apache.accumulo.core.util.cache.Caches;
import org.apache.accumulo.core.util.tables.TableZooHelper;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
@@ -159,6 +162,7 @@ public class ClientContext implements AccumuloClient {
private Caches caches;
private final AtomicBoolean zooKeeperOpened = new AtomicBoolean(false);
+ private final AtomicBoolean zooCacheCreated = new AtomicBoolean(false);
private final Supplier<ZooSession> zooSession;
private void ensureOpen() {
@@ -235,7 +239,12 @@ public class ClientContext implements AccumuloClient {
return zk;
});
- this.zooCache = memoize(() -> new ZooCache(getZooSession()));
+ this.zooCache = memoize(() -> {
+ var zc = new ZooCache(getZooSession(),
+ createPersistentWatcherPaths(ZooUtil.getRoot(getInstanceID())));
+ zooCacheCreated.set(true);
+ return zc;
+ });
this.accumuloConf = serverConf;
timeoutSupplier = memoizeWithExpiration(
() -> getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT), 100, MILLISECONDS);
@@ -784,6 +793,9 @@ public class ClientContext implements AccumuloClient {
@Override
public synchronized void close() {
if (closed.compareAndSet(false, true)) {
+ if (zooCacheCreated.get()) {
+ zooCache.get().close();
+ }
if (zooKeeperOpened.get()) {
zooSession.get().close();
}
@@ -1063,7 +1075,9 @@ public class ClientContext implements AccumuloClient {
// because that client could be closed, and its ZooSession also closed
// this needs to be fixed; TODO https://github.com/apache/accumulo/issues/2301
var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName(), "").get();
- this.zkLockChecker = new ZookeeperLockChecker(new ZooCache(zk), getZooKeeperRoot());
+ String zkRoot = getZooKeeperRoot();
+ this.zkLockChecker =
+ new ZookeeperLockChecker(new ZooCache(zk, Set.of(zkRoot + Constants.ZTSERVERS)), zkRoot);
}
return this.zkLockChecker;
}
@@ -1077,4 +1091,15 @@ public class ClientContext implements AccumuloClient {
return namespaces;
}
+ private static Set<String> createPersistentWatcherPaths(String zkRoot) {
+ Set<String> pathsToWatch = new HashSet<>();
+ for (String path : Set.of(Constants.ZCOMPACTORS, Constants.ZDEADTSERVERS, Constants.ZGC_LOCK,
+ Constants.ZMANAGER_LOCK, Constants.ZMINI_LOCK, Constants.ZMONITOR_LOCK,
+ Constants.ZNAMESPACES, Constants.ZRECOVERY, Constants.ZSSERVERS, Constants.ZTABLES,
+ Constants.ZTSERVERS, Constants.ZUSERS, RootTable.ZROOT_TABLET, Constants.ZTEST_LOCK)) {
+ pathsToWatch.add(zkRoot + path);
+ }
+ return pathsToWatch;
+ }
+
}
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceMapping.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceMapping.java
index 9d9073fdac..3b68fff26c 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceMapping.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceMapping.java
@@ -32,8 +32,9 @@ import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
import org.apache.accumulo.core.data.NamespaceId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.zookeeper.KeeperException;
import com.google.common.collect.ImmutableSortedMap;
@@ -147,7 +148,7 @@ public class NamespaceMapping {
private synchronized void update() {
final ZooCache zc = context.getZooCache();
final String zPath = context.getZooKeeperRoot() + Constants.ZNAMESPACES;
- final ZooCache.ZcStat stat = new ZooCache.ZcStat();
+ final ZcStat stat = new ZcStat();
byte[] data = zc.get(zPath, stat);
if (stat.getMzxid() > lastMzxid) {
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
index 314a82788a..8ae286c79e 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
@@ -24,11 +24,11 @@ import java.util.Set;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.clientImpl.ClientTabletCacheImpl.TabletServerLockChecker;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import com.google.common.net.HostAndPort;
diff --git a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java
index 18aeda770a..5df974a3af 100644
--- a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java
+++ b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java
@@ -26,8 +26,6 @@ import java.util.List;
import java.util.Optional;
import java.util.UUID;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.LockID;
@@ -35,6 +33,8 @@ import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.UuidUtil;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
@@ -670,8 +670,8 @@ public class ServiceLock implements Watcher {
return ServiceLockData.parse(data);
}
- public static Optional<ServiceLockData> getLockData(
- org.apache.accumulo.core.fate.zookeeper.ZooCache zc, ServiceLockPath path, ZcStat stat) {
+ public static Optional<ServiceLockData> getLockData(ZooCache zc, ServiceLockPath path,
+ ZcStat stat) {
List<String> children = validateAndSort(path, zc.getChildren(path.toString()));
diff --git a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
index 8fd3f361dc..7e4378ceec 100644
--- a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
+++ b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
@@ -34,10 +34,10 @@ import java.util.concurrent.FutureTask;
import java.util.function.Predicate;
import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.util.threads.ThreadPoolNames;
import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import com.google.common.base.Preconditions;
import com.google.common.net.HostAndPort;
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
index 2b59f78431..0efebce8db 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
@@ -50,7 +50,6 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.fate.FateId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
@@ -76,6 +75,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Su
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.UserCompactionRequestedColumnFamily;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.accumulo.core.zookeeper.ZcStat;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.apache.hadoop.io.Text;
@@ -663,7 +663,7 @@ public class TabletMetadata {
*/
private static Optional<TServerInstance> checkTabletServer(ClientContext context,
ServiceLockPath slp) {
- ZooCache.ZcStat stat = new ZooCache.ZcStat();
+ ZcStat stat = new ZcStat();
log.trace("Checking server at ZK path: {}", slp);
return ServiceLock.getLockData(context.getZooCache(), slp, stat)
.map(sld -> sld.getAddress(ServiceLockData.ThriftService.TSERV))
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
index 45ab3c86fa..e84a7beb34 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
@@ -36,7 +36,6 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.AccumuloServerException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
@@ -45,6 +44,7 @@ import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes.Exec;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes.ExecVoid;
import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
import org.apache.thrift.TServiceClient;
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
index 3d5fc44576..092f51b04e 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
@@ -37,7 +37,6 @@ import java.util.concurrent.Future;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.compaction.thrift.CompactorService;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
@@ -50,6 +49,7 @@ import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.core.zookeeper.ZcStat;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/core/src/main/java/org/apache/accumulo/core/util/tables/TableMap.java b/core/src/main/java/org/apache/accumulo/core/util/tables/TableMap.java
index 859d82dbc7..0fa46b7d41 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/tables/TableMap.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/tables/TableMap.java
@@ -31,7 +31,7 @@ import org.apache.accumulo.core.clientImpl.Namespace;
import org.apache.accumulo.core.clientImpl.Namespaces;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java b/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java
index 1ab1e1d708..410ddd40d1 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java
@@ -34,10 +34,10 @@ import org.apache.accumulo.core.clientImpl.Namespace;
import org.apache.accumulo.core.clientImpl.Namespaces;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.metadata.AccumuloTable;
import org.apache.accumulo.core.util.cache.Caches.CacheName;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import com.github.benmanes.caffeine.cache.Cache;
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZcNode.java b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZcNode.java
similarity index 96%
rename from core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZcNode.java
rename to core/src/main/java/org/apache/accumulo/core/zookeeper/ZcNode.java
index d4ed3c955b..78f33c9068 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZcNode.java
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZcNode.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.core.fate.zookeeper;
+package org.apache.accumulo.core.zookeeper;
import java.util.List;
import java.util.Objects;
@@ -48,7 +48,7 @@ import com.google.common.base.Preconditions;
class ZcNode {
private final byte[] data;
- private final ZooCache.ZcStat stat;
+ private final ZcStat stat;
private final List<String> children;
static final ZcNode NON_EXISTENT = new ZcNode();
@@ -82,7 +82,7 @@ class ZcNode {
* Creates a new ZcNode that combines the children from an existing ZcNode and sets the data and
* stat.
*/
- ZcNode(byte[] data, ZooCache.ZcStat zstat, ZcNode existing) {
+ ZcNode(byte[] data, ZcStat zstat, ZcNode existing) {
this.data = Objects.requireNonNull(data);
this.stat = Objects.requireNonNull(zstat);
if (existing == null) {
@@ -99,6 +99,7 @@ class ZcNode {
*/
byte[] getData() {
Preconditions.checkState(cachedData());
+ ;
accessCount.incrementAndGet();
return data;
}
@@ -108,7 +109,7 @@ class ZcNode {
* exist
* @throws IllegalStateException in the case where the node exists and the data was never set
*/
- ZooCache.ZcStat getStat() {
+ ZcStat getStat() {
Preconditions.checkState(cachedData());
return stat;
}
diff --git a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZcStat.java b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZcStat.java
new file mode 100644
index 0000000000..a641d42a0a
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZcStat.java
@@ -0,0 +1,53 @@
+/*
+ * 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 org.apache.zookeeper.data.Stat;
+
+import com.google.common.annotations.VisibleForTesting;
+
+public class ZcStat {
+ private long ephemeralOwner;
+ private long mzxid;
+
+ public ZcStat() {}
+
+ public ZcStat(Stat stat) {
+ this.ephemeralOwner = stat.getEphemeralOwner();
+ this.mzxid = stat.getMzxid();
+ }
+
+ public long getEphemeralOwner() {
+ return ephemeralOwner;
+ }
+
+ public void set(ZcStat cachedStat) {
+ this.ephemeralOwner = cachedStat.ephemeralOwner;
+ this.mzxid = cachedStat.mzxid;
+ }
+
+ @VisibleForTesting
+ public void setEphemeralOwner(long ephemeralOwner) {
+ this.ephemeralOwner = ephemeralOwner;
+ }
+
+ public long getMzxid() {
+ return mzxid;
+ }
+}
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZooCache.java b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
similarity index 61%
rename from core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZooCache.java
rename to core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
index 6e2d334db7..e5f52a32d2 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZooCache.java
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
@@ -16,19 +16,22 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.core.fate.zookeeper;
+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.ConcurrentModificationException;
+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.concurrent.locks.LockSupport;
import java.util.function.Consumer;
import java.util.function.Predicate;
@@ -36,7 +39,6 @@ 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.accumulo.core.zookeeper.ZooSession;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
@@ -46,7 +48,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.Ticker;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -59,54 +61,31 @@ public class ZooCache {
private static final Logger log = LoggerFactory.getLogger(ZooCache.class);
- private final ZCacheWatcher watcher = new ZCacheWatcher();
- private final Optional<ZooCacheWatcher> externalWatcher;
+ private final NavigableSet<String> watchedPaths;
+
+ // visible for tests
+ protected 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();
- // The concurrent map returned by Caffiene 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.
+ 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;
- public static class ZcStat {
- private long ephemeralOwner;
- private long mzxid;
-
- public ZcStat() {}
-
- private ZcStat(Stat stat) {
- this.ephemeralOwner = stat.getEphemeralOwner();
- this.mzxid = stat.getMzxid();
- }
+ private final AtomicLong updateCount = new AtomicLong();
- public long getEphemeralOwner() {
- return ephemeralOwner;
- }
+ private final AtomicLong zkClientTracker = new AtomicLong();
- private void set(ZcStat cachedStat) {
- this.ephemeralOwner = cachedStat.ephemeralOwner;
- this.mzxid = cachedStat.mzxid;
- }
-
- @VisibleForTesting
- public void setEphemeralOwner(long ephemeralOwner) {
- this.ephemeralOwner = ephemeralOwner;
- }
-
- public long getMzxid() {
- return mzxid;
- }
- }
-
- private final AtomicLong updateCount = new AtomicLong(0);
-
- private class ZCacheWatcher implements Watcher {
+ class ZCacheWatcher implements Watcher {
@Override
public void process(WatchedEvent event) {
if (log.isTraceEnabled()) {
@@ -114,22 +93,40 @@ public class ZooCache {
}
switch (event.getType()) {
- case NodeDataChanged:
case NodeChildrenChanged:
- case NodeCreated:
- case NodeDeleted:
+ // 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:
- remove(event.getPath());
+ // 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:
- // Closed is a new event in ZK 3.5 generated client-side when the ZK client is closed
- // These are ignored, because they are generated by SingletonManager closing
- // ZooKeepers for ZooSession, and SingletonManager is already responsible for clearing
- // caches via the registered ZooCacheFactory singleton
- log.trace("{} ZooKeeper connection closed, ignoring; {}", cacheId, event);
+ log.trace("{} ZooKeeper connection closed, clearing cache; {}", cacheId, event);
+ clear();
break;
case Disconnected:
log.trace("{} ZooKeeper connection disconnected, clearing cache; {}", cacheId, event);
@@ -143,65 +140,117 @@ public class ZooCache {
clear();
break;
default:
- log.warn("{} Unhandled {}", cacheId, event);
+ log.warn("{} Unhandled state {}", cacheId, event);
break;
}
break;
default:
- log.warn("{} Unhandled {}", cacheId, event);
+ log.warn("{} Unhandled event type {}", cacheId, event);
break;
}
- externalWatcher.ifPresent(w -> w.accept(event));
+ 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 new cache without an external watcher.
+ * 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 the ZooKeeper instance
- * @throws NullPointerException if zk is {@code null}
+ * @param zk ZooSession for this instance
+ * @param pathsToWatch Paths in ZooKeeper to watch
*/
- public ZooCache(ZooSession zk) {
- this(zk, Optional.empty(), Duration.ofMinutes(3));
+ 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.zkClientTracker.set(this.getZKClientObjectVersion());
+ 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() {
+ return zk.getConnectionCounter();
}
/**
- * Creates a new cache. The given watcher is called whenever a watched node changes.
- *
- * @param zk the ZooKeeper instance
- * @param watcher watcher object
- * @throws NullPointerException if zk or watcher is {@code null}
+ * @return true if ZK has changed; false otherwise
*/
- public ZooCache(ZooSession zk, ZooCacheWatcher watcher) {
- this(zk, Optional.of(watcher), Duration.ofMinutes(3));
+ private boolean handleZKConnectionChange() {
+ final long currentCount = getZKClientObjectVersion();
+ final long oldCount = zkClientTracker.get();
+ if (oldCount != currentCount) {
+ if (zkClientTracker.compareAndSet(oldCount, currentCount)) {
+ setupWatchers();
+ }
+ return true;
+ }
+ return false;
}
- public ZooCache(ZooSession zk, Optional<ZooCacheWatcher> watcher, Duration timeout) {
- this.zk = requireNonNull(zk);
- this.externalWatcher = watcher;
- RemovalListener<String,ZcNode> removalListerner = (path, zcNode, reason) -> {
- try {
- log.trace("{} removing watches for {} because {} accesses {}", cacheId, path, reason,
- zcNode == null ? -1 : zcNode.getAccessCount());
- zk.removeWatches(path, ZooCache.this.watcher, Watcher.WatcherType.Any, false);
- } catch (InterruptedException | KeeperException | RuntimeException e) {
- log.warn("{} failed to remove watches on path {} in zookeeper", cacheId, path, e);
+ // Called on construction and when ZooKeeper connection changes
+ synchronized void setupWatchers() {
+
+ for (String left : watchedPaths) {
+ for (String right : watchedPaths) {
+ if (!left.equals(right) && left.contains(right)) {
+ throw new IllegalArgumentException(
+ "Overlapping paths found in paths to watch. left: " + left + ", right: " + right);
+ }
}
- };
- // Must register the removal listener using evictionListener inorder for removal to be mutually
- // exclusive with any other operations on the same path. This is important for watcher
- // consistency, concurrently adding and removing watches for the same path would leave zoocache
- // in a really bad state. The cache builder has another way to register a removal listener that
- // is not mutually exclusive.
- Cache<String,ZcNode> cache =
- Caches.getInstance().createNewBuilder(Caches.CacheName.ZOO_CACHE, false)
- .expireAfterAccess(timeout).evictionListener(removalListerner).build();
- nodeCache = cache.asMap();
- log.trace("{} created new cache", cacheId, new Exception());
+ }
+
+ try {
+ zk.addPersistentRecursiveWatchers(watchedPaths, watcher);
+ clear();
+ log.trace("{} Reinitialized persistent watchers and cleared cache {}", cacheId,
+ zkClientTracker.get());
+ } catch (KeeperException | InterruptedException e) {
+ 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 abstract static class ZooRunnable<T> {
+ 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.
@@ -231,7 +280,11 @@ public class ZooCache {
while (true) {
try {
- return run();
+ T result = run();
+ if (handleZKConnectionChange()) {
+ continue;
+ }
+ return result;
} catch (KeeperException | ZcException e) {
KeeperException ke;
if (e instanceof ZcException) {
@@ -251,8 +304,6 @@ public class ZooCache {
}
} catch (InterruptedException | ZcInterruptedException e) {
log.info("Zookeeper error, will retry", e);
- } catch (ConcurrentModificationException e) {
- log.debug("Zookeeper was modified, will retry");
}
try {
@@ -261,7 +312,6 @@ public class ZooCache {
} catch (InterruptedException e) {
log.debug("Wait in retry() was interrupted.", e);
}
- LockSupport.parkNanos(sleepTime);
if (sleepTime < 10_000) {
sleepTime = (int) (sleepTime + sleepTime * RANDOM.get().nextDouble());
}
@@ -291,14 +341,14 @@ public class ZooCache {
}
/**
- * Gets the children of the given node. A watch is established by this call.
+ * 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);
-
+ ensureWatched(zPath);
ZooRunnable<List<String>> zr = new ZooRunnable<>() {
@Override
@@ -317,23 +367,12 @@ public class ZooCache {
return zcn;
}
try {
- // Register a watcher on the node to monitor creation/deletion events for the node. It
- // is possible that an event from this watch could trigger prior to calling getChildren.
- // That is ok because the compute() call on the map has a lock and processing the event
- // will block until compute() returns. After compute() returns the event processing
- // would clear the map entry.
- Stat stat = zk.exists(zPath, watcher);
- if (stat == null) {
- log.trace("{} getChildren saw that {} does not exists", cacheId, zPath);
- return ZcNode.NON_EXISTENT;
- }
- List<String> children = zk.getChildren(zPath, watcher);
+ 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("{} get children saw race condition for {}, node deleted after exists call",
- cacheId, zPath);
- throw new ConcurrentModificationException(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) {
@@ -350,8 +389,7 @@ public class ZooCache {
}
/**
- * Gets data at the given path. Status information is not returned. A watch is established by this
- * call.
+ * Gets data at the given path. Status information is not returned.
*
* @param zPath path to get
* @return path data, or null if non-existent
@@ -362,7 +400,7 @@ public class ZooCache {
/**
* Gets data at the given path, filling status information into the given <code>Stat</code>
- * object. A watch is established by this call.
+ * object.
*
* @param zPath path to get
* @param status status object to populate
@@ -370,6 +408,7 @@ public class ZooCache {
*/
public byte[] get(final String zPath, final ZcStat status) {
Preconditions.checkState(!closed);
+ ensureWatched(zPath);
ZooRunnable<byte[]> zr = new ZooRunnable<>() {
@Override
@@ -391,44 +430,27 @@ public class ZooCache {
if (zcn != null && zcn.cachedData()) {
return zcn;
}
- /*
- * The following call to exists() is important, since we are caching that a node does not
- * exist. Once the node comes into existence, it will be added to the cache. But this
- * notification of a node coming into existence will only be given if exists() was
- * previously called. If the call to exists() is bypassed and only getData() is called
- * with a special case that looks for Code.NONODE in the KeeperException, then
- * non-existence can not be cached.
- */
try {
- Stat stat = zk.exists(zPath, watcher);
- if (stat == null) {
- if (log.isTraceEnabled()) {
- log.trace("{} zookeeper did not contain {}", cacheId, zPath);
- }
+ 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;
- } else {
- byte[] data = null;
- ZcStat zstat = null;
- try {
- data = zk.getData(zPath, watcher, stat);
- zstat = new ZcStat(stat);
- } catch (KeeperException.BadVersionException | KeeperException.NoNodeException e1) {
- throw new ConcurrentModificationException(e1);
- } catch (InterruptedException e) {
- 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 (InterruptedException e) {
+ 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);
- } catch (InterruptedException e) {
- throw new ZcInterruptedException(e);
}
-
});
// update this after the compute call completes when the change is visible
@@ -456,16 +478,10 @@ public class ZooCache {
}
}
- private void remove(String zPath) {
- nodeCache.remove(zPath);
- log.trace("{} removed {} from cache", cacheId, zPath);
- updateCount.incrementAndGet();
- }
-
/**
* Clears this cache.
*/
- public void clear() {
+ protected void clear() {
Preconditions.checkState(!closed);
nodeCache.clear();
updateCount.incrementAndGet();
@@ -473,6 +489,7 @@ public class ZooCache {
}
public void close() {
+ clear();
closed = true;
}
@@ -493,6 +510,7 @@ public class ZooCache {
*/
@VisibleForTesting
public boolean dataCached(String zPath) {
+ ensureWatched(zPath);
var zcn = nodeCache.get(zPath);
return zcn != null && zcn.cachedData();
}
@@ -505,6 +523,7 @@ public class ZooCache {
*/
@VisibleForTesting
public boolean childrenCached(String zPath) {
+ ensureWatched(zPath);
var zcn = nodeCache.get(zPath);
return zcn != null && zcn.cachedChildren();
}
@@ -514,18 +533,15 @@ public class ZooCache {
*/
public void clear(Predicate<String> pathPredicate) {
Preconditions.checkState(!closed);
-
- Predicate<String> pathPredicateToUse;
- if (log.isTraceEnabled()) {
- pathPredicateToUse = pathPredicate.and(path -> {
- log.trace("removing {} from cache", path);
- return true;
- });
- } else {
- pathPredicateToUse = pathPredicate;
- }
- nodeCache.keySet().removeIf(pathPredicateToUse);
- updateCount.incrementAndGet();
+ 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);
}
/**
@@ -534,10 +550,15 @@ public class ZooCache {
* @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();
diff --git a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooSession.java b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooSession.java
index 032eecc453..850d6481d2 100644
--- a/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooSession.java
+++ b/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooSession.java
@@ -25,7 +25,10 @@ import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
import java.io.IOException;
import java.net.UnknownHostException;
+import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -37,6 +40,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooReader;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.zookeeper.AddWatchMode;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.CreateMode;
@@ -290,6 +294,33 @@ public class ZooSession implements AutoCloseable {
verifyConnected().sync(path, cb, ctx);
}
+ public void addPersistentRecursiveWatchers(Set<String> paths, Watcher watcher)
+ throws KeeperException, InterruptedException {
+
+ ZooKeeper localZK = verifyConnected();
+ Set<String> remainingPaths = new HashSet<>(paths);
+ while (true) {
+ try {
+ Iterator<String> remainingPathsIter = remainingPaths.iterator();
+ while (remainingPathsIter.hasNext()) {
+ String path = remainingPathsIter.next();
+ localZK.addWatch(path, watcher, AddWatchMode.PERSISTENT_RECURSIVE);
+ remainingPathsIter.remove();
+ }
+ break;
+ } catch (KeeperException e) {
+ log.error("Error setting persistent watcher in ZooKeeper, retrying...", e);
+ ZooKeeper currentZK = verifyConnected();
+ // If ZooKeeper object is different, then reset the localZK variable
+ // and start over.
+ if (localZK != currentZK) {
+ localZK = currentZK;
+ remainingPaths = new HashSet<>(paths);
+ }
+ }
+ }
+ }
+
@Override
public void close() {
if (closed.compareAndSet(false, true)) {
@@ -309,4 +340,15 @@ public class ZooSession implements AutoCloseable {
return zrw;
}
+ /**
+ * Connection counter is incremented internal when ZooSession creates a new ZooKeeper client.
+ * Clients of ZooSession can use this counter as a way to determine if a new ZooKeeper connection
+ * has been created.
+ *
+ * @return connection counter
+ */
+ public long getConnectionCounter() {
+ return connectCounter.get();
+ }
+
}
diff --git a/core/src/test/java/org/apache/accumulo/core/clientImpl/ZookeeperLockCheckerTest.java b/core/src/test/java/org/apache/accumulo/core/clientImpl/ZookeeperLockCheckerTest.java
index bb7402f78d..7e0470aae8 100644
--- a/core/src/test/java/org/apache/accumulo/core/clientImpl/ZookeeperLockCheckerTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/clientImpl/ZookeeperLockCheckerTest.java
@@ -30,8 +30,8 @@ import java.util.UUID;
import java.util.function.Predicate;
import org.apache.accumulo.core.data.InstanceId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
diff --git a/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java b/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
index 0232376237..c58a6a363a 100644
--- a/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
@@ -45,11 +45,11 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.easymock.EasyMock;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
diff --git a/core/src/test/java/org/apache/accumulo/core/fate/zookeeper/ZooCacheTest.java b/core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
similarity index 58%
rename from core/src/test/java/org/apache/accumulo/core/fate/zookeeper/ZooCacheTest.java
rename to core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
index 6d091d3aae..e4e9426c64 100644
--- a/core/src/test/java/org/apache/accumulo/core/fate/zookeeper/ZooCacheTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/zookeeper/ZooCacheTest.java
@@ -16,13 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.core.fate.zookeeper;
+package org.apache.accumulo.core.zookeeper;
-import static org.easymock.EasyMock.anyObject;
-import static org.easymock.EasyMock.capture;
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;
@@ -30,72 +30,126 @@ 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.fate.zookeeper.ZooCache.ZcStat;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZooCacheWatcher;
-import org.apache.accumulo.core.zookeeper.ZooSession;
+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.Capture;
import org.easymock.EasyMock;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
public class ZooCacheTest {
- private static final String ZPATH = "/some/path/in/zk";
+
+ /**
+ * 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 void setupWatchers() {
+ clear();
+ }
+
+ 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() + Constants.ZTSERVERS;
+ 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 ZooCache zc;
+ private TestZooCache zc;
@BeforeEach
public void setUp() {
zk = createStrictMock(ZooSession.class);
- zc = new ZooCache(zk);
+ zc = new TestZooCache(zk, Set.of(root));
}
@Test
- public void testGet() throws Exception {
- testGet(false);
+ @SuppressWarnings("unchecked")
+ public void testOverlappingPaths() throws Exception {
+ expect(zk.getConnectionCounter()).andReturn(2L).times(2);
+ zk.addPersistentRecursiveWatchers(isA(Set.class), isA(Watcher.class));
+ replay(zk);
+ assertThrows(IllegalArgumentException.class,
+ () -> new ZooCache(zk, Set.of(root, root + "/localhost:9995")));
+
+ Set<String> goodPaths = Set.of("/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/compactors",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/dead/tservers",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/gc/lock",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/managers/lock",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/namespaces",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/recovery",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/root_tablet",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/sservers",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/tables",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/tservers",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/users",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/mini",
+ "/accumulo/8247eee6-a176-4e19-baf7-e3da965fe050/monitor/lock");
+ new ZooCache(zk, goodPaths);
+ verify(zk);
}
@Test
- public void testGet_FillStat() throws Exception {
- testGet(true);
+ @SuppressWarnings("unchecked")
+ public void testUnwatchedPaths() throws Exception {
+ expect(zk.getConnectionCounter()).andReturn(2L).anyTimes();
+ zk.addPersistentRecursiveWatchers(isA(Set.class), isA(Watcher.class));
+ 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);
}
- private void testGet(boolean fillStat) throws Exception {
- ZcStat myStat = null;
- if (fillStat) {
- myStat = new ZcStat();
- }
- final long ephemeralOwner = 123456789L;
- Stat existsStat = new Stat();
- existsStat.setEphemeralOwner(ephemeralOwner);
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat))).andReturn(DATA);
+ @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));
- assertArrayEquals(DATA, (fillStat ? zc.get(ZPATH, myStat) : zc.get(ZPATH)));
+ ZcStat zcStat = new ZcStat();
+ assertSame(DATA, zc.get(ZPATH, zcStat));
+ assertEquals(123456789, zcStat.getEphemeralOwner());
+ assertSame(DATA, zc.get(ZPATH));
verify(zk);
- if (fillStat) {
- assertEquals(ephemeralOwner, myStat.getEphemeralOwner());
- }
-
- assertTrue(zc.dataCached(ZPATH));
- assertSame(DATA, zc.get(ZPATH)); // cache hit
}
@Test
public void testGet_NonExistent() throws Exception {
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(null);
+ expect(zk.getData(ZPATH, null, new Stat()))
+ .andThrow(new KeeperException.NoNodeException(ZPATH));
replay(zk);
assertNull(zc.get(ZPATH));
@@ -112,66 +166,32 @@ public class ZooCacheTest {
testGet_Retry(new KeeperException.ConnectionLossException());
}
- @Test
- public void testGet_Retry_BadVersion() throws Exception {
- testGet_Retry(new KeeperException.BadVersionException(ZPATH));
- }
-
@Test
public void testGet_Retry_Interrupted() throws Exception {
testGet_Retry(new InterruptedException());
}
private void testGet_Retry(Exception e) throws Exception {
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andThrow(e);
Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat))).andReturn(DATA);
- replay(zk);
-
- assertArrayEquals(DATA, zc.get(ZPATH));
- verify(zk);
- }
-
- @Test
- public void testGet_Retry2_NoNode() throws Exception {
- testGet_Retry2(new KeeperException.NoNodeException(ZPATH));
- }
-
- @Test
- public void testGet_Retry2_ConnectionLoss() throws Exception {
- testGet_Retry2(new KeeperException.ConnectionLossException());
- }
-
- @Test
- public void testGet_Retry2_BadVersion() throws Exception {
- testGet_Retry2(new KeeperException.BadVersionException(ZPATH));
- }
-
- @Test
- public void testGet_Retry2_Interrupted() throws Exception {
- testGet_Retry2(new InterruptedException());
- }
-
- private void testGet_Retry2(Exception e) throws Exception {
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat))).andThrow(e);
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat))).andReturn(DATA);
+ expect(zk.getData(ZPATH, null, existsStat)).andThrow(e);
+ if (!(e instanceof KeeperException.NoNodeException)) {
+ // Retry will not happen on NoNodeException or BadVersionException
+ expect(zk.getData(ZPATH, null, existsStat)).andReturn(DATA);
+ }
replay(zk);
- assertArrayEquals(DATA, zc.get(ZPATH));
+ if (e instanceof KeeperException.NoNodeException
+ || e instanceof KeeperException.BadVersionException) {
+ assertNull(zc.get(ZPATH));
+ } else {
+ assertArrayEquals(DATA, zc.get(ZPATH));
+ }
verify(zk);
}
- // ---
-
@Test
public void testGetChildren() throws Exception {
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class))).andReturn(CHILDREN);
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
replay(zk);
assertFalse(zc.childrenCached(ZPATH));
@@ -185,9 +205,7 @@ public class ZooCacheTest {
@Test
public void testGetChildren_NoKids() throws Exception {
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class))).andReturn(List.of());
+ expect(zk.getChildren(ZPATH, null)).andReturn(List.of());
replay(zk);
assertEquals(List.of(), zc.getChildren(ZPATH));
@@ -196,28 +214,10 @@ public class ZooCacheTest {
assertEquals(List.of(), zc.getChildren(ZPATH)); // cache hit
}
- @Test
- public void testGetChildren_RaceCondition() throws Exception {
- // simulate the node being deleted between calling zookeeper.exists and zookeeper.getChildren
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class)))
- .andThrow(new KeeperException.NoNodeException(ZPATH));
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(null);
- replay(zk);
- assertNull(zc.getChildren(ZPATH));
- verify(zk);
- assertNull(zc.getChildren(ZPATH));
- }
-
@Test
public void testGetChildren_Retry() throws Exception {
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class)))
- .andThrow(new KeeperException.BadVersionException(ZPATH));
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class))).andReturn(CHILDREN);
+ expect(zk.getChildren(ZPATH, null)).andThrow(new KeeperException.BadVersionException(ZPATH));
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
replay(zk);
assertEquals(CHILDREN, zc.getChildren(ZPATH));
@@ -229,7 +229,7 @@ public class ZooCacheTest {
public void testGetChildren_NoNode() throws Exception {
assertFalse(zc.childrenCached(ZPATH));
assertFalse(zc.dataCached(ZPATH));
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(null);
+ expect(zk.getChildren(ZPATH, null)).andThrow(new KeeperException.NoNodeException(ZPATH));
replay(zk);
assertNull(zc.getChildren(ZPATH));
@@ -287,29 +287,26 @@ public class ZooCacheTest {
WatchedEvent event =
new WatchedEvent(eventType, Watcher.Event.KeeperState.SyncConnected, ZPATH);
TestWatcher exw = new TestWatcher(event);
- zc = new ZooCache(zk, exw);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
- Watcher w = watchData(initialData);
- w.process(event);
+ watchData(initialData, stillCached);
+ zc.executeWatcher(event);
assertTrue(exw.wasCalled());
assertEquals(stillCached, zc.dataCached(ZPATH));
}
- private Watcher watchData(byte[] initialData) throws Exception {
- Capture<Watcher> cw = EasyMock.newCapture();
+ private void watchData(byte[] initialData, boolean cached) throws Exception {
Stat existsStat = new Stat();
if (initialData != null) {
- expect(zk.exists(eq(ZPATH), capture(cw))).andReturn(existsStat);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat)))
- .andReturn(initialData);
+ expect(zk.getData(ZPATH, null, existsStat)).andReturn(initialData);
} else {
- expect(zk.exists(eq(ZPATH), capture(cw))).andReturn(null);
+ expect(zk.getData(ZPATH, null, existsStat))
+ .andThrow(new KeeperException.NoNodeException(ZPATH));
}
replay(zk);
zc.get(ZPATH);
assertTrue(zc.dataCached(ZPATH));
-
- return cw.getValue();
}
@Test
@@ -338,24 +335,12 @@ public class ZooCacheTest {
var uc1 = zc.getUpdateCount();
- final long ephemeralOwner1 = 123456789L;
- Stat existsStat1 = new Stat();
- existsStat1.setEphemeralOwner(ephemeralOwner1);
-
- final long ephemeralOwner2 = 987654321L;
- Stat existsStat2 = new Stat();
- existsStat2.setEphemeralOwner(ephemeralOwner2);
-
if (getDataFirst) {
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat1);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat1))).andReturn(DATA);
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat2);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class))).andReturn(CHILDREN);
+ expect(zk.getData(ZPATH, null, new Stat())).andReturn(DATA);
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
} else {
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat2);
- expect(zk.getChildren(eq(ZPATH), anyObject(Watcher.class))).andReturn(CHILDREN);
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat1);
- expect(zk.getData(eq(ZPATH), anyObject(Watcher.class), eq(existsStat1))).andReturn(DATA);
+ expect(zk.getChildren(ZPATH, null)).andReturn(CHILDREN);
+ expect(zk.getData(ZPATH, null, new Stat())).andReturn(DATA);
}
replay(zk);
@@ -363,7 +348,6 @@ public class ZooCacheTest {
if (getDataFirst) {
var zcStat = new ZcStat();
var data = zc.get(ZPATH, zcStat);
- assertEquals(ephemeralOwner1, zcStat.getEphemeralOwner());
assertArrayEquals(DATA, data);
} else {
var children = zc.getChildren(ZPATH);
@@ -378,7 +362,6 @@ public class ZooCacheTest {
} else {
var zcStat = new ZcStat();
var data = zc.get(ZPATH, zcStat);
- assertEquals(ephemeralOwner1, zcStat.getEphemeralOwner());
assertArrayEquals(DATA, data);
}
var uc3 = zc.getUpdateCount();
@@ -390,7 +373,6 @@ public class ZooCacheTest {
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
- assertEquals(ephemeralOwner1, zcStat.getEphemeralOwner());
assertArrayEquals(DATA, data);
var children = zc.getChildren(ZPATH);
assertEquals(CHILDREN, children);
@@ -401,11 +383,12 @@ public class ZooCacheTest {
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 ZooCache(zk, exw);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
- Watcher w = watchData(DATA);
+ watchData(DATA, true);
assertTrue(zc.dataCached(ZPATH));
- w.process(event);
+ zc.executeWatcher(event);
assertTrue(exw.wasCalled());
assertFalse(zc.dataCached(ZPATH));
}
@@ -417,7 +400,7 @@ public class ZooCacheTest {
@Test
public void testWatchChildrenNode_ChildrenChanged() throws Exception {
- testWatchChildrenNode(CHILDREN, Watcher.Event.EventType.NodeChildrenChanged, false);
+ testWatchChildrenNode(CHILDREN, Watcher.Event.EventType.NodeChildrenChanged, true);
}
@Test
@@ -435,27 +418,23 @@ public class ZooCacheTest {
WatchedEvent event =
new WatchedEvent(eventType, Watcher.Event.KeeperState.SyncConnected, ZPATH);
TestWatcher exw = new TestWatcher(event);
- zc = new ZooCache(zk, exw);
+ zc = new TestZooCache(zk, Set.of(root));
+ zc.addZooCacheWatcher(exw);
- Watcher w = watchChildren(initialChildren);
- w.process(event);
+ watchChildren(initialChildren);
+ zc.executeWatcher(event);
assertTrue(exw.wasCalled());
assertEquals(stillCached, zc.childrenCached(ZPATH));
}
- private Watcher watchChildren(List<String> initialChildren) throws Exception {
- Capture<Watcher> cw = EasyMock.newCapture();
+ private void watchChildren(List<String> initialChildren) throws Exception {
if (initialChildren == null) {
- expect(zk.exists(eq(ZPATH), capture(cw))).andReturn(null);
+ expect(zk.getChildren(ZPATH, null)).andReturn(List.of());
} else {
- Stat existsStat = new Stat();
- expect(zk.exists(eq(ZPATH), anyObject(Watcher.class))).andReturn(existsStat);
- expect(zk.getChildren(eq(ZPATH), capture(cw))).andReturn(initialChildren);
+ expect(zk.getChildren(ZPATH, null)).andReturn(initialChildren);
}
replay(zk);
zc.getChildren(ZPATH);
assertTrue(zc.childrenCached(ZPATH));
-
- return cw.getValue();
}
}
diff --git a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
index 4eddb61c9e..a39a38250d 100644
--- a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
+++ b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
@@ -57,7 +57,6 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
-import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -993,8 +992,7 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
control.stop(ServerType.COMPACTOR, null);
control.stop(ServerType.SCAN_SERVER, null);
- // The method calls above kill the server
- // Clean up the locks in ZooKeeper fo that if the cluster
+ // Clean up the locks in ZooKeeper so that if the cluster
// is restarted, then the processes will start right away
// and not wait for the old locks to be cleaned up.
try {
@@ -1003,31 +1001,21 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
} catch (RuntimeException e) {
log.error("Error zapping zookeeper locks", e);
}
- control.stop(ServerType.ZOOKEEPER, null);
// Clear the location of the servers in ZooCache.
- // When ZooKeeper was stopped in the previous method call,
- // the local ZooKeeper watcher did not fire. If MAC is
- // restarted, then ZooKeeper will start on the same port with
- // the same data, but no Watchers will fire.
- boolean startCalled = true;
+ boolean macStarted = false;
try {
getServerContext().getZooKeeperRoot();
+ macStarted = true;
} catch (IllegalStateException e) {
- if (e.getMessage().startsWith("Accumulo not initialized")) {
- startCalled = false;
+ if (!e.getMessage().startsWith("Accumulo not initialized")) {
+ throw e;
}
}
- if (startCalled) {
- final ServerContext ctx = getServerContext();
- final String zRoot = ctx.getZooKeeperRoot();
- Predicate<String> pred = path -> false;
- for (String lockPath : Set.of(Constants.ZMANAGER_LOCK, Constants.ZGC_LOCK,
- Constants.ZCOMPACTORS, Constants.ZSSERVERS, Constants.ZTSERVERS)) {
- pred = pred.or(path -> path.startsWith(zRoot + lockPath));
- }
- ctx.getZooCache().clear(pred);
+ if (macStarted) {
+ getServerContext().getZooCache().clear(path -> path.startsWith("/"));
}
+ control.stop(ServerType.ZOOKEEPER, null);
// ACCUMULO-2985 stop the ExecutorService after we finished using it to stop accumulo procs
if (executor != null) {
diff --git a/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java b/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
index ab593585b0..82689c0ee2 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
@@ -31,15 +31,11 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZooCacheWatcher;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockPaths;
@@ -57,7 +53,10 @@ import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.Halt;
import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache.ZooCacheWatcher;
import org.apache.accumulo.server.ServerContext;
+import org.apache.commons.lang3.StringUtils;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransport;
import org.apache.zookeeper.KeeperException;
@@ -67,7 +66,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Suppliers;
import com.google.common.net.HostAndPort;
public class LiveTServerSet implements ZooCacheWatcher {
@@ -213,16 +211,10 @@ public class LiveTServerSet implements ZooCacheWatcher {
// The set of entries in zookeeper without locks, and the first time each was noticed
private final Map<ServiceLockPath,Long> locklessServers = new HashMap<>();
- private final Supplier<ZooCache> zcSupplier;
-
public LiveTServerSet(ServerContext context, Listener cback) {
this.cback = cback;
this.context = context;
- this.zcSupplier = Suppliers.memoize(() -> new ZooCache(context.getZooSession(), this));
- }
-
- public ZooCache getZooCache() {
- return zcSupplier.get();
+ this.context.getZooCache().addZooCacheWatcher(this);
}
public synchronized void startListeningForTabletServerChanges() {
@@ -285,23 +277,31 @@ public class LiveTServerSet implements ZooCacheWatcher {
final TServerInfo info = current.get(tserverPath.getServer());
ZcStat stat = new ZcStat();
- Optional<ServiceLockData> sld = ServiceLock.getLockData(getZooCache(), tserverPath, stat);
+ Optional<ServiceLockData> sld =
+ ServiceLock.getLockData(context.getZooCache(), tserverPath, stat);
if (sld.isEmpty()) {
+ log.trace("lock does not exist for server: {}", tserverPath.getServer());
if (info != null) {
doomed.add(info.instance);
current.remove(tserverPath.getServer());
serversShuttingDown.remove(tserverPath.toString());
+ log.trace("removed {} from current set and adding to doomed list", tserverPath.getServer());
}
Long firstSeen = locklessServers.get(tserverPath);
if (firstSeen == null) {
locklessServers.put(tserverPath, System.currentTimeMillis());
+ log.trace("first seen, added {} to list of lockless servers", tserverPath.getServer());
} else if (System.currentTimeMillis() - firstSeen > MINUTES.toMillis(10)) {
deleteServerNode(tserverPath.toString());
locklessServers.remove(tserverPath);
+ log.trace(
+ "deleted zookeeper node for server: {}, has been without lock for over 10 minutes",
+ tserverPath.getServer());
}
} else {
+ log.trace("Lock exists for server: {}, adding to current set", tserverPath.getServer());
locklessServers.remove(tserverPath);
HostAndPort address = sld.orElseThrow().getAddress(ServiceLockData.ThriftService.TSERV);
String resourceGroup = sld.orElseThrow().getGroup(ServiceLockData.ThriftService.TSERV);
@@ -325,32 +325,60 @@ public class LiveTServerSet implements ZooCacheWatcher {
@Override
public void accept(WatchedEvent event) {
+ log.trace("Received event: {}", event);
// its important that these event are propagated by ZooCache, because this ensures when reading
// zoocache that is has already processed the event and cleared
// relevant nodes before code below reads from zoocache
-
- if (event.getPath() != null) {
- if (event.getPath().endsWith(Constants.ZTSERVERS)) {
+ final String tserverZPath = context.getZooKeeperRoot() + Constants.ZTSERVERS;
+ if (event.getPath() != null && event.getPath().startsWith(tserverZPath)) {
+ if (event.getPath().equals(tserverZPath)) {
scanServers();
- } else {
- try {
- final ServiceLockPath slp =
- ServiceLockPaths.parse(Optional.of(Constants.ZTSERVERS), event.getPath());
- if (slp.getType().equals(Constants.ZTSERVERS)) {
- try {
- final Set<TServerInstance> updates = new HashSet<>();
- final Set<TServerInstance> doomed = new HashSet<>();
- checkServer(updates, doomed, slp);
- this.cback.update(this, doomed, updates);
- } catch (Exception ex) {
- log.error("Error processing event for tserver: " + slp.toString(), ex);
+ } else if (event.getPath().contains(Constants.ZTSERVERS)) {
+ // It's possible that the path contains more than the tserver address, it
+ // could contain it's children. We need to fix the path before parsing it
+ // path should be: zooKeeperRoot + Constants.ZTSERVERS + "/" + resourceGroup + "/" address
+ String pathToUse = null;
+ String remaining = event.getPath().substring(tserverZPath.length() + 1);
+ int numSlashes = StringUtils.countMatches(remaining, '/');
+ if (numSlashes == 1) {
+ // event path is the server
+ pathToUse = event.getPath();
+ } else if (numSlashes > 1) {
+ // event path is the children of the server, maybe zlock
+ int idx = remaining.indexOf("/");
+ String rg = remaining.substring(0, idx);
+ String server = remaining.substring(idx + 1, remaining.indexOf("/", idx + 1));
+ pathToUse = tserverZPath + "/" + rg + "/" + server;
+ } else {
+ // malformed path
+ pathToUse = null;
+ log.debug("Received event for path that can't be parsed, path: " + event.getPath());
+ }
+ if (pathToUse != null) {
+ try {
+ final ServiceLockPath slp =
+ ServiceLockPaths.parse(Optional.of(Constants.ZTSERVERS), pathToUse);
+ if (slp.getType().equals(Constants.ZTSERVERS)) {
+ try {
+ log.trace("Processing event for server: {}", slp.toString());
+ final Set<TServerInstance> updates = new HashSet<>();
+ final Set<TServerInstance> doomed = new HashSet<>();
+ checkServer(updates, doomed, slp);
+ this.cback.update(this, doomed, updates);
+ } catch (Exception ex) {
+ log.error("Error processing event for tserver: " + slp.toString(), ex);
+ }
}
+ } catch (IllegalArgumentException e) {
+ log.debug("Received event for path that can't be parsed, path: {} pathToUse:{} ",
+ event.getPath(), pathToUse);
}
- } catch (IllegalArgumentException e) {
- log.debug("Received event for path that can't be parsed, path: " + event.getPath());
}
+ } else {
+ // we don't care about other paths
}
}
+
}
public synchronized TServerConnection getConnection(TServerInstance server) {
@@ -503,7 +531,7 @@ public class LiveTServerSet implements ZooCacheWatcher {
log.error("FATAL: {}", msg, e);
Halt.halt(msg, -1);
}
- getZooCache().clear(slp.toString());
+ context.getZooCache().clear(slp.toString());
}
}
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/handler/KerberosAuthenticator.java b/server/base/src/main/java/org/apache/accumulo/server/security/handler/KerberosAuthenticator.java
index f38cb30d61..71ac8e94c4 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/handler/KerberosAuthenticator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/handler/KerberosAuthenticator.java
@@ -29,7 +29,6 @@ import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.KerberosToken;
import org.apache.accumulo.core.clientImpl.DelegationTokenImpl;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
@@ -51,7 +50,6 @@ public class KerberosAuthenticator implements Authenticator {
Set.of(KerberosToken.class.getName(), SystemToken.class.getName());
private final ZKAuthenticator zkAuthenticator = new ZKAuthenticator();
- private ZooCache zooCache;
private ServerContext context;
private String zkUserPath;
private UserImpersonation impersonation;
@@ -59,7 +57,6 @@ public class KerberosAuthenticator implements Authenticator {
@Override
public void initialize(ServerContext context) {
this.context = context;
- zooCache = new ZooCache(context.getZooSession());
impersonation = new UserImpersonation(context.getConfiguration());
zkAuthenticator.initialize(context);
zkUserPath = context.zkUserPath();
@@ -71,12 +68,9 @@ public class KerberosAuthenticator implements Authenticator {
}
private void createUserNodeInZk(String principal) throws KeeperException, InterruptedException {
- synchronized (zooCache) {
- zooCache.clear();
- ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
- zoo.putPrivatePersistentData(zkUserPath + "/" + principal, new byte[0],
- NodeExistsPolicy.FAIL);
- }
+ context.getZooCache().clear(zkUserPath + "/" + principal);
+ ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
+ zoo.putPrivatePersistentData(zkUserPath + "/" + principal, new byte[0], NodeExistsPolicy.FAIL);
}
@Override
@@ -84,22 +78,20 @@ public class KerberosAuthenticator implements Authenticator {
try {
// remove old settings from zookeeper first, if any
ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
- synchronized (zooCache) {
- zooCache.clear();
- if (zoo.exists(zkUserPath)) {
- zoo.recursiveDelete(zkUserPath, NodeMissingPolicy.SKIP);
- log.info("Removed {}/ from zookeeper", zkUserPath);
- }
-
- // prep parent node of users with root username
- // ACCUMULO-4140 The root user needs to be stored un-base64 encoded in the znode's value
- byte[] principalData = principal.getBytes(UTF_8);
- zoo.putPersistentData(zkUserPath, principalData, NodeExistsPolicy.FAIL);
-
- // Create the root user in ZK using base64 encoded name (since the name is included in the
- // znode)
- createUserNodeInZk(Base64.getEncoder().encodeToString(principalData));
+ context.getZooCache().clear((path) -> path.startsWith(zkUserPath));
+ if (zoo.exists(zkUserPath)) {
+ zoo.recursiveDelete(zkUserPath, NodeMissingPolicy.SKIP);
+ log.info("Removed {}/ from zookeeper", zkUserPath);
}
+
+ // prep parent node of users with root username
+ // ACCUMULO-4140 The root user needs to be stored un-base64 encoded in the znode's value
+ byte[] principalData = principal.getBytes(UTF_8);
+ zoo.putPersistentData(zkUserPath, principalData, NodeExistsPolicy.FAIL);
+
+ // Create the root user in ZK using base64 encoded name (since the name is included in the
+ // znode)
+ createUserNodeInZk(Base64.getEncoder().encodeToString(principalData));
} catch (KeeperException | InterruptedException e) {
log.error("Failed to initialize security", e);
throw new IllegalStateException(e);
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthenticator.java b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthenticator.java
index 1bcb54f0e2..7cf087ac47 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthenticator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthenticator.java
@@ -29,7 +29,6 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
@@ -44,12 +43,10 @@ public final class ZKAuthenticator implements Authenticator {
private ServerContext context;
private String zkUserPath;
- private ZooCache zooCache;
@Override
public void initialize(ServerContext context) {
this.context = context;
- zooCache = new ZooCache(context.getZooSession());
zkUserPath = context.zkUserPath();
}
@@ -58,18 +55,16 @@ public final class ZKAuthenticator implements Authenticator {
try {
// remove old settings from zookeeper first, if any
ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
- synchronized (zooCache) {
- zooCache.clear();
- if (zoo.exists(zkUserPath)) {
- zoo.recursiveDelete(zkUserPath, NodeMissingPolicy.SKIP);
- log.info("Removed {}/ from zookeeper", zkUserPath);
- }
+ context.getZooCache().clear((path) -> path.startsWith(zkUserPath));
+ if (zoo.exists(zkUserPath)) {
+ zoo.recursiveDelete(zkUserPath, NodeMissingPolicy.SKIP);
+ log.info("Removed {}/ from zookeeper", zkUserPath);
+ }
- // prep parent node of users with root username
- zoo.putPersistentData(zkUserPath, principal.getBytes(UTF_8), NodeExistsPolicy.FAIL);
+ // prep parent node of users with root username
+ zoo.putPersistentData(zkUserPath, principal.getBytes(UTF_8), NodeExistsPolicy.FAIL);
- constructUser(principal, ZKSecurityTool.createPass(token));
- }
+ constructUser(principal, ZKSecurityTool.createPass(token));
} catch (KeeperException | AccumuloException | InterruptedException e) {
log.error("{}", e.getMessage(), e);
throw new IllegalStateException(e);
@@ -82,16 +77,15 @@ public final class ZKAuthenticator implements Authenticator {
*/
private void constructUser(String user, byte[] pass)
throws KeeperException, InterruptedException {
- synchronized (zooCache) {
- zooCache.clear();
- ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
- zoo.putPrivatePersistentData(zkUserPath + "/" + user, pass, NodeExistsPolicy.FAIL);
- }
+ String userPath = zkUserPath + "/" + user;
+ context.getZooCache().clear((path) -> path.startsWith(userPath));
+ context.getZooSession().asReaderWriter().putPrivatePersistentData(userPath, pass,
+ NodeExistsPolicy.FAIL);
}
@Override
public Set<String> listUsers() {
- return new TreeSet<>(zooCache.getChildren(zkUserPath));
+ return new TreeSet<>(context.getZooCache().getChildren(zkUserPath));
}
@Override
@@ -120,11 +114,9 @@ public final class ZKAuthenticator implements Authenticator {
@Override
public void dropUser(String user) throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- zooCache.clear();
- context.getZooSession().asReaderWriter().recursiveDelete(zkUserPath + "/" + user,
- NodeMissingPolicy.FAIL);
- }
+ String userPath = zkUserPath + "/" + user;
+ context.getZooCache().clear((path) -> path.startsWith(userPath));
+ context.getZooSession().asReaderWriter().recursiveDelete(userPath, NodeMissingPolicy.FAIL);
} catch (InterruptedException e) {
log.error("{}", e.getMessage(), e);
throw new IllegalStateException(e);
@@ -146,12 +138,10 @@ public final class ZKAuthenticator implements Authenticator {
PasswordToken pt = (PasswordToken) token;
if (userExists(principal)) {
try {
- synchronized (zooCache) {
- zooCache.clear(zkUserPath + "/" + principal);
- context.getZooSession().asReaderWriter().putPrivatePersistentData(
- zkUserPath + "/" + principal, ZKSecurityTool.createPass(pt.getPassword()),
- NodeExistsPolicy.OVERWRITE);
- }
+ String userPath = zkUserPath + "/" + principal;
+ context.getZooCache().clear(userPath);
+ context.getZooSession().asReaderWriter().putPrivatePersistentData(userPath,
+ ZKSecurityTool.createPass(pt.getPassword()), NodeExistsPolicy.OVERWRITE);
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
throw new AccumuloSecurityException(principal, SecurityErrorCode.CONNECTION_ERROR, e);
@@ -170,7 +160,7 @@ public final class ZKAuthenticator implements Authenticator {
@Override
public boolean userExists(String user) {
- return zooCache.get(zkUserPath + "/" + user) != null;
+ return context.getZooCache().get(zkUserPath + "/" + user) != null;
}
@Override
@@ -187,11 +177,11 @@ public final class ZKAuthenticator implements Authenticator {
PasswordToken pt = (PasswordToken) token;
byte[] zkData;
String zpath = zkUserPath + "/" + principal;
- zkData = zooCache.get(zpath);
+ zkData = context.getZooCache().get(zpath);
boolean result = authenticateUser(principal, pt, zkData);
if (!result) {
- zooCache.clear(zpath);
- zkData = zooCache.get(zpath);
+ context.getZooCache().clear(zpath);
+ zkData = context.getZooCache().get(zpath);
result = authenticateUser(principal, pt, zkData);
}
return result;
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthorizor.java b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthorizor.java
index 5d18ac4625..a63eee6e82 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthorizor.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKAuthorizor.java
@@ -25,7 +25,6 @@ import java.util.List;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
@@ -44,18 +43,16 @@ public class ZKAuthorizor implements Authorizor {
private ServerContext context;
private String zkUserPath;
- private ZooCache zooCache;
@Override
public void initialize(ServerContext context) {
this.context = context;
- zooCache = new ZooCache(context.getZooSession());
zkUserPath = context.zkUserPath();
}
@Override
public Authorizations getCachedUserAuthorizations(String user) {
- byte[] authsBytes = zooCache.get(zkUserPath + "/" + user + ZKUserAuths);
+ byte[] authsBytes = context.getZooCache().get(zkUserPath + "/" + user + ZKUserAuths);
if (authsBytes != null) {
return ZKSecurityTool.convertAuthorizations(authsBytes);
}
@@ -105,11 +102,9 @@ public class ZKAuthorizor implements Authorizor {
@Override
public void dropUser(String user) throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserAuths, NodeMissingPolicy.SKIP);
- zooCache.clear(zkUserPath + "/" + user);
- }
+ context.getZooSession().asReaderWriter()
+ .recursiveDelete(zkUserPath + "/" + user + ZKUserAuths, NodeMissingPolicy.SKIP);
+ context.getZooCache().clear((path) -> path.startsWith(zkUserPath + "/" + user));
} catch (InterruptedException e) {
log.error("{}", e.getMessage(), e);
throw new IllegalStateException(e);
@@ -127,12 +122,10 @@ public class ZKAuthorizor implements Authorizor {
public void changeAuthorizations(String user, Authorizations authorizations)
throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- zooCache.clear();
- context.getZooSession().asReaderWriter().putPersistentData(
- zkUserPath + "/" + user + ZKUserAuths,
- ZKSecurityTool.convertAuthorizations(authorizations), NodeExistsPolicy.OVERWRITE);
- }
+ String userAuths = zkUserPath + "/" + user + ZKUserAuths;
+ context.getZooCache().clear(userAuths);
+ context.getZooSession().asReaderWriter().putPersistentData(userAuths,
+ ZKSecurityTool.convertAuthorizations(authorizations), NodeExistsPolicy.OVERWRITE);
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
throw new AccumuloSecurityException(user, SecurityErrorCode.CONNECTION_ERROR, e);
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKPermHandler.java b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKPermHandler.java
index 1cc933c347..4318b9448b 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKPermHandler.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/handler/ZKPermHandler.java
@@ -35,7 +35,6 @@ import org.apache.accumulo.core.clientImpl.Namespace;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
@@ -53,18 +52,18 @@ import org.slf4j.LoggerFactory;
public class ZKPermHandler implements PermissionHandler {
private static final Logger log = LoggerFactory.getLogger(ZKPermHandler.class);
+ private ServerContext ctx;
private ZooReaderWriter zoo;
private String zkUserPath;
private String ZKTablePath;
private String ZKNamespacePath;
- private ZooCache zooCache;
private final String ZKUserSysPerms = "/System";
private final String ZKUserTablePerms = "/Tables";
private final String ZKUserNamespacePerms = "/Namespaces";
@Override
public void initialize(ServerContext context) {
- zooCache = new ZooCache(context.getZooSession());
+ this.ctx = context;
zoo = context.getZooSession().asReaderWriter();
zkUserPath = context.zkUserPath();
ZKTablePath = context.getZooKeeperRoot() + Constants.ZTABLES;
@@ -113,7 +112,8 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public boolean hasCachedTablePermission(String user, String table, TablePermission permission) {
- byte[] serializedPerms = zooCache.get(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table);
+ byte[] serializedPerms =
+ ctx.getZooCache().get(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table);
if (serializedPerms != null) {
return ZKSecurityTool.convertTablePermissions(serializedPerms).contains(permission);
}
@@ -165,7 +165,7 @@ public class ZKPermHandler implements PermissionHandler {
public boolean hasCachedNamespacePermission(String user, String namespace,
NamespacePermission permission) {
byte[] serializedPerms =
- zooCache.get(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace);
+ ctx.getZooCache().get(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace);
if (serializedPerms != null) {
return ZKSecurityTool.convertNamespacePermissions(serializedPerms).contains(permission);
}
@@ -175,8 +175,9 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void grantSystemPermission(String user, SystemPermission permission)
throws AccumuloSecurityException {
+ final String sysPermPath = zkUserPath + "/" + user + ZKUserSysPerms;
try {
- byte[] permBytes = zooCache.get(zkUserPath + "/" + user + ZKUserSysPerms);
+ byte[] permBytes = ctx.getZooCache().get(sysPermPath);
Set<SystemPermission> perms;
if (permBytes == null) {
perms = new TreeSet<>();
@@ -185,11 +186,9 @@ public class ZKPermHandler implements PermissionHandler {
}
if (perms.add(permission)) {
- synchronized (zooCache) {
- zooCache.clear();
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserSysPerms,
- ZKSecurityTool.convertSystemPermissions(perms), NodeExistsPolicy.OVERWRITE);
- }
+ ctx.getZooCache().clear(sysPermPath);
+ zoo.putPersistentData(sysPermPath, ZKSecurityTool.convertSystemPermissions(perms),
+ NodeExistsPolicy.OVERWRITE);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -204,7 +203,8 @@ public class ZKPermHandler implements PermissionHandler {
public void grantTablePermission(String user, String table, TablePermission permission)
throws AccumuloSecurityException {
Set<TablePermission> tablePerms;
- byte[] serializedPerms = zooCache.get(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table);
+ final String tablePermPath = zkUserPath + "/" + user + ZKUserTablePerms + "/" + table;
+ byte[] serializedPerms = ctx.getZooCache().get(tablePermPath);
if (serializedPerms != null) {
tablePerms = ZKSecurityTool.convertTablePermissions(serializedPerms);
} else {
@@ -213,11 +213,9 @@ public class ZKPermHandler implements PermissionHandler {
try {
if (tablePerms.add(permission)) {
- synchronized (zooCache) {
- zooCache.clear(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table);
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table,
- ZKSecurityTool.convertTablePermissions(tablePerms), NodeExistsPolicy.OVERWRITE);
- }
+ ctx.getZooCache().clear(tablePermPath);
+ zoo.putPersistentData(tablePermPath, ZKSecurityTool.convertTablePermissions(tablePerms),
+ NodeExistsPolicy.OVERWRITE);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -231,9 +229,9 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void grantNamespacePermission(String user, String namespace,
NamespacePermission permission) throws AccumuloSecurityException {
+ final String nsPermPath = zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace;
Set<NamespacePermission> namespacePerms;
- byte[] serializedPerms =
- zooCache.get(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace);
+ byte[] serializedPerms = ctx.getZooCache().get(nsPermPath);
if (serializedPerms != null) {
namespacePerms = ZKSecurityTool.convertNamespacePermissions(serializedPerms);
} else {
@@ -242,12 +240,9 @@ public class ZKPermHandler implements PermissionHandler {
try {
if (namespacePerms.add(permission)) {
- synchronized (zooCache) {
- zooCache.clear(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace);
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace,
- ZKSecurityTool.convertNamespacePermissions(namespacePerms),
- NodeExistsPolicy.OVERWRITE);
- }
+ ctx.getZooCache().clear(nsPermPath);
+ zoo.putPersistentData(nsPermPath,
+ ZKSecurityTool.convertNamespacePermissions(namespacePerms), NodeExistsPolicy.OVERWRITE);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -261,7 +256,8 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void revokeSystemPermission(String user, SystemPermission permission)
throws AccumuloSecurityException {
- byte[] sysPermBytes = zooCache.get(zkUserPath + "/" + user + ZKUserSysPerms);
+ final String sysPermPath = zkUserPath + "/" + user + ZKUserSysPerms;
+ byte[] sysPermBytes = ctx.getZooCache().get(sysPermPath);
// User had no system permission, nothing to revoke.
if (sysPermBytes == null) {
@@ -272,11 +268,9 @@ public class ZKPermHandler implements PermissionHandler {
try {
if (sysPerms.remove(permission)) {
- synchronized (zooCache) {
- zooCache.clear();
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserSysPerms,
- ZKSecurityTool.convertSystemPermissions(sysPerms), NodeExistsPolicy.OVERWRITE);
- }
+ ctx.getZooCache().clear((path) -> path.startsWith(sysPermPath));
+ zoo.putPersistentData(sysPermPath, ZKSecurityTool.convertSystemPermissions(sysPerms),
+ NodeExistsPolicy.OVERWRITE);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -290,7 +284,8 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void revokeTablePermission(String user, String table, TablePermission permission)
throws AccumuloSecurityException {
- byte[] serializedPerms = zooCache.get(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table);
+ final String tablePermPath = zkUserPath + "/" + user + ZKUserTablePerms + "/" + table;
+ byte[] serializedPerms = ctx.getZooCache().get(tablePermPath);
// User had no table permission, nothing to revoke.
if (serializedPerms == null) {
@@ -300,13 +295,12 @@ public class ZKPermHandler implements PermissionHandler {
Set<TablePermission> tablePerms = ZKSecurityTool.convertTablePermissions(serializedPerms);
try {
if (tablePerms.remove(permission)) {
- zooCache.clear();
+ ctx.getZooCache().clear((path) -> path.startsWith(tablePermPath));
if (tablePerms.isEmpty()) {
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table,
- NodeMissingPolicy.SKIP);
+ zoo.recursiveDelete(tablePermPath, NodeMissingPolicy.SKIP);
} else {
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table,
- ZKSecurityTool.convertTablePermissions(tablePerms), NodeExistsPolicy.OVERWRITE);
+ zoo.putPersistentData(tablePermPath, ZKSecurityTool.convertTablePermissions(tablePerms),
+ NodeExistsPolicy.OVERWRITE);
}
}
} catch (KeeperException e) {
@@ -321,8 +315,8 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void revokeNamespacePermission(String user, String namespace,
NamespacePermission permission) throws AccumuloSecurityException {
- byte[] serializedPerms =
- zooCache.get(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace);
+ final String nsPermPath = zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace;
+ byte[] serializedPerms = ctx.getZooCache().get(nsPermPath);
// User had no namespace permission, nothing to revoke.
if (serializedPerms == null) {
@@ -333,12 +327,11 @@ public class ZKPermHandler implements PermissionHandler {
ZKSecurityTool.convertNamespacePermissions(serializedPerms);
try {
if (namespacePerms.remove(permission)) {
- zooCache.clear();
+ ctx.getZooCache().clear((path) -> path.startsWith(nsPermPath));
if (namespacePerms.isEmpty()) {
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace,
- NodeMissingPolicy.SKIP);
+ zoo.recursiveDelete(nsPermPath, NodeMissingPolicy.SKIP);
} else {
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace,
+ zoo.putPersistentData(nsPermPath,
ZKSecurityTool.convertNamespacePermissions(namespacePerms),
NodeExistsPolicy.OVERWRITE);
}
@@ -355,12 +348,10 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void cleanTablePermissions(String table) throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- zooCache.clear();
- for (String user : zooCache.getChildren(zkUserPath)) {
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table,
- NodeMissingPolicy.SKIP);
- }
+ for (String user : ctx.getZooCache().getChildren(zkUserPath)) {
+ final String tablePermPath = zkUserPath + "/" + user + ZKUserTablePerms + "/" + table;
+ ctx.getZooCache().clear((path) -> path.startsWith(tablePermPath));
+ zoo.recursiveDelete(tablePermPath, NodeMissingPolicy.SKIP);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -374,12 +365,10 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public void cleanNamespacePermissions(String namespace) throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- zooCache.clear();
- for (String user : zooCache.getChildren(zkUserPath)) {
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace,
- NodeMissingPolicy.SKIP);
- }
+ for (String user : ctx.getZooCache().getChildren(zkUserPath)) {
+ final String nsPermPath = zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace;
+ ctx.getZooCache().clear((path) -> path.startsWith(nsPermPath));
+ zoo.recursiveDelete(nsPermPath, NodeMissingPolicy.SKIP);
}
} catch (KeeperException e) {
log.error("{}", e.getMessage(), e);
@@ -455,11 +444,10 @@ public class ZKPermHandler implements PermissionHandler {
*/
private void createTablePerm(String user, TableId table, Set<TablePermission> perms)
throws KeeperException, InterruptedException {
- synchronized (zooCache) {
- zooCache.clear();
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserTablePerms + "/" + table,
- ZKSecurityTool.convertTablePermissions(perms), NodeExistsPolicy.FAIL);
- }
+ final String tablePermPath = zkUserPath + "/" + user + ZKUserTablePerms + "/" + table;
+ ctx.getZooCache().clear((path) -> path.startsWith(tablePermPath));
+ zoo.putPersistentData(tablePermPath, ZKSecurityTool.convertTablePermissions(perms),
+ NodeExistsPolicy.FAIL);
}
/**
@@ -468,22 +456,19 @@ public class ZKPermHandler implements PermissionHandler {
*/
private void createNamespacePerm(String user, NamespaceId namespace,
Set<NamespacePermission> perms) throws KeeperException, InterruptedException {
- synchronized (zooCache) {
- zooCache.clear();
- zoo.putPersistentData(zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace,
- ZKSecurityTool.convertNamespacePermissions(perms), NodeExistsPolicy.FAIL);
- }
+ final String nsPermPath = zkUserPath + "/" + user + ZKUserNamespacePerms + "/" + namespace;
+ ctx.getZooCache().clear((path) -> path.startsWith(nsPermPath));
+ zoo.putPersistentData(nsPermPath, ZKSecurityTool.convertNamespacePermissions(perms),
+ NodeExistsPolicy.FAIL);
}
@Override
public void cleanUser(String user) throws AccumuloSecurityException {
try {
- synchronized (zooCache) {
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserSysPerms, NodeMissingPolicy.SKIP);
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserTablePerms, NodeMissingPolicy.SKIP);
- zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserNamespacePerms, NodeMissingPolicy.SKIP);
- zooCache.clear(zkUserPath + "/" + user);
- }
+ zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserSysPerms, NodeMissingPolicy.SKIP);
+ zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserTablePerms, NodeMissingPolicy.SKIP);
+ zoo.recursiveDelete(zkUserPath + "/" + user + ZKUserNamespacePerms, NodeMissingPolicy.SKIP);
+ ctx.getZooCache().clear((path) -> path.startsWith(zkUserPath + "/" + user));
} catch (InterruptedException e) {
log.error("{}", e.getMessage(), e);
throw new IllegalStateException(e);
@@ -523,7 +508,7 @@ public class ZKPermHandler implements PermissionHandler {
@Override
public boolean hasCachedSystemPermission(String user, SystemPermission permission) {
- byte[] perms = zooCache.get(zkUserPath + "/" + user + ZKUserSysPerms);
+ byte[] perms = ctx.getZooCache().get(zkUserPath + "/" + user + ZKUserSysPerms);
if (perms == null) {
return false;
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java b/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
index 3bd0f091de..be97a3bf0b 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
@@ -35,8 +35,6 @@ import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
import org.apache.accumulo.core.data.InstanceId;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZooCacheWatcher;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
@@ -44,6 +42,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.tables.TableNameUtil;
+import org.apache.accumulo.core.zookeeper.ZooCache.ZooCacheWatcher;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.PropStore;
@@ -69,7 +68,6 @@ public class TableManager {
private final String zkRoot;
private final InstanceId instanceID;
private final ZooReaderWriter zoo;
- private final ZooCache zooStateCache;
public static void prepareNewNamespaceState(final ServerContext context, NamespaceId namespaceId,
String namespace, NodeExistsPolicy existsPolicy)
@@ -122,7 +120,8 @@ public class TableManager {
zkRoot = context.getZooKeeperRoot();
instanceID = context.getInstanceID();
zoo = context.getZooSession().asReaderWriter();
- zooStateCache = new ZooCache(context.getZooSession(), new TableStateWatcher());
+ // add our Watcher to the shared ZooCache
+ context.getZooCache().addZooCacheWatcher(new TableStateWatcher());
updateTableStateCache();
}
@@ -179,9 +178,9 @@ public class TableManager {
private void updateTableStateCache() {
synchronized (tableStateCache) {
- for (String tableId : zooStateCache.getChildren(zkRoot + Constants.ZTABLES)) {
- if (zooStateCache.get(zkRoot + Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_STATE)
- != null) {
+ for (String tableId : context.getZooCache().getChildren(zkRoot + Constants.ZTABLES)) {
+ if (context.getZooCache()
+ .get(zkRoot + Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_STATE) != null) {
updateTableStateCache(TableId.of(tableId));
}
}
@@ -191,8 +190,8 @@ public class TableManager {
public TableState updateTableStateCache(TableId tableId) {
synchronized (tableStateCache) {
TableState tState = TableState.UNKNOWN;
- byte[] data =
- zooStateCache.get(zkRoot + Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_STATE);
+ byte[] data = context.getZooCache()
+ .get(zkRoot + Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_STATE);
if (data != null) {
String sState = new String(data, UTF_8);
try {
@@ -270,27 +269,28 @@ public class TableManager {
}
}
if (tableId == null) {
- log.warn("Unknown path in {}", event);
+ log.trace("Unhandled path {}", event);
return;
}
}
switch (zType) {
case NodeChildrenChanged:
- if (zPath != null && zPath.equals(tablesPrefix)) {
- updateTableStateCache();
- } else {
- log.warn("Unexpected path {} in {}", zPath, event);
- }
+ // According to documentation we should not receive this event now
+ // that ZooCache is using Persistent Watchers. Not logging an error here.
+ // According to https://issues.apache.org/jira/browse/ZOOKEEPER-4475 we
+ // may receive this event (Fixed in 3.9.0)
break;
case NodeCreated:
case NodeDataChanged:
// state transition
- TableState tState = updateTableStateCache(tableId);
- log.debug("State transition to {} @ {}", tState, event);
- synchronized (observers) {
- for (TableObserver to : observers) {
- to.stateChanged(tableId, tState);
+ if (tableId != null) {
+ TableState tState = updateTableStateCache(tableId);
+ log.debug("State transition to {} @ {}", tState, event);
+ synchronized (observers) {
+ for (TableObserver to : observers) {
+ to.stateChanged(tableId, tState);
+ }
}
}
break;
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java b/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
index f8e8df3c52..47dcd1e09c 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
@@ -73,7 +73,6 @@ import org.apache.accumulo.core.fate.FateStore;
import org.apache.accumulo.core.fate.ReadOnlyFateStore;
import org.apache.accumulo.core.fate.user.UserFateStore;
import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
@@ -96,6 +95,7 @@ import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.AddressUtil;
import org.apache.accumulo.core.util.Halt;
import org.apache.accumulo.core.util.tables.TableMap;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.cli.ServerUtilOpts;
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java b/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
index f075f50cf9..9b427be39a 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
@@ -21,13 +21,13 @@ package org.apache.accumulo.server.util;
import java.util.Optional;
import java.util.Set;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.server.ServerContext;
import com.google.common.base.Preconditions;
diff --git a/server/base/src/test/java/org/apache/accumulo/server/manager/LiveTServerSetTest.java b/server/base/src/test/java/org/apache/accumulo/server/manager/LiveTServerSetTest.java
index 20326d4bf0..da6833b989 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/manager/LiveTServerSetTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/manager/LiveTServerSetTest.java
@@ -26,6 +26,7 @@ import java.util.Map;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
import org.apache.accumulo.server.manager.LiveTServerSet.TServerConnection;
@@ -47,13 +48,21 @@ public class LiveTServerSetTest {
mockConn, Constants.DEFAULT_RESOURCE_GROUP_NAME);
servers.put("server1", server1);
- LiveTServerSet tservers = new LiveTServerSet(EasyMock.createMock(ServerContext.class),
- EasyMock.createMock(Listener.class));
+ ServerContext ctx = EasyMock.createMock(ServerContext.class);
+ ZooCache zc = EasyMock.createMock(ZooCache.class);
+ EasyMock.expect(ctx.getZooCache()).andReturn(zc);
+ zc.addZooCacheWatcher(EasyMock.isA(LiveTServerSet.class));
+ EasyMock.replay(ctx, zc);
+
+ LiveTServerSet tservers = new LiveTServerSet(ctx, EasyMock.createMock(Listener.class));
assertEquals(server1.instance, tservers.find(servers, "localhost:1234"));
assertNull(tservers.find(servers, "localhost:4321"));
assertEquals(server1.instance, tservers.find(servers, "localhost:1234[5555]"));
assertNull(tservers.find(servers, "localhost:1234[55755]"));
+
+ EasyMock.verify(ctx, zc);
+
}
}
diff --git a/server/base/src/test/java/org/apache/accumulo/server/security/handler/ZKAuthenticatorTest.java b/server/base/src/test/java/org/apache/accumulo/server/security/handler/ZKAuthenticatorTest.java
index a3f91a7cf8..4eec2b650a 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/security/handler/ZKAuthenticatorTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/security/handler/ZKAuthenticatorTest.java
@@ -22,7 +22,6 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.matches;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -44,10 +43,10 @@ import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.SystemPermission;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.util.ByteArraySet;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.server.MockServerContext;
import org.apache.accumulo.server.ServerContext;
-import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.junit.jupiter.api.Test;
@@ -144,14 +143,16 @@ public class ZKAuthenticatorTest {
var instanceId = InstanceId.of("example");
ZooSession zk = createMock(ZooSession.class);
ServerContext context = MockServerContext.getWithMockZK(zk);
+ ZooCache zc = createMock(ZooCache.class);
expect(context.zkUserPath()).andReturn(ZooUtil.getRoot(instanceId) + Constants.ZUSERS)
.anyTimes();
expect(zk.getChildren(anyObject(), anyObject())).andReturn(Arrays.asList(principal)).anyTimes();
- expect(zk.exists(matches(ZooUtil.getRoot(instanceId) + Constants.ZUSERS + "/" + principal),
- anyObject(Watcher.class))).andReturn(new Stat()).anyTimes();
- expect(zk.getData(matches(ZooUtil.getRoot(instanceId) + Constants.ZUSERS + "/" + principal),
- anyObject(), anyObject())).andReturn(newHash).once();
- replay(context, zk);
+ expect(zk.exists(ZooUtil.getRoot(instanceId) + Constants.ZUSERS + "/" + principal, null))
+ .andReturn(new Stat()).anyTimes();
+ expect(context.getZooCache()).andReturn(zc).anyTimes();
+ expect(zc.get(ZooUtil.getRoot(instanceId) + Constants.ZUSERS + "/" + principal))
+ .andReturn(newHash);
+ replay(context, zk, zc);
// creating authenticator
ZKAuthenticator auth = new ZKAuthenticator();
@@ -160,6 +161,6 @@ public class ZKAuthenticatorTest {
PasswordToken token = new PasswordToken(rawPass.clone());
// verifying that if the new type of hash is stored in zk authentication works as expected
assertTrue(auth.authenticateUser(principal, token));
- verify(context, zk);
+ verify(context, zk, zc);
}
}
diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/AdminTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/AdminTest.java
index 3e77560e49..74b590ff35 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/util/AdminTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/util/AdminTest.java
@@ -50,8 +50,6 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.fate.FateInstanceType;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
@@ -63,6 +61,8 @@ import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletOperationId;
import org.apache.accumulo.core.metadata.schema.TabletOperationType;
import org.apache.accumulo.core.util.time.SteadyTime;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.hadoop.fs.Path;
import org.apache.zookeeper.KeeperException;
import org.easymock.EasyMock;
diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/ServiceStatusCmdTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/ServiceStatusCmdTest.java
index ffc2815a99..9d83992bfd 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/util/ServiceStatusCmdTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/util/ServiceStatusCmdTest.java
@@ -39,11 +39,11 @@ import java.util.UUID;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.data.InstanceId;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooReader;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLockPaths;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.util.serviceStatus.ServiceStatusReport;
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index f07b303c37..d659dd2884 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@ -79,7 +79,6 @@ import org.apache.accumulo.core.fate.FateInstanceType;
import org.apache.accumulo.core.fate.FateStore;
import org.apache.accumulo.core.fate.user.UserFateStore;
import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
@@ -121,6 +120,7 @@ import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.core.util.time.SteadyTime;
+import org.apache.accumulo.core.zookeeper.ZcStat;
import org.apache.accumulo.manager.compaction.coordinator.CompactionCoordinator;
import org.apache.accumulo.manager.metrics.BalancerMetrics;
import org.apache.accumulo.manager.metrics.ManagerMetrics;
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
index 9c2dd1411d..355432bb2e 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
@@ -37,7 +37,6 @@ import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.cache.Caches.CacheName;
import org.apache.accumulo.core.util.threads.ThreadPools;
@@ -65,7 +64,6 @@ public class RecoveryManager {
private final Cache<Path,Boolean> existenceCache;
private final ScheduledExecutorService executor;
private final Manager manager;
- private final ZooCache zooCache;
public RecoveryManager(Manager manager, long timeToCacheExistsInMillis) {
this.manager = manager;
@@ -76,7 +74,6 @@ public class RecoveryManager {
executor =
ThreadPools.getServerThreadPools().createScheduledExecutorService(4, "Walog sort starter");
- zooCache = new ZooCache(manager.getContext().getZooSession());
try {
List<String> workIDs =
new DistributedWorkQueue(manager.getContext().getZooKeeperRoot() + Constants.ZRECOVERY,
@@ -182,7 +179,7 @@ public class RecoveryManager {
sortQueued = sortsQueued.contains(sortId);
}
- if (sortQueued && zooCache.get(
+ if (sortQueued && this.manager.getContext().getZooCache().get(
manager.getContext().getZooKeeperRoot() + Constants.ZRECOVERY + "/" + sortId) == null) {
synchronized (this) {
sortsQueued.remove(sortId);
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
index 5e942a47c2..b590b1b191 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
@@ -66,7 +66,6 @@ import org.apache.accumulo.core.dataImpl.thrift.ScanResult;
import org.apache.accumulo.core.dataImpl.thrift.TColumn;
import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.TRange;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration;
@@ -204,8 +203,6 @@ public class ScanServer extends AbstractServer
private ScanServerMetrics scanServerMetrics;
private BlockCacheMetrics blockCacheMetrics;
- private final ZooCache managerLockCache;
-
public ScanServer(ConfigOpts opts, String[] args) {
super(ServerId.Type.SCAN_SERVER, opts, ServerContext::new, args);
@@ -216,8 +213,6 @@ public class ScanServer extends AbstractServer
this.resourceManager = new TabletServerResourceManager(context, this);
- this.managerLockCache = new ZooCache(context.getZooSession());
-
var readWriteLock = new ReentrantReadWriteLock();
reservationsReadLock = readWriteLock.readLock();
reservationsWriteLock = readWriteLock.writeLock();
@@ -1139,11 +1134,6 @@ public class ScanServer extends AbstractServer
return scanServerLock;
}
- @Override
- public ZooCache getManagerLockCache() {
- return managerLockCache;
- }
-
@Override
public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration acuConf) {
return BlockCacheConfiguration.forScanServer(acuConf);
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
index 388bf51409..d45978a9ca 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
@@ -70,6 +70,7 @@ import org.apache.accumulo.core.dataImpl.thrift.UpdateErrors;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.iteratorsImpl.system.IterationInterruptedException;
import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.logging.TabletLogger;
import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
import org.apache.accumulo.core.metadata.AccumuloTable;
@@ -970,11 +971,12 @@ public class TabletClientHandler implements TabletServerClientService.Iface,
new ZooUtil.LockID(context.getServerPaths().createManagerPath().toString(), lock);
try {
- if (!ServiceLock.isLockHeld(server.getManagerLockCache(), lid)) {
+ if (!ServiceLock.isLockHeld(server.getContext().getZooCache(), lid)) {
// maybe the cache is out of date and a new manager holds the
// lock?
- server.getManagerLockCache().clear();
- if (!ServiceLock.isLockHeld(server.getManagerLockCache(), lid)) {
+ server.getContext().getZooCache().clear(
+ (path) -> path.equals(ServiceLockPaths.parse(Optional.empty(), lid.path).toString()));
+ if (!ServiceLock.isLockHeld(server.getContext().getZooCache(), lid)) {
log.warn("Got {} message from a manager that does not hold the current lock {}",
request, lock);
throw new RuntimeException("bad manager lock");
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletHostingServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletHostingServer.java
index 8e5047bb40..9760341a99 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletHostingServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletHostingServer.java
@@ -20,7 +20,6 @@ package org.apache.accumulo.tserver;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.spi.cache.BlockCacheManager;
import org.apache.accumulo.core.spi.scan.ScanServerInfo;
@@ -58,7 +57,5 @@ public interface TabletHostingServer {
ServiceLock getLock();
- ZooCache getManagerLockCache();
-
BlockCacheManager.Configuration getBlockCacheConfiguration(AccumuloConfiguration acuConf);
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 05a6676cc0..a0911a451a 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -77,7 +77,6 @@ import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.data.InstanceId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration;
@@ -169,8 +168,6 @@ public class TabletServer extends AbstractServer implements TabletHostingServer
private static final Logger log = LoggerFactory.getLogger(TabletServer.class);
private static final long TIME_BETWEEN_LOCATOR_CACHE_CLEARS = TimeUnit.HOURS.toMillis(1);
- final ZooCache managerLockCache;
-
final TabletServerLogger logger;
private TabletServerMetrics metrics;
@@ -238,7 +235,6 @@ public class TabletServer extends AbstractServer implements TabletHostingServer
Function<SiteConfiguration,ServerContext> serverContextFactory, String[] args) {
super(ServerId.Type.TABLET_SERVER, opts, serverContextFactory, args);
context = super.getContext();
- this.managerLockCache = new ZooCache(context.getZooSession());
final AccumuloConfiguration aconf = getConfiguration();
log.info("Version " + Constants.VERSION);
log.info("Instance " + getInstanceID());
@@ -489,11 +485,6 @@ public class TabletServer extends AbstractServer implements TabletHostingServer
return tabletServerLock;
}
- @Override
- public ZooCache getManagerLockCache() {
- return managerLockCache;
- }
-
private void announceExistence() {
final ZooReaderWriter zoo = getContext().getZooSession().asReaderWriter();
try {
diff --git a/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java b/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java
index ce0e21c468..9f022be188 100644
--- a/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java
@@ -37,12 +37,12 @@ import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.metadata.AccumuloTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.DeletesSection;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
@@ -96,7 +96,6 @@ public class BadDeleteMarkersCreatedIT extends AccumuloClusterHarness {
try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build();
ClientContext context = (ClientContext) client) {
ZooCache zcache = context.getZooCache();
- zcache.clear();
var path = context.getServerPaths().createGarbageCollectorPath();
Optional<ServiceLockData> gcLockData;
do {
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/CacheTestReader.java b/test/src/main/java/org/apache/accumulo/test/functional/CacheTestReader.java
index 4d554c5c3e..b82ef144d8 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/CacheTestReader.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/CacheTestReader.java
@@ -25,17 +25,21 @@ import java.io.FileOutputStream;
import java.io.ObjectOutputStream;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
+import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
public class CacheTestReader {
+ private static final Logger LOG = LoggerFactory.getLogger(CacheTestReader.class);
+
@SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "path provided by test")
public static void main(String[] args) throws Exception {
String rootDir = args[0];
@@ -46,49 +50,54 @@ public class CacheTestReader {
File myfile = new File(reportDir + "/" + UUID.randomUUID());
myfile.deleteOnExit();
- try (var zk = new ZooSession(CacheTestReader.class.getSimpleName(), keepers, 30_000, null)) {
- ZooCache zc = new ZooCache(zk);
+ try {
+ try (var zk = new ZooSession(CacheTestReader.class.getSimpleName(), keepers, 30_000, null)) {
+ ZooCache zc = new ZooCache(zk, Set.of("/"));
- while (true) {
- if (myfile.exists() && !myfile.delete()) {
- LoggerFactory.getLogger(CacheTestReader.class).warn("Unable to delete {}", myfile);
- }
+ while (true) {
+ if (myfile.exists() && !myfile.delete()) {
+ LOG.warn("Unable to delete {}", myfile);
+ }
- if (zc.get(rootDir + "/die") != null) {
- return;
- }
+ if (zc.get(rootDir + "/die") != null) {
+ LOG.info("Exiting");
+ return;
+ }
- Map<String,String> readData = new TreeMap<>();
+ Map<String,String> readData = new TreeMap<>();
- for (int i = 0; i < numData; i++) {
- byte[] v = zc.get(rootDir + "/data" + i);
- if (v != null) {
- readData.put(rootDir + "/data" + i, new String(v, UTF_8));
+ for (int i = 0; i < numData; i++) {
+ byte[] v = zc.get(rootDir + "/data" + i);
+ if (v != null) {
+ readData.put(rootDir + "/data" + i, new String(v, UTF_8));
+ }
}
- }
- byte[] v = zc.get(rootDir + "/dataS");
- if (v != null) {
- readData.put(rootDir + "/dataS", new String(v, UTF_8));
- }
-
- List<String> children = zc.getChildren(rootDir + "/dir");
- if (children != null) {
- for (String child : children) {
- readData.put(rootDir + "/dir/" + child, "");
+ byte[] v = zc.get(rootDir + "/dataS");
+ if (v != null) {
+ readData.put(rootDir + "/dataS", new String(v, UTF_8));
}
- }
- FileOutputStream fos = new FileOutputStream(myfile);
- ObjectOutputStream oos = new ObjectOutputStream(fos);
+ List<String> children = zc.getChildren(rootDir + "/dir");
+ if (children != null) {
+ for (String child : children) {
+ readData.put(rootDir + "/dir/" + child, "");
+ }
+ }
- oos.writeObject(readData);
+ LOG.info("Data: {}", readData);
- fos.close();
- oos.close();
+ try (FileOutputStream fos = new FileOutputStream(myfile);
+ ObjectOutputStream oos = new ObjectOutputStream(fos)) {
+ oos.writeObject(readData);
+ }
- Thread.sleep(20);
+ Thread.sleep(20);
+ }
}
+ } catch (Exception e) {
+ LOG.error("CacheTestReader thread failed", e);
}
+
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
index 78edc22525..593f254b72 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedScanIT.java
@@ -50,7 +50,6 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
@@ -59,6 +58,7 @@ import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.spi.metrics.LoggingMeterRegistryFactory;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
import org.apache.accumulo.minicluster.MemoryUnit;
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ReadWriteIT.java b/test/src/main/java/org/apache/accumulo/test/functional/ReadWriteIT.java
index 7a3c1f2089..74d715a244 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ReadWriteIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ReadWriteIT.java
@@ -46,10 +46,10 @@ import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.util.MonitorUtil;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/RestartIT.java b/test/src/main/java/org/apache/accumulo/test/functional/RestartIT.java
index bf515d94bd..85e429dc7a 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/RestartIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/RestartIT.java
@@ -32,12 +32,13 @@ import java.util.concurrent.Future;
import org.apache.accumulo.cluster.ClusterControl;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.ClientProperty;
import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.metadata.AccumuloTable;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
@@ -129,7 +130,7 @@ public class RestartIT extends AccumuloClusterHarness {
control.stopAllServers(ServerType.GARBAGE_COLLECTOR);
control.stopAllServers(ServerType.MONITOR);
- ZooCache zcache = cluster.getServerContext().getZooCache();
+ ZooCache zcache = ((ClientContext) c).getZooCache();
var zLockPath = getServerContext().getServerPaths().createManagerPath();
Optional<ServiceLockData> managerLockData;
do {
@@ -179,7 +180,7 @@ public class RestartIT extends AccumuloClusterHarness {
control.stopAllServers(ServerType.MANAGER);
- ZooCache zcache = cluster.getServerContext().getZooCache();
+ ZooCache zcache = ((ClientContext) c).getZooCache();
var zLockPath = getServerContext().getServerPaths().createManagerPath();
Optional<ServiceLockData> managerLockData;
do {
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ZooCacheIT.java b/test/src/main/java/org/apache/accumulo/test/functional/ZooCacheIT.java
index 1816d834d0..fd8863dbcf 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ZooCacheIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ZooCacheIT.java
@@ -19,25 +19,44 @@
package org.apache.accumulo.test.functional;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.ClientInfo;
+import org.apache.accumulo.core.data.InstanceId;
+import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.core.zookeeper.ZooCache;
+import org.apache.accumulo.core.zookeeper.ZooCache.ZooCacheWatcher;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.test.util.Wait;
import org.apache.commons.io.FileUtils;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher.Event.EventType;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
public class ZooCacheIT extends ConfigurableMacBase {
+ private static final Logger LOG = LoggerFactory.getLogger(ZooCacheIT.class);
+
private static String pathName = "/zcTest-42";
private static File testDir;
@@ -81,4 +100,149 @@ public class ZooCacheIT extends ConfigurableMacBase {
}
}
+ @Test
+ public void testZooKeeperClientChanges() throws Exception {
+
+ try (final AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build();
+ final ClientContext ctx = ((ClientContext) client)) {
+
+ final InstanceId iid = ctx.getInstanceID();
+ final String zooRoot = ZooUtil.getRoot(iid);
+ final String tableName = "test_Table";
+
+ final AtomicBoolean tableCreatedEvent = new AtomicBoolean(false);
+ final AtomicBoolean tableDeletedEvent = new AtomicBoolean(false);
+ final AtomicBoolean connectionOpenEvent = new AtomicBoolean(false);
+ final AtomicBoolean connectionClosedEvent = new AtomicBoolean(false);
+ final AtomicBoolean sessionExpiredEvent = new AtomicBoolean(false);
+ final AtomicReference<String> tableId = new AtomicReference<>();
+ final Supplier<AtomicReference<String>> tableIdSup = () -> tableId;
+
+ final ZooCache cache = ctx.getZooCache();
+ cache.addZooCacheWatcher(new ZooCacheWatcher() {
+
+ @Override
+ @SuppressWarnings("deprecation")
+ public void accept(WatchedEvent event) {
+
+ final String eventPath = event.getPath();
+
+ if (event.getType() != EventType.None
+ && !eventPath.startsWith(zooRoot + Constants.ZTABLES)) {
+ return;
+ }
+ // LOG.info("received event {}", event);
+
+ String tableId = tableIdSup.get().get();
+
+ switch (event.getType()) {
+ case ChildWatchRemoved:
+ case DataWatchRemoved:
+ case NodeChildrenChanged:
+ case NodeDataChanged:
+ case PersistentWatchRemoved:
+ break;
+ case NodeCreated:
+ if (eventPath.equals(zooRoot + Constants.ZTABLES + "/" + tableId)) {
+ LOG.info("Setting tableCreatedEvent");
+ tableCreatedEvent.set(true);
+ }
+ break;
+ case NodeDeleted:
+ if (eventPath.equals(zooRoot + Constants.ZTABLES + "/" + tableId)) {
+ LOG.info("Setting tableDeletedEvent");
+ tableDeletedEvent.set(true);
+ }
+ break;
+ case None:
+ switch (event.getState()) {
+ case AuthFailed:
+ case ConnectedReadOnly:
+ case NoSyncConnected:
+ case SaslAuthenticated:
+ case Unknown:
+ break;
+ case Closed:
+ case Disconnected:
+ LOG.info("Setting connectionClosedEvent");
+ connectionClosedEvent.set(true);
+ break;
+ case Expired:
+ LOG.info("Setting sessionExpiredEvent");
+ sessionExpiredEvent.set(true);
+ break;
+ case SyncConnected:
+ LOG.info("Setting connectionOpenEvent");
+ connectionOpenEvent.set(true);
+ break;
+ default:
+ break;
+
+ }
+ break;
+ default:
+ break;
+
+ }
+ }
+
+ });
+
+ assertFalse(connectionOpenEvent.get()); // missed the event, connection already established
+
+ client.tableOperations().create(tableName);
+ final String tid = ctx.tableOperations().tableIdMap().get(tableName);
+ tableId.set(tid);
+ // we might miss the table created event, don't check for it
+ final String tableZPath = zooRoot + Constants.ZTABLES + "/" + tid;
+ assertFalse(cache.childrenCached(tableZPath));
+ assertNotNull(cache.getChildren(tableZPath));
+ assertTrue(cache.childrenCached(tableZPath));
+
+ client.tableOperations().delete(tableName);
+ Wait.waitFor(() -> tableDeletedEvent.get(), 60_000);
+ assertFalse(cache.childrenCached(tableZPath)); // ZooCache watcher fired and deleted the data
+ assertNull(cache.getChildren(tableZPath));
+
+ assertFalse(connectionOpenEvent.get());
+ assertFalse(connectionClosedEvent.get());
+ tableDeletedEvent.set(false);
+
+ getCluster().getClusterControl().stop(ServerType.ZOOKEEPER);
+
+ Wait.waitFor(() -> connectionClosedEvent.get(), 30_000);
+ connectionClosedEvent.set(false);
+ // Cache should be cleared
+ assertFalse(cache.childrenCached(zooRoot + Constants.ZTABLES));
+
+ getCluster().getClusterControl().start(ServerType.ZOOKEEPER);
+
+ Wait.waitFor(() -> connectionOpenEvent.get(), 60_000);
+ connectionOpenEvent.set(false);
+
+ // Cache should be cleared
+ assertFalse(cache.childrenCached(zooRoot + Constants.ZTABLES));
+
+ // let's assume that are tableId will be one more than the previous (safe assumption)
+ String newTableId = Integer.parseInt(tid) + 1 + "";
+ tableId.set(newTableId);
+ client.tableOperations().create(tableName);
+ Wait.waitFor(() -> tableCreatedEvent.get(), 60_000);
+
+ final String newTableZPath = zooRoot + Constants.ZTABLES + "/" + newTableId;
+ assertFalse(cache.childrenCached(newTableZPath));
+ assertNotNull(cache.getChildren(newTableZPath));
+ assertTrue(cache.childrenCached(newTableZPath));
+
+ client.tableOperations().delete(tableName);
+ Wait.waitFor(() -> tableDeletedEvent.get(), 60_000);
+ assertFalse(cache.childrenCached(newTableZPath)); // ZooCache watcher fired and deleted the
+ // data
+ assertNull(cache.getChildren(newTableZPath));
+
+ assertFalse(connectionOpenEvent.get());
+ assertFalse(connectionClosedEvent.get());
+
+ }
+ }
}
diff --git a/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooCacheIT.java b/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooCacheIT.java
index ef15a30959..ad7bb38f12 100644
--- a/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooCacheIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooCacheIT.java
@@ -20,33 +20,61 @@ package org.apache.accumulo.test.zookeeper;
import static org.apache.accumulo.harness.AccumuloITBase.ZOOKEEPER_TESTING_SERVER;
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.assertTrue;
import java.io.File;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
-import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZooCacheWatcher;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.test.util.Wait;
-import org.apache.zookeeper.Watcher;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import com.github.benmanes.caffeine.cache.Ticker;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
@Tag(ZOOKEEPER_TESTING_SERVER)
public class ZooCacheIT {
+ public static class ZooCacheTicker implements Ticker {
+
+ private int advanceCounter = 0;
+
+ @Override
+ public long read() {
+ return System.nanoTime() + (advanceCounter * ZooCache.CACHE_DURATION.toNanos());
+ }
+
+ public void advance() {
+ advanceCounter++;
+ }
+
+ public void reset() {
+ advanceCounter = 0;
+ }
+
+ }
+
+ public static class TestZooCache extends ZooCache {
+
+ private static final ZooCacheTicker ticker = new ZooCacheTicker();
+
+ public TestZooCache(ZooSession zk, Set<String> pathsToWatch) {
+ super(zk, pathsToWatch, ticker);
+ }
+
+ }
+
private ZooKeeperTestingServer szk;
private ZooSession zk;
private ZooReaderWriter zrw;
@@ -55,6 +83,8 @@ public class ZooCacheIT {
private File tempDir;
@BeforeEach
+ @SuppressFBWarnings(value = "ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD",
+ justification = "setting ticker in test for eviction test")
public void setup() throws Exception {
szk = new ZooKeeperTestingServer(tempDir);
zk = szk.newClient();
@@ -73,132 +103,129 @@ public class ZooCacheIT {
@Test
public void testGetChildren() throws Exception {
- Set<String> watchesRemoved = Collections.synchronizedSet(new HashSet<>());
- ZooCacheWatcher watcher = event -> {
- if (event.getType() == Watcher.Event.EventType.ChildWatchRemoved
- || event.getType() == Watcher.Event.EventType.DataWatchRemoved) {
- watchesRemoved.add(event.getPath());
- }
- };
- ZooCache zooCache = new ZooCache(zk, Optional.of(watcher), Duration.ofSeconds(3));
+ final String root = Constants.ZROOT + UUID.randomUUID().toString();
+ final String base = root + Constants.ZTSERVERS;
+ TestZooCache zooCache = new TestZooCache(zk, Set.of(base));
- zrw.mkdirs("/test2");
- zrw.mkdirs("/test3/c1");
- zrw.mkdirs("/test3/c2");
+ zrw.mkdirs(base + "/test2");
+ zrw.mkdirs(base + "/test3/c1");
+ zrw.mkdirs(base + "/test3/c2");
// cache non-existence of /test1 and existence of /test2 and /test3
long uc1 = zooCache.getUpdateCount();
- assertNull(zooCache.getChildren("/test1"));
+ assertNull(zooCache.getChildren(base + "/test1"));
long uc2 = zooCache.getUpdateCount();
assertTrue(uc1 < uc2);
- assertEquals(List.of(), zooCache.getChildren("/test2"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
long uc3 = zooCache.getUpdateCount();
assertTrue(uc2 < uc3);
- assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren("/test3")));
+ assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren(base + "/test3")));
long uc4 = zooCache.getUpdateCount();
assertTrue(uc3 < uc4);
// The cache should be stable now and new accesses should not change the update count
- assertNull(zooCache.getChildren("/test1"));
+ assertNull(zooCache.getChildren(base + "/test1"));
// once getChildren discovers that a node does not exists, then get data will also know this
- assertNull(zooCache.get("/test1"));
- assertEquals(List.of(), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren("/test3")));
+ assertNull(zooCache.get(base + "/test1"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren(base + "/test3")));
assertEquals(uc4, zooCache.getUpdateCount());
// Had cached non-existence of "/test1", should get a notification that it was created
- zrw.mkdirs("/test1");
+ zrw.mkdirs(base + "/test1");
Wait.waitFor(() -> {
- var children = zooCache.getChildren("/test1");
+ var children = zooCache.getChildren(base + "/test1");
return children != null && children.isEmpty();
});
long uc5 = zooCache.getUpdateCount();
assertTrue(uc4 < uc5);
- assertEquals(List.of(), zooCache.getChildren("/test1"));
- assertEquals(List.of(), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren("/test3")));
- assertEquals(uc5, zooCache.getUpdateCount());
+ assertEquals(List.of(), zooCache.getChildren(base + "/test1"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c2"), Set.copyOf(zooCache.getChildren(base + "/test3")));
+ long uc5b = zooCache.getUpdateCount();
+ assertTrue(uc5 < uc5b);
// add a child to /test3, should get a notification of the change
- zrw.mkdirs("/test3/c3");
+ zrw.mkdirs(base + "/test3/c3");
Wait.waitFor(() -> {
- var children = zooCache.getChildren("/test3");
+ var children = zooCache.getChildren(base + "/test3");
return children != null && children.size() == 3;
});
long uc6 = zooCache.getUpdateCount();
- assertTrue(uc5 < uc6);
- assertEquals(List.of(), zooCache.getChildren("/test1"));
- assertEquals(List.of(), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c2", "c3"), Set.copyOf(zooCache.getChildren("/test3")));
+ assertTrue(uc5b < uc6);
+ assertEquals(List.of(), zooCache.getChildren(base + "/test1"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c2", "c3"), Set.copyOf(zooCache.getChildren(base + "/test3")));
assertEquals(uc6, zooCache.getUpdateCount());
// remove a child from /test3
- zrw.delete("/test3/c2");
+ zrw.delete(base + "/test3/c2");
Wait.waitFor(() -> {
- var children = zooCache.getChildren("/test3");
+ var children = zooCache.getChildren(base + "/test3");
return children != null && children.size() == 2;
});
long uc7 = zooCache.getUpdateCount();
assertTrue(uc6 < uc7);
- assertEquals(List.of(), zooCache.getChildren("/test1"));
- assertEquals(List.of(), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren("/test3")));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test1"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren(base + "/test3")));
assertEquals(uc7, zooCache.getUpdateCount());
// remove /test2, should start caching that it does not exist
- zrw.delete("/test2");
- Wait.waitFor(() -> zooCache.getChildren("/test2") == null);
+ zrw.delete(base + "/test2");
+ Wait.waitFor(() -> zooCache.getChildren(base + "/test2") == null);
long uc8 = zooCache.getUpdateCount();
assertTrue(uc7 < uc8);
- assertEquals(List.of(), zooCache.getChildren("/test1"));
- assertNull(zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren("/test3")));
- assertEquals(uc8, zooCache.getUpdateCount());
+ assertEquals(List.of(), zooCache.getChildren(base + "/test1"));
+ assertNull(zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren(base + "/test3")));
+ long uc8b = zooCache.getUpdateCount();
+ assertTrue(uc8 < uc8b);
// add /test2 back, should update
- zrw.mkdirs("/test2");
- Wait.waitFor(() -> zooCache.getChildren("/test2") != null);
+ zrw.mkdirs(base + "/test2");
+ Wait.waitFor(() -> zooCache.getChildren(base + "/test2") != null);
long uc9 = zooCache.getUpdateCount();
assertTrue(uc8 < uc9);
- assertEquals(List.of(), zooCache.getChildren("/test1"));
- assertEquals(List.of(), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren("/test3")));
- assertEquals(uc9, zooCache.getUpdateCount());
+ assertEquals(List.of(), zooCache.getChildren(base + "/test1"));
+ assertEquals(List.of(), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c1", "c3"), Set.copyOf(zooCache.getChildren(base + "/test3")));
+ long uc9b = zooCache.getUpdateCount();
+ assertTrue(uc9 < uc9b);
// make multiple changes. the cache should see all of these
- zrw.delete("/test1");
- zrw.mkdirs("/test2/ca");
- zrw.delete("/test3/c1");
- zrw.mkdirs("/test3/c4");
- zrw.delete("/test3/c4");
- zrw.mkdirs("/test3/c5");
+ zrw.delete(base + "/test1");
+ zrw.mkdirs(base + "/test2/ca");
+ zrw.delete(base + "/test3/c1");
+ zrw.mkdirs(base + "/test3/c4");
+ zrw.delete(base + "/test3/c4");
+ zrw.mkdirs(base + "/test3/c5");
Wait.waitFor(() -> {
- var children1 = zooCache.getChildren("/test1");
- var children2 = zooCache.getChildren("/test2");
- var children3 = zooCache.getChildren("/test3");
+ var children1 = zooCache.getChildren(base + "/test1");
+ var children2 = zooCache.getChildren(base + "/test2");
+ var children3 = zooCache.getChildren(base + "/test3");
return children1 == null && children2 != null && children2.size() == 1 && children3 != null
&& Set.copyOf(children3).equals(Set.of("c3", "c5"));
});
long uc10 = zooCache.getUpdateCount();
- assertTrue(uc9 < uc10);
- assertNull(zooCache.getChildren("/test1"));
- assertEquals(List.of("ca"), zooCache.getChildren("/test2"));
- assertEquals(Set.of("c3", "c5"), Set.copyOf(zooCache.getChildren("/test3")));
+ assertTrue(uc9b < uc10);
+ assertNull(zooCache.getChildren(base + "/test1"));
+ assertEquals(List.of("ca"), zooCache.getChildren(base + "/test2"));
+ assertEquals(Set.of("c3", "c5"), Set.copyOf(zooCache.getChildren(base + "/test3")));
assertEquals(uc10, zooCache.getUpdateCount());
// wait for the cache to evict and clear watches
+ TestZooCache.ticker.advance();
Wait.waitFor(() -> {
// the cache will not run its eviction handler unless accessed, so access something that is
// not expected to be evicted
- zooCache.getChildren("/test4");
- return watchesRemoved.equals(Set.of("/test1", "/test2", "/test3"));
+ zooCache.getChildren(base + "/test4");
+ return zooCache.childrenCached(base + "/test1") == false
+ && zooCache.childrenCached(base + "/test2") == false
+ && zooCache.childrenCached(base + "/test3") == false;
});
-
- assertFalse(zooCache.childrenCached("/test1"));
- assertFalse(zooCache.childrenCached("/test2"));
- assertFalse(zooCache.childrenCached("/test3"));
}
}