This is an automated email from the ASF dual-hosted git repository. ctubbsii 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 c630630 Improvements to ZooKeeper Lock classes (#1972) c630630 is described below commit c63063081e0f87ae5a61a93bda93046648b1e866 Author: Jeffrey Manno <jeffreymann...@gmail.com> AuthorDate: Thu Mar 18 12:07:26 2021 -0400 Improvements to ZooKeeper Lock classes (#1972) * Create specific types for path parameters for `ZooLock` and `ZooQueueLock` to ensure operations intended for one lock type aren't coded incorrectly against the other lock type's methods (this fixes #1966) * Rename `ZooLock` to `ServiceLock` and `ZooQueueLock` to `FateLock` to make it more clear the purpose of these very similar lock types, including any related classes (this fixes #1967) * Shorten lots of lines by renaming `validateAndSortByLockPrefix` to just `validateAndSort`, using `var` keyword * Remove unused `ephemeral` field (`FateLock` only uses persistent nodes, and the extra parameter when not used makes it harder to reason about the actual implementation) * Guard against null parameters by using `Objects.requireNonNull` in key places * Add javadoc to `FateLock` class and make its fields `final` * Add brief javadoc to `FateLock.validateAndSort` Co-authored-by: Christopher Tubbs <ctubb...@apache.org> --- .../accumulo/core/clientImpl/ClientContext.java | 5 +- .../core/clientImpl/ConditionalWriterImpl.java | 4 +- .../accumulo/core/clientImpl/ServerClient.java | 6 +- .../core/clientImpl/ZookeeperLockChecker.java | 5 +- .../accumulo/core/metadata/schema/Ample.java | 4 +- .../core/metadata/schema/TabletMetadata.java | 8 +- .../java/org/apache/accumulo/fate/AdminUtil.java | 19 ++-- .../zookeeper/{ZooQueueLock.java => FateLock.java} | 56 ++++++---- .../zookeeper/{ZooLock.java => ServiceLock.java} | 90 +++++++++------- .../apache/accumulo/fate/zookeeper/ZooCache.java | 5 +- .../{ZooLockTest.java => ServiceLockTest.java} | 14 +-- .../server/constraints/MetadataConstraints.java | 4 +- .../org/apache/accumulo/server/fs/VolumeUtil.java | 4 +- .../accumulo/server/manager/LiveTServerSet.java | 24 ++--- .../server/metadata/TabletMutatorBase.java | 4 +- .../org/apache/accumulo/server/util/Admin.java | 5 +- .../apache/accumulo/server/util/ListInstances.java | 7 +- .../accumulo/server/util/ManagerMetadataUtil.java | 14 +-- .../accumulo/server/util/MetadataTableUtil.java | 38 +++---- .../accumulo/server/util/TabletServerLocks.java | 7 +- .../org/apache/accumulo/server/util/ZooZap.java | 8 +- .../apache/accumulo/gc/SimpleGarbageCollector.java | 12 +-- .../java/org/apache/accumulo/manager/Manager.java | 17 +-- .../apache/accumulo/manager/tableOps/Utils.java | 7 +- .../manager/tableOps/create/PopulateMetadata.java | 6 +- .../manager/tserverOps/ShutdownTServer.java | 13 ++- .../apache/accumulo/manager/util/FateAdmin.java | 7 +- .../java/org/apache/accumulo/monitor/Monitor.java | 26 ++--- .../org/apache/accumulo/tserver/TabletServer.java | 21 ++-- .../accumulo/tserver/ThriftClientHandler.java | 6 +- .../accumulo/shell/commands/FateCommand.java | 7 +- .../accumulo/test/BadDeleteMarkersCreatedIT.java | 10 +- .../{ZooLockIT.java => ServiceLockIT.java} | 114 +++++++++++---------- .../accumulo/test/functional/BackupManagerIT.java | 6 +- .../test/functional/GarbageCollectorIT.java | 12 +-- .../accumulo/test/functional/ReadWriteIT.java | 10 +- .../apache/accumulo/test/functional/RestartIT.java | 18 ++-- .../accumulo/test/functional/SplitRecoveryIT.java | 16 +-- .../functional/TabletStateChangeIteratorIT.java | 8 +- .../accumulo/test/functional/ZombieTServer.java | 14 +-- .../accumulo/test/replication/ReplicationIT.java | 10 +- .../test/upgrade/GCUpgrade9to10TestIT.java | 6 +- 42 files changed, 371 insertions(+), 306 deletions(-) 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 f20ec82..9af9a10 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 @@ -71,6 +71,7 @@ import org.apache.accumulo.core.securityImpl.thrift.TCredentials; import org.apache.accumulo.core.singletons.SingletonManager; import org.apache.accumulo.core.singletons.SingletonReservation; import org.apache.accumulo.core.util.OpTimer; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; import org.apache.accumulo.fate.zookeeper.ZooCacheFactory; import org.apache.accumulo.fate.zookeeper.ZooUtil; @@ -366,7 +367,7 @@ public class ClientContext implements AccumuloClient { // available only for sharing code with old ZooKeeperInstance public static List<String> getManagerLocations(ZooCache zooCache, String instanceId) { - String managerLocPath = ZooUtil.getRoot(instanceId) + Constants.ZMANAGER_LOCK; + var zLockManagerPath = ServiceLock.path(ZooUtil.getRoot(instanceId) + Constants.ZMANAGER_LOCK); OpTimer timer = null; @@ -375,7 +376,7 @@ public class ClientContext implements AccumuloClient { timer = new OpTimer().start(); } - byte[] loc = zooCache.getLockData(managerLocPath); + byte[] loc = zooCache.getLockData(zLockManagerPath); if (timer != null) { timer.stop(); diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java index 5044aac..15d57ed 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java @@ -78,7 +78,7 @@ import org.apache.accumulo.core.util.ByteBufferUtil; import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.core.util.threads.ThreadPools; import org.apache.accumulo.core.util.threads.Threads; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooUtil.LockID; import org.apache.commons.collections4.map.LRUMap; import org.apache.commons.lang3.mutable.MutableLong; @@ -659,7 +659,7 @@ class ConditionalWriterImpl implements ConditionalWriter { LockID lid = new LockID(context.getZooKeeperRoot() + Constants.ZTSERVERS, sessionId.lockId); while (true) { - if (!ZooLock.isLockHeld(context.getZooCache(), lid)) { + if (!ServiceLock.isLockHeld(context.getZooCache(), lid)) { // ACCUMULO-1152 added a tserver lock check to the tablet location cache, so this // invalidation prevents future attempts to contact the // tserver even its gone zombie and is still running w/o a lock diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ServerClient.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ServerClient.java index 470ad7f..0532179 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ServerClient.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ServerClient.java @@ -35,6 +35,7 @@ import org.apache.accumulo.core.rpc.ThriftUtil; import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; import org.apache.thrift.TApplicationException; import org.apache.thrift.TServiceClient; @@ -143,8 +144,9 @@ public class ServerClient { // add tservers ZooCache zc = context.getZooCache(); for (String tserver : zc.getChildren(context.getZooKeeperRoot() + Constants.ZTSERVERS)) { - String path = context.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + tserver; - byte[] data = zc.getLockData(path); + var zLocPath = + ServiceLock.path(context.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + tserver); + byte[] data = zc.getLockData(zLocPath); if (data != null) { String strData = new String(data, UTF_8); if (!strData.equals("manager")) 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 d451f3e..fccd89c 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 @@ -20,8 +20,8 @@ package org.apache.accumulo.core.clientImpl; import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.clientImpl.TabletLocatorImpl.TabletServerLockChecker; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; public class ZookeeperLockChecker implements TabletServerLockChecker { @@ -35,7 +35,8 @@ public class ZookeeperLockChecker implements TabletServerLockChecker { @Override public boolean isLockHeld(String tserver, String session) { - return ZooLock.getSessionId(zc, root + "/" + tserver) == Long.parseLong(session, 16); + var zLockPath = ServiceLock.path(root + "/" + tserver); + return ServiceLock.getSessionId(zc, zLockPath) == Long.parseLong(session, 16); } @Override diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java index 1860122..28c39a6 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java @@ -32,7 +32,7 @@ import org.apache.accumulo.core.metadata.TabletFile; import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType; import org.apache.accumulo.core.tabletserver.log.LogEntry; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.hadoop.io.Text; /** @@ -248,7 +248,7 @@ public interface Ample { TabletMutator deleteLocation(TServerInstance tserver, LocationType type); - TabletMutator putZooLock(ZooLock zooLock); + TabletMutator putZooLock(ServiceLock zooLock); TabletMutator putDirName(String dirName); 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 ee5283c..a9e86ef 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 @@ -19,7 +19,6 @@ package org.apache.accumulo.core.metadata.schema; import static java.nio.charset.StandardCharsets.UTF_8; -import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SuspendLocationColumn; import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.COMPACT_QUAL; import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_QUAL; import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.FLUSH_QUAL; @@ -67,12 +66,13 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.La import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily; +import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SuspendLocationColumn; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily; import org.apache.accumulo.core.tabletserver.log.LogEntry; import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.core.util.ServerServices; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.hadoop.io.Text; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -471,9 +471,9 @@ public class TabletMetadata { private static Optional<TServerInstance> checkServer(ClientContext context, String path, String zPath) { Optional<TServerInstance> server = Optional.empty(); - final String lockPath = path + "/" + zPath; + final var lockPath = ServiceLock.path(path + "/" + zPath); ZooCache.ZcStat stat = new ZooCache.ZcStat(); - byte[] lockData = ZooLock.getLockData(context.getZooCache(), lockPath, stat); + byte[] lockData = ServiceLock.getLockData(context.getZooCache(), lockPath, stat); log.trace("Checking server at ZK path = " + lockPath); if (lockData != null) { diff --git a/core/src/main/java/org/apache/accumulo/fate/AdminUtil.java b/core/src/main/java/org/apache/accumulo/fate/AdminUtil.java index 12481b1..3016d01 100644 --- a/core/src/main/java/org/apache/accumulo/fate/AdminUtil.java +++ b/core/src/main/java/org/apache/accumulo/fate/AdminUtil.java @@ -31,8 +31,10 @@ import java.util.Map.Entry; import java.util.Set; import org.apache.accumulo.fate.ReadOnlyTStore.TStatus; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooQueueLock; +import org.apache.accumulo.fate.zookeeper.FateLock; +import org.apache.accumulo.fate.zookeeper.FateLock.FateLockPath; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.ServiceLockPath; import org.apache.accumulo.fate.zookeeper.ZooReader; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy; @@ -274,9 +276,9 @@ public class AdminUtil<T> { try { - String path = lockPath + "/" + id; + FateLockPath fLockPath = FateLock.path(lockPath + "/" + id); List<String> lockNodes = - ZooQueueLock.validateAndSortChildrenByLockPrefix(path, zk.getChildren(path)); + FateLock.validateAndSort(fLockPath, zk.getChildren(fLockPath.toString())); int pos = 0; boolean sawWriteLock = false; @@ -413,7 +415,8 @@ public class AdminUtil<T> { } } - public boolean prepDelete(TStore<T> zs, ZooReaderWriter zk, String path, String txidStr) { + public boolean prepDelete(TStore<T> zs, ZooReaderWriter zk, ServiceLockPath path, + String txidStr) { if (!checkGlobalLock(zk, path)) { return false; } @@ -448,7 +451,7 @@ public class AdminUtil<T> { return state; } - public boolean prepFail(TStore<T> zs, ZooReaderWriter zk, String path, String txidStr) { + public boolean prepFail(TStore<T> zs, ZooReaderWriter zk, ServiceLockPath path, String txidStr) { if (!checkGlobalLock(zk, path)) { return false; } @@ -510,9 +513,9 @@ public class AdminUtil<T> { @SuppressFBWarnings(value = "DM_EXIT", justification = "TODO - should probably avoid System.exit here; " + "this code is used by the fate admin shell command") - public boolean checkGlobalLock(ZooReaderWriter zk, String path) { + public boolean checkGlobalLock(ZooReaderWriter zk, ServiceLockPath path) { try { - if (ZooLock.getLockData(zk.getZooKeeper(), path) != null) { + if (ServiceLock.getLockData(zk.getZooKeeper(), path) != null) { System.err.println("ERROR: Manager lock is held, not running"); if (this.exitOnError) System.exit(1); diff --git a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooQueueLock.java b/core/src/main/java/org/apache/accumulo/fate/zookeeper/FateLock.java similarity index 79% rename from core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooQueueLock.java rename to core/src/main/java/org/apache/accumulo/fate/zookeeper/FateLock.java index 3b123ab..987654e 100644 --- a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooQueueLock.java +++ b/core/src/main/java/org/apache/accumulo/fate/zookeeper/FateLock.java @@ -18,6 +18,8 @@ */ package org.apache.accumulo.fate.zookeeper; +import static java.util.Objects.requireNonNull; + import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -33,19 +35,37 @@ import org.apache.zookeeper.KeeperException.NotEmptyException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class ZooQueueLock implements QueueLock { - private static final Logger log = LoggerFactory.getLogger(ZooQueueLock.class); +/** + * A persistent lock mechanism in ZooKeeper used for locking tables during FaTE operations. + */ +public class FateLock implements QueueLock { + private static final Logger log = LoggerFactory.getLogger(FateLock.class); private static final String PREFIX = "flock#"; - private ZooReaderWriter zoo; - private String path; - private boolean ephemeral; + private final ZooReaderWriter zoo; + private final FateLockPath path; + + public static class FateLockPath { + private final String path; - public ZooQueueLock(ZooReaderWriter zrw, String path, boolean ephemeral) { - this.zoo = zrw; - this.path = path; - this.ephemeral = ephemeral; + private FateLockPath(String path) { + this.path = requireNonNull(path); + } + + @Override + public String toString() { + return this.path; + } + } + + public static FateLockPath path(String path) { + return new FateLockPath(path); + } + + public FateLock(ZooReaderWriter zrw, FateLockPath path) { + this.zoo = requireNonNull(zrw); + this.path = requireNonNull(path); } @Override @@ -54,17 +74,13 @@ public class ZooQueueLock implements QueueLock { try { while (true) { try { - if (ephemeral) { - newPath = zoo.putEphemeralSequential(path + "/" + PREFIX, data); - } else { - newPath = zoo.putPersistentSequential(path + "/" + PREFIX, data); - } + newPath = zoo.putPersistentSequential(path + "/" + PREFIX, data); String[] parts = newPath.split("/"); String last = parts[parts.length - 1]; return Long.parseLong(last.substring(PREFIX.length())); } catch (NoNodeException nne) { // the parent does not exist so try to create it - zoo.putPersistentData(path, new byte[] {}, NodeExistsPolicy.SKIP); + zoo.putPersistentData(path.toString(), new byte[] {}, NodeExistsPolicy.SKIP); } } } catch (Exception ex) { @@ -78,7 +94,7 @@ public class ZooQueueLock implements QueueLock { try { List<String> children = Collections.emptyList(); try { - children = zoo.getChildren(path); + children = zoo.getChildren(path.toString()); } catch (KeeperException.NoNodeException ex) { // the path does not exist (it was deleted or not created yet), that is ok there are no // earlier entries then @@ -107,7 +123,7 @@ public class ZooQueueLock implements QueueLock { zoo.recursiveDelete(path + String.format("/%s%010d", PREFIX, entry), NodeMissingPolicy.SKIP); try { // try to delete the parent if it has no children - zoo.delete(path); + zoo.delete(path.toString()); } catch (NotEmptyException nee) { // the path had other lock nodes, no big deal } @@ -116,8 +132,10 @@ public class ZooQueueLock implements QueueLock { } } - public static List<String> validateAndSortChildrenByLockPrefix(String path, - List<String> children) { + /** + * Validate and sort child nodes at this lock path by the lock prefix + */ + public static List<String> validateAndSort(FateLockPath path, List<String> children) { log.trace("validating and sorting children at path {}", path); List<String> validChildren = new ArrayList<>(); if (children == null || children.isEmpty()) { diff --git a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooLock.java b/core/src/main/java/org/apache/accumulo/fate/zookeeper/ServiceLock.java similarity index 90% rename from core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooLock.java rename to core/src/main/java/org/apache/accumulo/fate/zookeeper/ServiceLock.java index 15ff591..22377c8 100644 --- a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooLock.java +++ b/core/src/main/java/org/apache/accumulo/fate/zookeeper/ServiceLock.java @@ -19,6 +19,7 @@ package org.apache.accumulo.fate.zookeeper; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Objects.requireNonNull; import java.util.ArrayList; import java.util.List; @@ -41,8 +42,8 @@ import org.apache.zookeeper.data.Stat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class ZooLock implements Watcher { - private static final Logger LOG = LoggerFactory.getLogger(ZooLock.class); +public class ServiceLock implements Watcher { + private static final Logger LOG = LoggerFactory.getLogger(ServiceLock.class); private static final String ZLOCK_PREFIX = "zlock#"; @@ -60,6 +61,23 @@ public class ZooLock implements Watcher { } + public static class ServiceLockPath { + private final String path; + + private ServiceLockPath(String path) { + this.path = requireNonNull(path); + } + + @Override + public String toString() { + return this.path; + } + } + + public static ServiceLockPath path(String path) { + return new ServiceLockPath(path); + } + public enum LockLossReason { LOCK_DELETED, SESSION_EXPIRED } @@ -79,7 +97,7 @@ public class ZooLock implements Watcher { void failedToAcquireLock(Exception e); } - private final String path; + private final ServiceLockPath path; protected final ZooKeeper zooKeeper; private final Prefix vmLockPrefix; @@ -91,13 +109,13 @@ public class ZooLock implements Watcher { private String createdNodeName; private String watchingNodeName; - public ZooLock(AccumuloConfiguration conf, String path, UUID uuid) { + public ServiceLock(AccumuloConfiguration conf, ServiceLockPath path, UUID uuid) { this.zooKeeper = ZooSession.getAuthenticatedSession(conf.get(Property.INSTANCE_ZK_HOST), (int) conf.getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT), "digest", ("accumulo" + ":" + conf.get(Property.INSTANCE_SECRET)).getBytes(UTF_8)); - this.path = path; + this.path = requireNonNull(path); try { - zooKeeper.exists(path, this); + zooKeeper.exists(path.toString(), this); watchingParent = true; this.vmLockPrefix = new Prefix(ZLOCK_PREFIX + uuid.toString() + "#"); } catch (Exception ex) { @@ -106,11 +124,11 @@ public class ZooLock implements Watcher { } } - protected ZooLock(ZooKeeper zookeeper, String path, UUID uuid) { - this.zooKeeper = zookeeper; - this.path = path; + protected ServiceLock(ZooKeeper zookeeper, ServiceLockPath path, UUID uuid) { + this.zooKeeper = requireNonNull(zookeeper); + this.path = requireNonNull(path); try { - zooKeeper.exists(path, this); + zooKeeper.exists(path.toString(), this); watchingParent = true; this.vmLockPrefix = new Prefix(ZLOCK_PREFIX + uuid.toString() + "#"); } catch (Exception ex) { @@ -179,8 +197,7 @@ public class ZooLock implements Watcher { * list of ephemeral nodes * @return list of ephemeral nodes that have valid formats, sorted by sequence number */ - public static List<String> validateAndSortChildrenByLockPrefix(String path, - List<String> children) { + public static List<String> validateAndSort(ServiceLockPath path, List<String> children) { LOG.trace("validating and sorting children at path {}", path); List<String> validChildren = new ArrayList<>(); if (children == null || children.isEmpty()) { @@ -276,8 +293,7 @@ public class ZooLock implements Watcher { "Called determineLockOwnership() when ephemeralNodeName == null"); } - List<String> children = - validateAndSortChildrenByLockPrefix(path, zooKeeper.getChildren(path, null)); + List<String> children = validateAndSort(path, zooKeeper.getChildren(path.toString(), null)); if (null == children || !children.contains(createdEphemeralNode)) { LOG.error("Expected ephemeral node {} to be in the list of children {}", createdEphemeralNode, @@ -319,7 +335,7 @@ public class ZooLock implements Watcher { if (event.getType() == EventType.NodeDeleted && event.getPath().equals(nodeToWatch)) { LOG.debug("[{}] Detected deletion of prior node {}, attempting to acquire lock", vmLockPrefix, nodeToWatch); - synchronized (ZooLock.this) { + synchronized (ServiceLock.this) { try { if (createdNodeName != null) { determineLockOwnership(createdEphemeralNode, lw); @@ -339,7 +355,7 @@ public class ZooLock implements Watcher { if (event.getState() == KeeperState.Expired || event.getState() == KeeperState.Disconnected) { - synchronized (ZooLock.this) { + synchronized (ServiceLock.this) { if (lockNodeName == null) { LOG.info("Zookeeper Session expired / disconnected"); lw.failedToAcquireLock(new Exception("Zookeeper Session expired / disconnected")); @@ -407,9 +423,9 @@ public class ZooLock implements Watcher { // It's possible that the call above was retried several times and multiple ephemeral nodes // were created but the client missed the response for some reason. Find the ephemeral nodes // with this ZLOCK_UUID and lowest sequential number. - List<String> children = - validateAndSortChildrenByLockPrefix(path, zooKeeper.getChildren(path, null)); - if (null == children || !children.contains(createPath.substring(path.length() + 1))) { + List<String> children = validateAndSort(path, zooKeeper.getChildren(path.toString(), null)); + if (null == children + || !children.contains(createPath.substring(path.toString().length() + 1))) { LOG.error("Expected ephemeral node {} to be in the list of children {}", createPath, children); throw new RuntimeException("Lock attempt ephemeral node no longer exist " + createPath); @@ -462,7 +478,7 @@ public class ZooLock implements Watcher { @Override public void process(WatchedEvent event) { - synchronized (ZooLock.this) { + synchronized (ServiceLock.this) { if (lockNodeName != null && event.getType() == EventType.NodeDeleted && event.getPath().equals(path + "/" + lockNodeName)) { LOG.debug("[{}] {} was deleted", vmLockPrefix, lockNodeName); @@ -507,7 +523,7 @@ public class ZooLock implements Watcher { return; } - createdNodeName = pathForWatcher.substring(path.length() + 1); + createdNodeName = pathForWatcher.substring(path.toString().length() + 1); // We have created a node, do we own the lock? determineLockOwnership(createdNodeName, lw); @@ -577,7 +593,7 @@ public class ZooLock implements Watcher { if (lockNodeName == null) { throw new IllegalStateException("Lock not held"); } - return new LockID(path, lockNodeName, zooKeeper.getSessionId()); + return new LockID(path.toString(), lockNodeName, zooKeeper.getSessionId()); } /** @@ -612,7 +628,7 @@ public class ZooLock implements Watcher { } else { try { // set the watch on the parent node again - zooKeeper.exists(path, this); + zooKeeper.exists(path.toString(), this); watchingParent = true; } catch (KeeperException.ConnectionLossException ex) { // we can't look at the lock because we aren't connected, but our session is still good @@ -631,7 +647,8 @@ public class ZooLock implements Watcher { public static boolean isLockHeld(ZooCache zc, LockID lid) { - List<String> children = validateAndSortChildrenByLockPrefix(lid.path, zc.getChildren(lid.path)); + var zLockPath = path(lid.path); + List<String> children = validateAndSort(zLockPath, zc.getChildren(zLockPath.toString())); if (children == null || children.isEmpty()) { return false; @@ -645,10 +662,10 @@ public class ZooLock implements Watcher { return zc.get(lid.path + "/" + lid.node, stat) != null && stat.getEphemeralOwner() == lid.eid; } - public static byte[] getLockData(ZooKeeper zk, String path) + public static byte[] getLockData(ZooKeeper zk, ServiceLockPath path) throws KeeperException, InterruptedException { - List<String> children = validateAndSortChildrenByLockPrefix(path, zk.getChildren(path, false)); + List<String> children = validateAndSort(path, zk.getChildren(path.toString(), null)); if (children == null || children.isEmpty()) { return null; @@ -659,10 +676,10 @@ public class ZooLock implements Watcher { return zk.getData(path + "/" + lockNode, false, null); } - public static byte[] getLockData(org.apache.accumulo.fate.zookeeper.ZooCache zc, String path, - ZcStat stat) { + public static byte[] getLockData(org.apache.accumulo.fate.zookeeper.ZooCache zc, + ServiceLockPath path, ZcStat stat) { - List<String> children = validateAndSortChildrenByLockPrefix(path, zc.getChildren(path)); + List<String> children = validateAndSort(path, zc.getChildren(path.toString())); if (children == null || children.isEmpty()) { return null; @@ -677,9 +694,9 @@ public class ZooLock implements Watcher { return zc.get(path + "/" + lockNode, stat); } - public static long getSessionId(ZooCache zc, String path) { + public static long getSessionId(ZooCache zc, ServiceLockPath path) { - List<String> children = validateAndSortChildrenByLockPrefix(path, zc.getChildren(path)); + List<String> children = validateAndSort(path, zc.getChildren(path.toString())); if (children == null || children.isEmpty()) { return 0; @@ -695,8 +712,7 @@ public class ZooLock implements Watcher { public long getSessionId() throws KeeperException, InterruptedException { - List<String> children = - validateAndSortChildrenByLockPrefix(path, zooKeeper.getChildren(path, null)); + List<String> children = validateAndSort(path, zooKeeper.getChildren(path.toString(), null)); String lockNode = children.get(0); @@ -708,10 +724,10 @@ public class ZooLock implements Watcher { } } - public static void deleteLock(ZooReaderWriter zk, String path) + public static void deleteLock(ZooReaderWriter zk, ServiceLockPath path) throws InterruptedException, KeeperException { - List<String> children = validateAndSortChildrenByLockPrefix(path, zk.getChildren(path)); + List<String> children = validateAndSort(path, zk.getChildren(path.toString())); if (children == null || children.isEmpty()) { throw new IllegalStateException("No lock is held at " + path); @@ -729,10 +745,10 @@ public class ZooLock implements Watcher { } - public static boolean deleteLock(ZooReaderWriter zk, String path, String lockData) + public static boolean deleteLock(ZooReaderWriter zk, ServiceLockPath path, String lockData) throws InterruptedException, KeeperException { - List<String> children = validateAndSortChildrenByLockPrefix(path, zk.getChildren(path)); + List<String> children = validateAndSort(path, zk.getChildren(path.toString())); if (children == null || children.isEmpty()) { throw new IllegalStateException("No lock is held at " + path); diff --git a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java b/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java index 0301d96..2a0ced3 100644 --- a/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java +++ b/core/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java @@ -31,6 +31,7 @@ import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.ServiceLockPath; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.WatchedEvent; @@ -565,8 +566,8 @@ public class ZooCache { } } - public byte[] getLockData(String path) { - List<String> children = ZooLock.validateAndSortChildrenByLockPrefix(path, getChildren(path)); + public byte[] getLockData(ServiceLockPath path) { + List<String> children = ServiceLock.validateAndSort(path, getChildren(path.toString())); if (children == null || children.isEmpty()) { return null; } diff --git a/core/src/test/java/org/apache/accumulo/fate/zookeeper/ZooLockTest.java b/core/src/test/java/org/apache/accumulo/fate/zookeeper/ServiceLockTest.java similarity index 89% rename from core/src/test/java/org/apache/accumulo/fate/zookeeper/ZooLockTest.java rename to core/src/test/java/org/apache/accumulo/fate/zookeeper/ServiceLockTest.java index 8d46c09..d7c3c18 100644 --- a/core/src/test/java/org/apache/accumulo/fate/zookeeper/ZooLockTest.java +++ b/core/src/test/java/org/apache/accumulo/fate/zookeeper/ServiceLockTest.java @@ -26,7 +26,7 @@ import java.util.List; import org.junit.Test; -public class ZooLockTest { +public class ServiceLockTest { @Test public void testSortAndFindLowestPrevPrefix() throws Exception { @@ -42,7 +42,7 @@ public class ZooLockTest { children.add("zlock#987654321"); children.add("zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001"); - final List<String> validChildren = ZooLock.validateAndSortChildrenByLockPrefix("", children); + final List<String> validChildren = ServiceLock.validateAndSort(ServiceLock.path(""), children); assertEquals(8, validChildren.size()); assertEquals("zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001", validChildren.get(0)); @@ -55,24 +55,24 @@ public class ZooLockTest { assertEquals("zlock#00000000-0000-0000-0000-eeeeeeeeeeee#0000000010", validChildren.get(7)); assertEquals("zlock#00000000-0000-0000-0000-bbbbbbbbbbbb#0000000004", - ZooLock.findLowestPrevPrefix(validChildren, + ServiceLock.findLowestPrevPrefix(validChildren, "zlock#00000000-0000-0000-0000-ffffffffffff#0000000007")); assertEquals("zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001", - ZooLock.findLowestPrevPrefix(validChildren, + ServiceLock.findLowestPrevPrefix(validChildren, "zlock#00000000-0000-0000-0000-cccccccccccc#0000000003")); assertEquals("zlock#00000000-0000-0000-0000-dddddddddddd#0000000008", - ZooLock.findLowestPrevPrefix(validChildren, + ServiceLock.findLowestPrevPrefix(validChildren, "zlock#00000000-0000-0000-0000-eeeeeeeeeeee#0000000010")); assertThrows(IndexOutOfBoundsException.class, () -> { - ZooLock.findLowestPrevPrefix(validChildren, + ServiceLock.findLowestPrevPrefix(validChildren, "zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001"); }); assertThrows(IndexOutOfBoundsException.class, () -> { - ZooLock.findLowestPrevPrefix(validChildren, + ServiceLock.findLowestPrevPrefix(validChildren, "zlock#00000000-0000-0000-0000-XXXXXXXXXXXX#0000000099"); }); } diff --git a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java index c9acb53..1476430 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java +++ b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java @@ -49,8 +49,8 @@ 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.util.ColumnFQ; import org.apache.accumulo.core.util.cleaner.CleanerUtil; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.zookeeper.TransactionWatcher.Arbitrator; @@ -287,7 +287,7 @@ public class MetadataConstraints implements Constraint { String lockId = new String(columnUpdate.getValue(), UTF_8); try { - lockHeld = ZooLock.isLockHeld(zooCache, new ZooUtil.LockID(zooRoot, lockId)); + lockHeld = ServiceLock.isLockHeld(zooCache, new ZooUtil.LockID(zooRoot, lockId)); } catch (Exception e) { log.debug("Failed to verify lock was held {} {}", lockId, e.getMessage()); } diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java index 736bc2a..95eb497 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java @@ -31,7 +31,7 @@ import org.apache.accumulo.core.metadata.schema.DataFileValue; import org.apache.accumulo.core.protobuf.ProtobufUtil; import org.apache.accumulo.core.tabletserver.log.LogEntry; import org.apache.accumulo.core.util.Pair; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.server.ServerConstants; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.fs.VolumeManager.FileType; @@ -135,7 +135,7 @@ public class VolumeUtil { * configured in instance.volumes.replacements. Second, if a tablet dir is no longer configured * for use it chooses a new tablet directory. */ - public static TabletFiles updateTabletVolumes(ServerContext context, ZooLock zooLock, + public static TabletFiles updateTabletVolumes(ServerContext context, ServiceLock zooLock, KeyExtent extent, TabletFiles tabletFiles, boolean replicate) { List<Pair<Path,Path>> replacements = ServerConstants.getVolumeReplacements(context.getConfiguration(), context.getHadoopConf()); 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 8e85250..0f03600 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 @@ -45,9 +45,9 @@ import org.apache.accumulo.core.util.Halt; import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.threads.ThreadPools; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; import org.apache.accumulo.fate.zookeeper.ZooCache.ZcStat; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.server.ServerContext; import org.apache.hadoop.io.Text; import org.apache.thrift.TException; @@ -78,17 +78,17 @@ public class LiveTServerSet implements Watcher { address = addr; } - private String lockString(ZooLock mlock) { + private String lockString(ServiceLock mlock) { return mlock.getLockID().serialize(context.getZooKeeperRoot() + Constants.ZMANAGER_LOCK); } - private void loadTablet(TabletClientService.Client client, ZooLock lock, KeyExtent extent) + private void loadTablet(TabletClientService.Client client, ServiceLock lock, KeyExtent extent) throws TException { client.loadTablet(TraceUtil.traceInfo(), context.rpcCreds(), lockString(lock), extent.toThrift()); } - public void assignTablet(ZooLock lock, KeyExtent extent) throws TException { + public void assignTablet(ServiceLock lock, KeyExtent extent) throws TException { if (extent.isMeta()) { // see ACCUMULO-3597 try (TTransport transport = ThriftUtil.createTransport(address, context)) { @@ -107,7 +107,7 @@ public class LiveTServerSet implements Watcher { } } - public void unloadTablet(ZooLock lock, KeyExtent extent, TUnloadTabletGoal goal, + public void unloadTablet(ServiceLock lock, KeyExtent extent, TUnloadTabletGoal goal, long requestTime) throws TException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); @@ -139,7 +139,7 @@ public class LiveTServerSet implements Watcher { } } - public void halt(ZooLock lock) throws TException, ThriftSecurityException { + public void halt(ServiceLock lock) throws TException, ThriftSecurityException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); try { @@ -149,7 +149,7 @@ public class LiveTServerSet implements Watcher { } } - public void fastHalt(ZooLock lock) throws TException { + public void fastHalt(ServiceLock lock) throws TException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); try { @@ -159,7 +159,7 @@ public class LiveTServerSet implements Watcher { } } - public void flush(ZooLock lock, TableId tableId, byte[] startRow, byte[] endRow) + public void flush(ServiceLock lock, TableId tableId, byte[] startRow, byte[] endRow) throws TException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); @@ -172,7 +172,7 @@ public class LiveTServerSet implements Watcher { } } - public void chop(ZooLock lock, KeyExtent extent) throws TException { + public void chop(ServiceLock lock, KeyExtent extent) throws TException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); try { @@ -194,7 +194,7 @@ public class LiveTServerSet implements Watcher { } } - public void compact(ZooLock lock, String tableId, byte[] startRow, byte[] endRow) + public void compact(ServiceLock lock, String tableId, byte[] startRow, byte[] endRow) throws TException { TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(), address, context); @@ -293,9 +293,9 @@ public class LiveTServerSet implements Watcher { TServerInfo info = current.get(zPath); - final String lockPath = path + "/" + zPath; + final var zLockPath = ServiceLock.path(path + "/" + zPath); ZcStat stat = new ZcStat(); - byte[] lockData = ZooLock.getLockData(getZooCache(), lockPath, stat); + byte[] lockData = ServiceLock.getLockData(getZooCache(), zLockPath, stat); if (lockData == null) { if (info != null) { diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java index fda24bd..b9ee684 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java @@ -42,7 +42,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataTime; import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType; import org.apache.accumulo.core.tabletserver.log.LogEntry; import org.apache.accumulo.fate.FateTxId; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.server.ServerContext; import org.apache.hadoop.io.Text; @@ -155,7 +155,7 @@ public abstract class TabletMutatorBase implements Ample.TabletMutator { } @Override - public Ample.TabletMutator putZooLock(ZooLock zooLock) { + public Ample.TabletMutator putZooLock(ServiceLock zooLock) { Preconditions.checkState(updatesEnabled, "Cannot make updates after calling mutate."); ServerColumnFamily.LOCK_COLUMN.put(mutation, new Value(zooLock.getLockID().serialize(context.getZooKeeperRoot() + "/"))); 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 333cbce..95383db 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 @@ -58,8 +58,8 @@ import org.apache.accumulo.core.singletons.SingletonManager.Mode; import org.apache.accumulo.core.trace.TraceUtil; import org.apache.accumulo.core.util.AddressUtil; import org.apache.accumulo.core.util.HostAndPort; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.cli.ServerUtilOpts; import org.apache.accumulo.server.security.SecurityUtil; @@ -414,7 +414,8 @@ public class Admin implements KeywordExecutable { */ static String qualifyWithZooKeeperSessionId(String zTServerRoot, ZooCache zooCache, String hostAndPort) { - long sessionId = ZooLock.getSessionId(zooCache, zTServerRoot + "/" + hostAndPort); + var zLockPath = ServiceLock.path(zTServerRoot + "/" + hostAndPort); + long sessionId = ServiceLock.getSessionId(zooCache, zLockPath); if (sessionId == 0) { return hostAndPort; } diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java b/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java index 14ba4cd..dbed4c6 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java @@ -32,8 +32,8 @@ import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.cli.Help; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.conf.SiteConfiguration; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooReader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -167,8 +167,9 @@ public class ListInstances { } try { - String managerLocPath = Constants.ZROOT + "/" + iid + Constants.ZMANAGER_LOCK; - byte[] manager = ZooLock.getLockData(cache, managerLocPath, null); + var zLockManagerPath = + ServiceLock.path(Constants.ZROOT + "/" + iid + Constants.ZMANAGER_LOCK); + byte[] manager = ServiceLock.getLockData(cache, zLockManagerPath, null); if (manager == null) { return null; } diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ManagerMetadataUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/ManagerMetadataUtil.java index ab07c06..48ca88d 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/ManagerMetadataUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/ManagerMetadataUtil.java @@ -52,7 +52,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataTime; import org.apache.accumulo.core.metadata.schema.TabletMetadata; import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType; import org.apache.accumulo.core.security.Authorizations; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.server.ServerContext; import org.apache.hadoop.io.Text; import org.apache.zookeeper.KeeperException; @@ -66,7 +66,7 @@ public class ManagerMetadataUtil { public static void addNewTablet(ServerContext context, KeyExtent extent, String dirName, TServerInstance location, Map<StoredTabletFile,DataFileValue> datafileSizes, Map<Long,? extends Collection<TabletFile>> bulkLoadedFiles, MetadataTime time, - long lastFlushID, long lastCompactID, ZooLock zooLock) { + long lastFlushID, long lastCompactID, ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putPrevEndRow(extent.prevEndRow()); @@ -96,7 +96,7 @@ public class ManagerMetadataUtil { tablet.mutate(); } - public static KeyExtent fixSplit(ServerContext context, TabletMetadata meta, ZooLock lock) + public static KeyExtent fixSplit(ServerContext context, TabletMetadata meta, ServiceLock lock) throws AccumuloException { log.info("Incomplete split {} attempting to fix", meta.getExtent()); @@ -115,7 +115,7 @@ public class ManagerMetadataUtil { } private static KeyExtent fixSplit(ServerContext context, TableId tableId, Text metadataEntry, - Text metadataPrevEndRow, Text oper, double splitRatio, ZooLock lock) + Text metadataPrevEndRow, Text oper, double splitRatio, ServiceLock lock) throws AccumuloException { if (metadataPrevEndRow == null) // something is wrong, this should not happen... if a tablet is split, it will always have a @@ -168,7 +168,7 @@ public class ManagerMetadataUtil { } } - private static TServerInstance getTServerInstance(String address, ZooLock zooLock) { + private static TServerInstance getTServerInstance(String address, ServiceLock zooLock) { while (true) { try { return new TServerInstance(address, zooLock.getSessionId()); @@ -182,7 +182,7 @@ public class ManagerMetadataUtil { public static void replaceDatafiles(ServerContext context, KeyExtent extent, Set<StoredTabletFile> datafilesToDelete, Set<StoredTabletFile> scanFiles, TabletFile path, Long compactionId, DataFileValue size, String address, TServerInstance lastLocation, - ZooLock zooLock) { + ServiceLock zooLock) { context.getAmple().putGcCandidates(extent.tableId(), datafilesToDelete); @@ -217,7 +217,7 @@ public class ManagerMetadataUtil { * */ public static StoredTabletFile updateTabletDataFile(ServerContext context, KeyExtent extent, - TabletFile path, DataFileValue dfv, MetadataTime time, String address, ZooLock zooLock, + TabletFile path, DataFileValue dfv, MetadataTime time, String address, ServiceLock zooLock, Set<String> unusedWalLogs, TServerInstance lastLocation, long flushId) { TabletMutator tablet = context.getAmple().mutateTablet(extent); diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java index 5e9ff18..27c53ac 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java @@ -91,7 +91,7 @@ import org.apache.accumulo.core.util.ColumnFQ; import org.apache.accumulo.core.util.FastFormat; import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.fate.FateTxId; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.gc.GcVolumeUtil; import org.apache.hadoop.io.Text; @@ -132,7 +132,7 @@ public class MetadataTableUtil { return rootTable; } - public static void putLockID(ServerContext context, ZooLock zooLock, Mutation m) { + public static void putLockID(ServerContext context, ServiceLock zooLock, Mutation m) { ServerColumnFamily.LOCK_COLUMN.put(m, new Value(zooLock.getLockID().serialize(context.getZooKeeperRoot() + "/"))); } @@ -141,12 +141,13 @@ public class MetadataTableUtil { update(context, null, m, extent); } - public static void update(ServerContext context, ZooLock zooLock, Mutation m, KeyExtent extent) { + public static void update(ServerContext context, ServiceLock zooLock, Mutation m, + KeyExtent extent) { Writer t = extent.isMeta() ? getRootTable(context) : getMetadataTable(context); update(context, t, zooLock, m); } - public static void update(ServerContext context, Writer t, ZooLock zooLock, Mutation m) { + public static void update(ServerContext context, Writer t, ServiceLock zooLock, Mutation m) { if (zooLock != null) putLockID(context, zooLock, m); while (true) { @@ -165,7 +166,7 @@ public class MetadataTableUtil { } public static void updateTabletFlushID(KeyExtent extent, long flushID, ServerContext context, - ZooLock zooLock) { + ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putFlushId(flushID); tablet.putZooLock(zooLock); @@ -173,7 +174,7 @@ public class MetadataTableUtil { } public static void updateTabletCompactID(KeyExtent extent, long compactID, ServerContext context, - ZooLock zooLock) { + ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putCompactionId(compactID); tablet.putZooLock(zooLock); @@ -182,7 +183,7 @@ public class MetadataTableUtil { public static Map<StoredTabletFile,DataFileValue> updateTabletDataFile(long tid, KeyExtent extent, Map<TabletFile,DataFileValue> estSizes, MetadataTime time, ServerContext context, - ZooLock zooLock) { + ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putTime(time); @@ -198,7 +199,7 @@ public class MetadataTableUtil { } public static void updateTabletDir(KeyExtent extent, String newDir, ServerContext context, - ZooLock zooLock) { + ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putDirName(newDir); tablet.putZooLock(zooLock); @@ -206,7 +207,7 @@ public class MetadataTableUtil { } public static void addTablet(KeyExtent extent, String path, ServerContext context, - TimeType timeType, ZooLock zooLock) { + TimeType timeType, ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putPrevEndRow(extent.prevEndRow()); tablet.putDirName(path); @@ -218,7 +219,7 @@ public class MetadataTableUtil { public static void updateTabletVolumes(KeyExtent extent, List<LogEntry> logsToRemove, List<LogEntry> logsToAdd, List<StoredTabletFile> filesToRemove, - SortedMap<TabletFile,DataFileValue> filesToAdd, ZooLock zooLock, ServerContext context) { + SortedMap<TabletFile,DataFileValue> filesToAdd, ServiceLock zooLock, ServerContext context) { TabletMutator tabletMutator = context.getAmple().mutateTablet(extent); logsToRemove.forEach(tabletMutator::deleteWal); @@ -233,7 +234,7 @@ public class MetadataTableUtil { } public static void rollBackSplit(Text metadataEntry, Text oldPrevEndRow, ServerContext context, - ZooLock zooLock) { + ServiceLock zooLock) { KeyExtent ke = KeyExtent.fromMetaRow(metadataEntry, oldPrevEndRow); Mutation m = TabletColumnFamily.createPrevRowMutation(ke); TabletColumnFamily.SPLIT_RATIO_COLUMN.putDelete(m); @@ -242,7 +243,7 @@ public class MetadataTableUtil { } public static void splitTablet(KeyExtent extent, Text oldPrevEndRow, double splitRatio, - ServerContext context, ZooLock zooLock) { + ServerContext context, ServiceLock zooLock) { Mutation m = TabletColumnFamily.createPrevRowMutation(extent); TabletColumnFamily.SPLIT_RATIO_COLUMN.put(m, new Value(Double.toString(splitRatio))); @@ -255,7 +256,8 @@ public class MetadataTableUtil { public static void finishSplit(Text metadataEntry, Map<StoredTabletFile,DataFileValue> datafileSizes, - List<StoredTabletFile> highDatafilesToRemove, final ServerContext context, ZooLock zooLock) { + List<StoredTabletFile> highDatafilesToRemove, final ServerContext context, + ServiceLock zooLock) { Mutation m = new Mutation(metadataEntry); TabletColumnFamily.SPLIT_RATIO_COLUMN.putDelete(m); TabletColumnFamily.OLD_PREV_ROW_COLUMN.putDelete(m); @@ -275,12 +277,12 @@ public class MetadataTableUtil { public static void finishSplit(KeyExtent extent, Map<StoredTabletFile,DataFileValue> datafileSizes, - List<StoredTabletFile> highDatafilesToRemove, ServerContext context, ZooLock zooLock) { + List<StoredTabletFile> highDatafilesToRemove, ServerContext context, ServiceLock zooLock) { finishSplit(extent.toMetaRow(), datafileSizes, highDatafilesToRemove, context, zooLock); } public static void removeScanFiles(KeyExtent extent, Set<StoredTabletFile> scanFiles, - ServerContext context, ZooLock zooLock) { + ServerContext context, ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); scanFiles.forEach(tablet::deleteScan); tablet.putZooLock(zooLock); @@ -339,7 +341,7 @@ public class MetadataTableUtil { } public static void deleteTable(TableId tableId, boolean insertDeletes, ServerContext context, - ZooLock lock) throws AccumuloException { + ServiceLock lock) throws AccumuloException { try (Scanner ms = new ScannerImpl(context, MetadataTable.ID, Authorizations.EMPTY); BatchWriter bw = new BatchWriterImpl(context, MetadataTable.ID, new BatchWriterConfig().setMaxMemory(1000000) @@ -417,7 +419,7 @@ public class MetadataTableUtil { } public static void removeUnusedWALEntries(ServerContext context, KeyExtent extent, - final List<LogEntry> entries, ZooLock zooLock) { + final List<LogEntry> entries, ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); entries.forEach(tablet::deleteWal); tablet.putZooLock(zooLock); @@ -630,7 +632,7 @@ public class MetadataTableUtil { } } - public static void chopped(ServerContext context, KeyExtent extent, ZooLock zooLock) { + public static void chopped(ServerContext context, KeyExtent extent, ServiceLock zooLock) { TabletMutator tablet = context.getAmple().mutateTablet(extent); tablet.putChopped(); tablet.putZooLock(zooLock); 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 b608501..b68dbd8 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 @@ -25,8 +25,8 @@ import java.util.List; import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.cli.Help; import org.apache.accumulo.core.conf.SiteConfiguration; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.server.ServerContext; @@ -56,7 +56,8 @@ public class TabletServerLocks { List<String> tabletServers = zoo.getChildren(tserverPath); for (String tabletServer : tabletServers) { - byte[] lockData = ZooLock.getLockData(cache, tserverPath + "/" + tabletServer, null); + var zLockPath = ServiceLock.path(tserverPath + "/" + tabletServer); + byte[] lockData = ServiceLock.getLockData(cache, zLockPath, null); String holder = null; if (lockData != null) { holder = new String(lockData, UTF_8); @@ -65,7 +66,7 @@ public class TabletServerLocks { System.out.printf("%32s %16s%n", tabletServer, holder); } } else if (opts.delete != null) { - ZooLock.deleteLock(zoo, tserverPath + "/" + args[1]); + ServiceLock.deleteLock(zoo, ServiceLock.path(tserverPath + "/" + args[1])); } else { System.out.println( "Usage : " + TabletServerLocks.class.getName() + " -list|-delete <tserver lock>"); diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java b/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java index d88c85d..5b561c6 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java @@ -27,7 +27,7 @@ import org.apache.accumulo.core.conf.SiteConfiguration; import org.apache.accumulo.core.singletons.SingletonManager; import org.apache.accumulo.core.singletons.SingletonManager.Mode; import org.apache.accumulo.core.volume.VolumeConfiguration; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy; import org.apache.accumulo.server.fs.VolumeManager; @@ -109,9 +109,9 @@ public class ZooZap { if (opts.zapManager || opts.zapMaster) { zoo.recursiveDelete(tserversPath + "/" + child, NodeMissingPolicy.SKIP); } else { - String path = tserversPath + "/" + child; - if (!zoo.getChildren(path).isEmpty()) { - if (!ZooLock.deleteLock(zoo, path, "tserver")) { + var zLockPath = ServiceLock.path(tserversPath + "/" + child); + if (!zoo.getChildren(zLockPath.toString()).isEmpty()) { + if (!ServiceLock.deleteLock(zoo, zLockPath, "tserver")) { message("Did not delete " + tserversPath + "/" + child, opts); } } diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java index 5208181..f314d95 100644 --- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java +++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java @@ -76,9 +76,9 @@ import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; import org.apache.accumulo.core.util.threads.ThreadPools; import org.apache.accumulo.core.volume.Volume; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher; import org.apache.accumulo.gc.metrics.GcCycleMetrics; import org.apache.accumulo.gc.metrics.GcMetricsFactory; import org.apache.accumulo.gc.replication.CloseWriteAheadLogReferences; @@ -118,7 +118,7 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface { private static final Logger log = LoggerFactory.getLogger(SimpleGarbageCollector.class); - private ZooLock lock; + private ServiceLock lock; private GCStatus status = new GCStatus(new GcCycleStats(), new GcCycleStats(), new GcCycleStats(), new GcCycleStats()); @@ -617,7 +617,7 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface { } private void getZooLock(HostAndPort addr) throws KeeperException, InterruptedException { - String path = getContext().getZooKeeperRoot() + Constants.ZGC_LOCK; + var path = ServiceLock.path(getContext().getZooKeeperRoot() + Constants.ZGC_LOCK); LockWatcher lockWatcher = new LockWatcher() { @Override @@ -635,7 +635,7 @@ public class SimpleGarbageCollector extends AbstractServer implements Iface { UUID zooLockUUID = UUID.randomUUID(); while (true) { - lock = new ZooLock(getContext().getSiteConfiguration(), path, zooLockUUID); + lock = new ServiceLock(getContext().getSiteConfiguration(), path, zooLockUUID); if (lock.tryLock(lockWatcher, new ServerServices(addr.toString(), Service.GC_CLIENT).toString().getBytes())) { log.debug("Got GC ZooKeeper lock"); 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 08c546b..db0deb6 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 @@ -97,8 +97,9 @@ import org.apache.accumulo.core.util.threads.Threads; import org.apache.accumulo.fate.AgeOffStore; import org.apache.accumulo.fate.Fate; import org.apache.accumulo.fate.util.Retry; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.ServiceLockPath; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; @@ -201,7 +202,7 @@ public class Manager extends AbstractServer private ZooAuthenticationKeyDistributor keyDistributor; private AuthenticationTokenKeyManager authenticationTokenKeyManager; - ZooLock managerLock = null; + ServiceLock managerLock = null; private TServer clientService = null; private volatile TabletBalancer tabletBalancer; private final BalancerEnvironment balancerEnvironment; @@ -1042,7 +1043,7 @@ public class Manager extends AbstractServer // block until we can obtain the ZK lock for the manager try { - getManagerLock(zroot + Constants.ZMANAGER_LOCK); + getManagerLock(ServiceLock.path(zroot + Constants.ZMANAGER_LOCK)); } catch (KeeperException | InterruptedException e) { throw new IllegalStateException("Exception getting manager lock", e); } @@ -1378,11 +1379,11 @@ public class Manager extends AbstractServer return Math.max(1, deadline - System.currentTimeMillis()); } - public ZooLock getManagerLock() { + public ServiceLock getManagerLock() { return managerLock; } - private static class ManagerLockWatcher implements ZooLock.AccumuloLockWatcher { + private static class ManagerLockWatcher implements ServiceLock.AccumuloLockWatcher { boolean acquiredLock = false; boolean failedToAcquireLock = false; @@ -1439,7 +1440,7 @@ public class Manager extends AbstractServer } } - private void getManagerLock(final String zManagerLoc) + private void getManagerLock(final ServiceLockPath zManagerLoc) throws KeeperException, InterruptedException { ServerContext context = getContext(); log.info("trying to get manager lock"); @@ -1451,7 +1452,7 @@ public class Manager extends AbstractServer while (true) { ManagerLockWatcher managerLockWatcher = new ManagerLockWatcher(); - managerLock = new ZooLock(context.getSiteConfiguration(), zManagerLoc, zooLockUUID); + managerLock = new ServiceLock(context.getSiteConfiguration(), zManagerLoc, zooLockUUID); managerLock.lock(managerLockWatcher, managerClientAddress.getBytes()); managerLockWatcher.waitForChange(); diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/Utils.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/Utils.java index 86dfbf1..6440e3c 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/Utils.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/Utils.java @@ -39,7 +39,7 @@ import org.apache.accumulo.core.data.AbstractId; import org.apache.accumulo.core.data.NamespaceId; import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.fate.zookeeper.DistributedReadWriteLock; -import org.apache.accumulo.fate.zookeeper.ZooQueueLock; +import org.apache.accumulo.fate.zookeeper.FateLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooReservation; import org.apache.accumulo.manager.Manager; @@ -156,8 +156,9 @@ public class Utils { private static Lock getLock(ServerContext context, AbstractId<?> id, long tid, boolean writeLock) { byte[] lockData = String.format("%016x", tid).getBytes(UTF_8); - ZooQueueLock qlock = new ZooQueueLock(context.getZooReaderWriter(), - context.getZooKeeperRoot() + Constants.ZTABLE_LOCKS + "/" + id.canonical(), false); + var fLockPath = + FateLock.path(context.getZooKeeperRoot() + Constants.ZTABLE_LOCKS + "/" + id.canonical()); + FateLock qlock = new FateLock(context.getZooReaderWriter(), fLockPath); Lock lock = DistributedReadWriteLock.recoverLock(qlock, lockData); if (lock == null) { DistributedReadWriteLock locker = new DistributedReadWriteLock(qlock, lockData); diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/PopulateMetadata.java b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/PopulateMetadata.java index b9398f3..e4daf3a 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/PopulateMetadata.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/PopulateMetadata.java @@ -36,7 +36,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Se import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily; import org.apache.accumulo.core.metadata.schema.MetadataTime; import org.apache.accumulo.fate.Repo; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.manager.Manager; import org.apache.accumulo.manager.tableOps.ManagerRepo; import org.apache.accumulo.manager.tableOps.TableInfo; @@ -82,8 +82,8 @@ class PopulateMetadata extends ManagerRepo { } private void writeSplitsToMetadataTable(ServerContext ctx, TableId tableId, - SortedSet<Text> splits, Map<Text,Text> data, TimeType timeType, ZooLock lock, BatchWriter bw) - throws MutationsRejectedException { + SortedSet<Text> splits, Map<Text,Text> data, TimeType timeType, ServiceLock lock, + BatchWriter bw) throws MutationsRejectedException { Text prevSplit = null; Value dirValue; for (Text split : Iterables.concat(splits, Collections.singleton(null))) { diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java index cb1a0c0..bd33415 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java @@ -25,7 +25,7 @@ import org.apache.accumulo.core.master.thrift.TabletServerStatus; import org.apache.accumulo.core.metadata.TServerInstance; import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.fate.Repo; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; import org.apache.accumulo.manager.Manager; @@ -92,10 +92,13 @@ public class ShutdownTServer extends ManagerRepo { // suppress assignment of tablets to the server if (force) { ZooReaderWriter zoo = manager.getContext().getZooReaderWriter(); - String path = manager.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + hostAndPort; - ZooLock.deleteLock(zoo, path); - path = manager.getZooKeeperRoot() + Constants.ZDEADTSERVERS + "/" + hostAndPort; - zoo.putPersistentData(path, "forced down".getBytes(UTF_8), NodeExistsPolicy.OVERWRITE); + var path = + ServiceLock.path(manager.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + hostAndPort); + ServiceLock.deleteLock(zoo, path); + path = ServiceLock + .path(manager.getZooKeeperRoot() + Constants.ZDEADTSERVERS + "/" + hostAndPort); + zoo.putPersistentData(path.toString(), "forced down".getBytes(UTF_8), + NodeExistsPolicy.OVERWRITE); } return null; diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/util/FateAdmin.java b/server/manager/src/main/java/org/apache/accumulo/manager/util/FateAdmin.java index 3b620bc..d485327 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/util/FateAdmin.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/util/FateAdmin.java @@ -29,6 +29,7 @@ import org.apache.accumulo.core.conf.SiteConfiguration; import org.apache.accumulo.fate.AdminUtil; import org.apache.accumulo.fate.ReadOnlyStore; import org.apache.accumulo.fate.ZooStore; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.manager.Manager; import org.apache.accumulo.server.ServerContext; @@ -82,19 +83,19 @@ public class FateAdmin { try (var context = new ServerContext(SiteConfiguration.auto())) { final String zkRoot = context.getZooKeeperRoot(); String path = zkRoot + Constants.ZFATE; - String managerPath = zkRoot + Constants.ZMANAGER_LOCK; + var zLockManagerPath = ServiceLock.path(zkRoot + Constants.ZMANAGER_LOCK); ZooReaderWriter zk = context.getZooReaderWriter(); ZooStore<Manager> zs = new ZooStore<>(path, zk); if (jc.getParsedCommand().equals("fail")) { for (String txid : txOpts.get(jc.getParsedCommand()).txids) { - if (!admin.prepFail(zs, zk, managerPath, txid)) { + if (!admin.prepFail(zs, zk, zLockManagerPath, txid)) { System.exit(1); } } } else if (jc.getParsedCommand().equals("delete")) { for (String txid : txOpts.get(jc.getParsedCommand()).txids) { - if (!admin.prepDelete(zs, zk, managerPath, txid)) { + if (!admin.prepDelete(zs, zk, zLockManagerPath, txid)) { System.exit(1); } admin.deleteLocks(zk, zkRoot + Constants.ZTABLE_LOCKS, txid); diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java index c1053b3..fdafca6 100644 --- a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java +++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java @@ -62,8 +62,8 @@ import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; import org.apache.accumulo.core.util.threads.Threads; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy; @@ -165,7 +165,7 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { private EmbeddedWebServer server; - private ZooLock monitorLock; + private ServiceLock monitorLock; private class EventCounter { @@ -391,8 +391,8 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { try { // Read the gc location from its lock ZooReaderWriter zk = context.getZooReaderWriter(); - String path = context.getZooKeeperRoot() + Constants.ZGC_LOCK; - List<String> locks = ZooLock.validateAndSortChildrenByLockPrefix(path, zk.getChildren(path)); + var path = ServiceLock.path(context.getZooKeeperRoot() + Constants.ZGC_LOCK); + List<String> locks = ServiceLock.validateAndSort(path, zk.getChildren(path.toString())); if (locks != null && !locks.isEmpty()) { address = new ServerServices(new String(zk.getData(path + "/" + locks.get(0)), UTF_8)) .getAddress(Service.GC_CLIENT); @@ -598,7 +598,7 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { ServerContext context = getContext(); final String zRoot = context.getZooKeeperRoot(); final String monitorPath = zRoot + Constants.ZMONITOR; - final String monitorLockPath = zRoot + Constants.ZMONITOR_LOCK; + final var monitorLockPath = ServiceLock.path(zRoot + Constants.ZMONITOR_LOCK); // Ensure that everything is kosher with ZK as this has changed. ZooReaderWriter zoo = context.getZooReaderWriter(); @@ -611,18 +611,18 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { // And then make the nodes that we expect for the incoming ephemeral nodes zoo.putPersistentData(monitorPath, new byte[0], NodeExistsPolicy.FAIL); - zoo.putPersistentData(monitorLockPath, new byte[0], NodeExistsPolicy.FAIL); - } else if (!zoo.exists(monitorLockPath)) { + zoo.putPersistentData(monitorLockPath.toString(), new byte[0], NodeExistsPolicy.FAIL); + } else if (!zoo.exists(monitorLockPath.toString())) { // monitor node in ZK exists and is empty as we expect // but the monitor/lock node does not - zoo.putPersistentData(monitorLockPath, new byte[0], NodeExistsPolicy.FAIL); + zoo.putPersistentData(monitorLockPath.toString(), new byte[0], NodeExistsPolicy.FAIL); } } else { // 1.5.0 and earlier zoo.putPersistentData(zRoot + Constants.ZMONITOR, new byte[0], NodeExistsPolicy.FAIL); - if (!zoo.exists(monitorLockPath)) { + if (!zoo.exists(monitorLockPath.toString())) { // Somehow the monitor node exists but not monitor/lock - zoo.putPersistentData(monitorLockPath, new byte[0], NodeExistsPolicy.FAIL); + zoo.putPersistentData(monitorLockPath.toString(), new byte[0], NodeExistsPolicy.FAIL); } } @@ -630,7 +630,7 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { UUID zooLockUUID = UUID.randomUUID(); while (true) { MoniterLockWatcher monitorLockWatcher = new MoniterLockWatcher(); - monitorLock = new ZooLock(context.getSiteConfiguration(), monitorLockPath, zooLockUUID); + monitorLock = new ServiceLock(context.getSiteConfiguration(), monitorLockPath, zooLockUUID); monitorLock.lock(monitorLockWatcher, new byte[0]); monitorLockWatcher.waitForChange(); @@ -656,7 +656,7 @@ public class Monitor extends AbstractServer implements HighlyAvailableService { /** * Async Watcher for monitor lock */ - private static class MoniterLockWatcher implements ZooLock.AccumuloLockWatcher { + private static class MoniterLockWatcher implements ServiceLock.AccumuloLockWatcher { boolean acquiredLock = false; boolean failedToAcquireLock = false; 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 e6b46b6..f67297e 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 @@ -93,10 +93,10 @@ import org.apache.accumulo.core.util.threads.Threads; import org.apache.accumulo.fate.util.Retry; import org.apache.accumulo.fate.util.Retry.RetryFactory; import org.apache.accumulo.fate.util.UtilWaitThread; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; import org.apache.accumulo.server.AbstractServer; @@ -213,7 +213,7 @@ public class TabletServer extends AbstractServer { private volatile boolean serverStopRequested = false; private volatile boolean shutdownComplete = false; - private ZooLock tabletServerLock; + private ServiceLock tabletServerLock; private TServer server; private volatile TServer replServer; @@ -620,18 +620,18 @@ public class TabletServer extends AbstractServer { } } - public ZooLock getLock() { + public ServiceLock getLock() { return tabletServerLock; } private void announceExistence() { ZooReaderWriter zoo = getContext().getZooReaderWriter(); try { - String zPath = - getContext().getZooKeeperRoot() + Constants.ZTSERVERS + "/" + getClientAddressString(); + var zLockPath = ServiceLock.path( + getContext().getZooKeeperRoot() + Constants.ZTSERVERS + "/" + getClientAddressString()); try { - zoo.putPersistentData(zPath, new byte[] {}, NodeExistsPolicy.SKIP); + zoo.putPersistentData(zLockPath.toString(), new byte[] {}, NodeExistsPolicy.SKIP); } catch (KeeperException e) { if (e.code() == KeeperException.Code.NOAUTH) { log.error("Failed to write to ZooKeeper. Ensure that" @@ -640,7 +640,8 @@ public class TabletServer extends AbstractServer { throw e; } - tabletServerLock = new ZooLock(getContext().getSiteConfiguration(), zPath, UUID.randomUUID()); + tabletServerLock = + new ServiceLock(getContext().getSiteConfiguration(), zLockPath, UUID.randomUUID()); LockWatcher lw = new LockWatcher() { @@ -664,7 +665,7 @@ public class TabletServer extends AbstractServer { byte[] lockContent = new ServerServices(getClientAddressString(), Service.TSERV_CLIENT) .toString().getBytes(UTF_8); for (int i = 0; i < 120 / 5; i++) { - zoo.putPersistentData(zPath, new byte[0], NodeExistsPolicy.SKIP); + zoo.putPersistentData(zLockPath.toString(), new byte[0], NodeExistsPolicy.SKIP); if (tabletServerLock.tryLock(lw, lockContent)) { log.debug("Obtained tablet server lock {}", tabletServerLock.getLockPath()); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java index 90a837f..2ab4f21 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java @@ -115,7 +115,7 @@ import org.apache.accumulo.core.util.ByteBufferUtil; import org.apache.accumulo.core.util.Halt; import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.core.util.threads.Threads; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.server.client.ClientServiceHandler; import org.apache.accumulo.server.conf.TableConfiguration; @@ -1365,11 +1365,11 @@ class ThriftClientHandler extends ClientServiceHandler implements TabletClientSe new ZooUtil.LockID(context.getZooKeeperRoot() + Constants.ZMANAGER_LOCK, lock); try { - if (!ZooLock.isLockHeld(server.managerLockCache, lid)) { + if (!ServiceLock.isLockHeld(server.managerLockCache, lid)) { // maybe the cache is out of date and a new manager holds the // lock? server.managerLockCache.clear(); - if (!ZooLock.isLockHeld(server.managerLockCache, lid)) { + if (!ServiceLock.isLockHeld(server.managerLockCache, 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/shell/src/main/java/org/apache/accumulo/shell/commands/FateCommand.java b/shell/src/main/java/org/apache/accumulo/shell/commands/FateCommand.java index 0dbad53..62ad71d 100644 --- a/shell/src/main/java/org/apache/accumulo/shell/commands/FateCommand.java +++ b/shell/src/main/java/org/apache/accumulo/shell/commands/FateCommand.java @@ -41,6 +41,7 @@ import org.apache.accumulo.fate.ReadOnlyRepo; import org.apache.accumulo.fate.ReadOnlyTStore.TStatus; import org.apache.accumulo.fate.Repo; import org.apache.accumulo.fate.ZooStore; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.shell.Shell; import org.apache.accumulo.shell.Shell.Command; @@ -130,7 +131,7 @@ public class FateCommand extends Command { AdminUtil<FateCommand> admin = new AdminUtil<>(false); String path = context.getZooKeeperRoot() + Constants.ZFATE; - String managerPath = context.getZooKeeperRoot() + Constants.ZMANAGER_LOCK; + var managerLockPath = ServiceLock.path(context.getZooKeeperRoot() + Constants.ZMANAGER_LOCK); ZooReaderWriter zk = getZooReaderWriter(context, siteConfig, cl.getOptionValue(secretOption.getOpt())); ZooStore<FateCommand> zs = new ZooStore<>(path, zk); @@ -140,7 +141,7 @@ public class FateCommand extends Command { throw new ParseException("Must provide transaction ID"); } for (int i = 1; i < args.length; i++) { - if (!admin.prepFail(zs, zk, managerPath, args[i])) { + if (!admin.prepFail(zs, zk, managerLockPath, args[i])) { System.out.printf("Could not fail transaction: %s%n", args[i]); failedCommand = true; } @@ -150,7 +151,7 @@ public class FateCommand extends Command { throw new ParseException("Must provide transaction ID"); } for (int i = 1; i < args.length; i++) { - if (admin.prepDelete(zs, zk, managerPath, args[i])) { + if (admin.prepDelete(zs, zk, managerLockPath, args[i])) { admin.deleteLocks(zk, context.getZooKeeperRoot() + Constants.ZTABLE_LOCKS, args[i]); } else { System.out.printf("Could not delete transaction: %s%n", args[i]); 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 3a78097..ae88fc0 100644 --- a/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java +++ b/test/src/main/java/org/apache/accumulo/test/BadDeleteMarkersCreatedIT.java @@ -41,8 +41,8 @@ import org.apache.accumulo.core.data.Value; import org.apache.accumulo.core.metadata.MetadataTable; import org.apache.accumulo.core.metadata.schema.MetadataSchema.DeletesSection; import org.apache.accumulo.core.security.Authorizations; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.minicluster.ServerType; @@ -103,11 +103,11 @@ public class BadDeleteMarkersCreatedIT extends AccumuloClusterHarness { ClientInfo info = ClientInfo.from(client.properties()); ZooCache zcache = new ZooCache(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()); zcache.clear(); - String path = - ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK; + var path = ServiceLock + .path(ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK); byte[] gcLockData; do { - gcLockData = ZooLock.getLockData(zcache, path, null); + gcLockData = ServiceLock.getLockData(zcache, path, null); if (gcLockData != null) { log.info("Waiting for GC ZooKeeper lock to expire"); Thread.sleep(2000); @@ -121,7 +121,7 @@ public class BadDeleteMarkersCreatedIT extends AccumuloClusterHarness { gcLockData = null; do { - gcLockData = ZooLock.getLockData(zcache, path, null); + gcLockData = ServiceLock.getLockData(zcache, path, null); if (gcLockData == null) { log.info("Waiting for GC ZooKeeper lock to be acquired"); Thread.sleep(2000); diff --git a/test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ZooLockIT.java b/test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ServiceLockIT.java similarity index 82% rename from test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ZooLockIT.java rename to test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ServiceLockIT.java index 06dd0b0..221ef52 100644 --- a/test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ZooLockIT.java +++ b/test/src/main/java/org/apache/accumulo/test/fate/zookeeper/ServiceLockIT.java @@ -38,9 +38,10 @@ import java.util.concurrent.locks.LockSupport; import org.apache.accumulo.core.conf.ConfigurationCopy; import org.apache.accumulo.core.conf.Property; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.AccumuloLockWatcher; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.AccumuloLockWatcher; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.ServiceLockPath; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.test.categories.ZooKeeperTestingServerTests; import org.apache.accumulo.test.zookeeper.ZooKeeperTestingServer; @@ -60,7 +61,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; @Category({ZooKeeperTestingServerTests.class}) -public class ZooLockIT { +public class ServiceLockIT { private static ZooKeeperTestingServer szk = null; @@ -98,9 +99,9 @@ public class ZooLockIT { } - private static class ZooLockWrapper extends ZooLock { + private static class ServiceLockWrapper extends ServiceLock { - protected ZooLockWrapper(ZooKeeper zookeeper, String path, UUID uuid) { + protected ServiceLockWrapper(ZooKeeper zookeeper, ServiceLockPath path, UUID uuid) { super(zookeeper, path, uuid); } @@ -189,34 +190,35 @@ public class ZooLockIT { private static final AtomicInteger pdCount = new AtomicInteger(0); - private static ZooLock getZooLock(String parent, UUID uuid) { + private static ServiceLock getZooLock(ServiceLockPath parent, UUID uuid) { Map<String,String> props = new HashMap<>(); props.put(Property.INSTANCE_ZK_HOST.toString(), szk.getConn()); props.put(Property.INSTANCE_ZK_TIMEOUT.toString(), "30000"); props.put(Property.INSTANCE_SECRET.toString(), "secret"); - return new ZooLock(new ConfigurationCopy(props), parent, uuid); + return new ServiceLock(new ConfigurationCopy(props), parent, uuid); } - private static ZooLock getZooLock(ZooKeeperWrapper zkw, String parent, UUID uuid) { - return new ZooLockWrapper(zkw, parent, uuid); + private static ServiceLock getZooLock(ZooKeeperWrapper zkw, ServiceLockPath parent, UUID uuid) { + return new ServiceLockWrapper(zkw, parent, uuid); } @Test(timeout = 10000) public void testDeleteParent() throws Exception { - String parent = "/zltestDeleteParent-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = ServiceLock + .path("/zltestDeleteParent-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); assertFalse(zl.isLocked()); ZooReaderWriter zk = new ZooReaderWriter(szk.getConn(), 30000, "secret"); // intentionally created parent after lock - zk.mkdirs(parent); + zk.mkdirs(parent.toString()); - zk.delete(parent); + zk.delete(parent.toString()); - zk.mkdirs(parent); + zk.mkdirs(parent.toString()); TestALW lw = new TestALW(); @@ -234,9 +236,10 @@ public class ZooLockIT { @Test(timeout = 10000) public void testNoParent() throws Exception { - String parent = "/zltestNoParent-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = + ServiceLock.path("/zltestNoParent-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); assertFalse(zl.isLocked()); @@ -254,12 +257,13 @@ public class ZooLockIT { @Test(timeout = 10000) public void testDeleteLock() throws Exception { - String parent = "/zltestDeleteLock-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = + ServiceLock.path("/zltestDeleteLock-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); ZooReaderWriter zk = new ZooReaderWriter(szk.getConn(), 30000, "secret"); - zk.mkdirs(parent); + zk.mkdirs(parent.toString()); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); assertFalse(zl.isLocked()); @@ -285,12 +289,13 @@ public class ZooLockIT { @Test(timeout = 15000) public void testDeleteWaiting() throws Exception { - String parent = "/zltestDeleteWaiting-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = ServiceLock + .path("/zltestDeleteWaiting-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); ZooReaderWriter zk = new ZooReaderWriter(szk.getConn(), 30000, "secret"); - zk.mkdirs(parent); + zk.mkdirs(parent.toString()); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); assertFalse(zl.isLocked()); @@ -305,7 +310,7 @@ public class ZooLockIT { assertNull(lw.exception); assertNull(lw.reason); - ZooLock zl2 = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl2 = getZooLock(parent, UUID.randomUUID()); TestALW lw2 = new TestALW(); @@ -314,14 +319,13 @@ public class ZooLockIT { assertFalse(lw2.locked); assertFalse(zl2.isLocked()); - ZooLock zl3 = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl3 = getZooLock(parent, UUID.randomUUID()); TestALW lw3 = new TestALW(); zl3.lock(lw3, "test3".getBytes(UTF_8)); - List<String> children = - ZooLock.validateAndSortChildrenByLockPrefix(parent, zk.getChildren(parent)); + List<String> children = ServiceLock.validateAndSort(parent, zk.getChildren(parent.toString())); zk.delete(parent + "/" + children.get(1)); @@ -351,7 +355,8 @@ public class ZooLockIT { @Test(timeout = 10000) public void testUnexpectedEvent() throws Exception { - String parent = "/zltestUnexpectedEvent-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = ServiceLock + .path("/zltestUnexpectedEvent-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); ConnectedWatcher watcher = new ConnectedWatcher(); try (ZooKeeper zk = new ZooKeeper(szk.getConn(), 30000, watcher)) { @@ -361,14 +366,14 @@ public class ZooLockIT { Thread.sleep(200); } - zk.create(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); assertFalse(zl.isLocked()); // would not expect data to be set on this node, but it should not cause problems..... - zk.setData(parent, "foo".getBytes(UTF_8), -1); + zk.setData(parent.toString(), "foo".getBytes(UTF_8), -1); TestALW lw = new TestALW(); @@ -396,7 +401,7 @@ public class ZooLockIT { @Test(timeout = 60000) public void testLockSerial() throws Exception { - String parent = "/zlretryLockSerial"; + var parent = ServiceLock.path("/zlretryLockSerial"); ConnectedWatcher watcher1 = new ConnectedWatcher(); ConnectedWatcher watcher2 = new ConnectedWatcher(); @@ -415,10 +420,11 @@ public class ZooLockIT { } // Create the parent node - zk1.createOnce(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk1.createOnce(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); final RetryLockWatcher zlw1 = new RetryLockWatcher(); - ZooLock zl1 = + ServiceLock zl1 = getZooLock(zk1, parent, UUID.fromString("00000000-0000-0000-0000-aaaaaaaaaaaa")); zl1.lock(zlw1, "test1".getBytes(UTF_8)); // The call above creates two nodes in ZK because of the overridden create method in @@ -433,7 +439,7 @@ public class ZooLockIT { // zl1 assumes that it has the lock. final RetryLockWatcher zlw2 = new RetryLockWatcher(); - ZooLock zl2 = + ServiceLock zl2 = getZooLock(zk2, parent, UUID.fromString("00000000-0000-0000-0000-bbbbbbbbbbbb")); zl2.lock(zlw2, "test1".getBytes(UTF_8)); // The call above creates two nodes in ZK because of the overridden create method in @@ -453,7 +459,7 @@ public class ZooLockIT { assertTrue(zlw1.isLockHeld()); assertFalse(zlw2.isLockHeld()); - List<String> children = zk1.getChildren(parent, false); + List<String> children = zk1.getChildren(parent.toString(), false); assertTrue(children.contains("zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000000")); assertFalse("this node should have been deleted", children.contains("zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001")); @@ -485,7 +491,7 @@ public class ZooLockIT { private static final Logger LOG = LoggerFactory.getLogger(LockWorker.class); - private final String parent; + private final ServiceLockPath parent; private final UUID uuid; private final CountDownLatch getLockLatch; private final CountDownLatch lockCompletedLatch; @@ -493,7 +499,7 @@ public class ZooLockIT { private final RetryLockWatcher lockWatcher = new RetryLockWatcher(); private volatile Exception ex = null; - public LockWorker(final String parent, final UUID uuid, final CountDownLatch lockLatch, + public LockWorker(final ServiceLockPath parent, final UUID uuid, final CountDownLatch lockLatch, final CountDownLatch lockCompletedLatch) { this.parent = parent; this.uuid = uuid; @@ -518,7 +524,7 @@ public class ZooLockIT { while (!watcher.isConnected()) { Thread.sleep(50); } - ZooLock zl = getZooLock(zk, parent, uuid); + ServiceLock zl = getZooLock(zk, parent, uuid); getLockLatch.countDown(); // signal we are done getLockLatch.await(); // wait for others to finish zl.lock(lockWatcher, "test1".getBytes(UTF_8)); // race to the lock @@ -559,7 +565,7 @@ public class ZooLockIT { @Test(timeout = 60000) public void testLockParallel() throws Exception { - String parent = "/zlParallel"; + var parent = ServiceLock.path("/zlParallel"); ConnectedWatcher watcher = new ConnectedWatcher(); try (ZooKeeperWrapper zk = new ZooKeeperWrapper(szk.getConn(), 30000, watcher)) { @@ -569,7 +575,8 @@ public class ZooLockIT { Thread.sleep(50); } // Create the parent node - zk.createOnce(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.createOnce(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); int numWorkers = 4; final CountDownLatch getLockLatch = new CountDownLatch(numWorkers); @@ -594,10 +601,10 @@ public class ZooLockIT { for (int i = 4; i > 0; i--) { List<String> children = - ZooLock.validateAndSortChildrenByLockPrefix(parent, zk.getChildren(parent, false)); + ServiceLock.validateAndSort(parent, zk.getChildren(parent.toString(), null)); while (children.size() != i) { Thread.sleep(100); - children = zk.getChildren(parent, false); + children = zk.getChildren(parent.toString(), false); } assertEquals(i, children.size()); String first = children.get(0); @@ -615,7 +622,7 @@ public class ZooLockIT { workers.forEach(w -> assertFalse(w.holdsLock())); workers.forEach(w -> assertNull(w.getException())); - assertEquals(0, zk.getChildren(parent, false).size()); + assertEquals(0, zk.getChildren(parent.toString(), false).size()); threads.forEach(t -> { try { @@ -630,9 +637,10 @@ public class ZooLockIT { @Test(timeout = 10000) public void testTryLock() throws Exception { - String parent = "/zltestTryLock-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = + ServiceLock.path("/zltestTryLock-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); ConnectedWatcher watcher = new ConnectedWatcher(); try (ZooKeeper zk = new ZooKeeper(szk.getConn(), 30000, watcher)) { @@ -643,11 +651,12 @@ public class ZooLockIT { } for (int i = 0; i < 10; i++) { - zk.create(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - zk.delete(parent, -1); + zk.create(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); + zk.delete(parent.toString(), -1); } - zk.create(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); TestALW lw = new TestALW(); @@ -668,7 +677,8 @@ public class ZooLockIT { @Test(timeout = 10000) public void testChangeData() throws Exception { - String parent = "/zltestChangeData-" + this.hashCode() + "-l" + pdCount.incrementAndGet(); + var parent = + ServiceLock.path("/zltestChangeData-" + this.hashCode() + "-l" + pdCount.incrementAndGet()); ConnectedWatcher watcher = new ConnectedWatcher(); try (ZooKeeper zk = new ZooKeeper(szk.getConn(), 30000, watcher)) { zk.addAuthInfo("digest", "accumulo:secret".getBytes(UTF_8)); @@ -677,9 +687,9 @@ public class ZooLockIT { Thread.sleep(200); } - zk.create(parent, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.create(parent.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); - ZooLock zl = getZooLock(parent, UUID.randomUUID()); + ServiceLock zl = getZooLock(parent, UUID.randomUUID()); TestALW lw = new TestALW(); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BackupManagerIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BackupManagerIT.java index 9368346..e84a492 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BackupManagerIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BackupManagerIT.java @@ -25,7 +25,7 @@ import org.apache.accumulo.core.client.Accumulo; import org.apache.accumulo.core.client.AccumuloClient; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.fate.util.UtilWaitThread; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy; import org.apache.accumulo.manager.Manager; @@ -52,8 +52,8 @@ public class BackupManagerIT extends ConfigurableMacBase { // wait for 2 lock entries do { UtilWaitThread.sleep(100); - String path = root + Constants.ZMANAGER_LOCK; - children = ZooLock.validateAndSortChildrenByLockPrefix(path, writer.getChildren(path)); + var path = ServiceLock.path(root + Constants.ZMANAGER_LOCK); + children = ServiceLock.validateAndSort(path, writer.getChildren(path.toString())); } while (children.size() != 2); // wait for the backup manager to learn to be the backup UtilWaitThread.sleep(1000); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorIT.java b/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorIT.java index 05116ac..f0ac9c1 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorIT.java @@ -50,7 +50,7 @@ import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.security.TablePermission; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.gc.SimpleGarbageCollector; @@ -100,10 +100,10 @@ public class GarbageCollectorIT extends ConfigurableMacBase { getCluster().killProcess(ServerType.GARBAGE_COLLECTOR, getCluster().getProcesses().get(ServerType.GARBAGE_COLLECTOR).iterator().next()); // delete lock in zookeeper if there, this will allow next GC to start quickly - String path = getServerContext().getZooKeeperRoot() + Constants.ZGC_LOCK; + var path = ServiceLock.path(getServerContext().getZooKeeperRoot() + Constants.ZGC_LOCK); ZooReaderWriter zk = new ZooReaderWriter(cluster.getZooKeepers(), 30000, OUR_SECRET); try { - ZooLock.deleteLock(zk, path); + ServiceLock.deleteLock(zk, path); } catch (IllegalStateException e) { log.error("Unable to delete ZooLock for mini accumulo-gc", e); } @@ -256,12 +256,12 @@ public class GarbageCollectorIT extends ConfigurableMacBase { try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) { ZooReaderWriter zk = new ZooReaderWriter(cluster.getZooKeepers(), 30000, OUR_SECRET); - String path = - ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK; + var path = ServiceLock + .path(ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK); for (int i = 0; i < 5; i++) { List<String> locks; try { - locks = ZooLock.validateAndSortChildrenByLockPrefix(path, zk.getChildren(path)); + locks = ServiceLock.validateAndSort(path, zk.getChildren(path.toString())); } catch (NoNodeException e) { Thread.sleep(5000); continue; 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 de61d76..04003b6 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 @@ -73,8 +73,8 @@ import org.apache.accumulo.core.metadata.MetadataTable; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily; import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.util.MonitorUtil; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooReader; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.harness.AccumuloClusterHarness; @@ -170,12 +170,12 @@ public class ReadWriteIT extends AccumuloClusterHarness { ClientInfo info = ClientInfo.from(accumuloClient.properties()); ZooReader zreader = new ZooReader(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()); ZooCache zcache = new ZooCache(zreader, null); + var zLockPath = + ServiceLock.path(ZooUtil.getRoot(accumuloClient.instanceOperations().getInstanceID()) + + Constants.ZMANAGER_LOCK); byte[] managerLockData; do { - managerLockData = ZooLock.getLockData(zcache, - ZooUtil.getRoot(accumuloClient.instanceOperations().getInstanceID()) - + Constants.ZMANAGER_LOCK, - null); + managerLockData = ServiceLock.getLockData(zcache, zLockPath, null); if (managerLockData != null) { log.info("Manager lock is still held"); Thread.sleep(1000); 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 198277a..579720f 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 @@ -37,8 +37,8 @@ import org.apache.accumulo.core.clientImpl.ClientInfo; import org.apache.accumulo.core.conf.ClientProperty; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.metadata.MetadataTable; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooReader; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.harness.AccumuloClusterHarness; @@ -141,11 +141,11 @@ public class RestartIT extends AccumuloClusterHarness { ClientInfo info = ClientInfo.from(c.properties()); ZooReader zreader = new ZooReader(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()); ZooCache zcache = new ZooCache(zreader, null); + var zLockPath = ServiceLock + .path(ZooUtil.getRoot(c.instanceOperations().getInstanceID()) + Constants.ZMANAGER_LOCK); byte[] managerLockData; do { - managerLockData = ZooLock.getLockData(zcache, - ZooUtil.getRoot(c.instanceOperations().getInstanceID()) + Constants.ZMANAGER_LOCK, - null); + managerLockData = ServiceLock.getLockData(zcache, zLockPath, null); if (managerLockData != null) { log.info("Manager lock is still held"); Thread.sleep(1000); @@ -158,9 +158,7 @@ public class RestartIT extends AccumuloClusterHarness { managerLockData = new byte[0]; do { - managerLockData = ZooLock.getLockData(zcache, - ZooUtil.getRoot(c.instanceOperations().getInstanceID()) + Constants.ZMANAGER_LOCK, - null); + managerLockData = ServiceLock.getLockData(zcache, zLockPath, null); if (managerLockData != null) { log.info("Manager lock is still held"); Thread.sleep(1000); @@ -196,11 +194,11 @@ public class RestartIT extends AccumuloClusterHarness { ClientInfo info = ClientInfo.from(c.properties()); ZooReader zreader = new ZooReader(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()); ZooCache zcache = new ZooCache(zreader, null); + var zLockPath = ServiceLock + .path(ZooUtil.getRoot(c.instanceOperations().getInstanceID()) + Constants.ZMANAGER_LOCK); byte[] managerLockData; do { - managerLockData = ZooLock.getLockData(zcache, - ZooUtil.getRoot(c.instanceOperations().getInstanceID()) + Constants.ZMANAGER_LOCK, - null); + managerLockData = ServiceLock.getLockData(zcache, zLockPath, null); if (managerLockData != null) { log.info("Manager lock is still held"); Thread.sleep(1000); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java b/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java index 5d7e48a..92986d2 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/SplitRecoveryIT.java @@ -62,9 +62,9 @@ import org.apache.accumulo.core.metadata.schema.TabletMetadata; import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType; import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.util.ColumnFQ; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; import org.apache.accumulo.server.ServerConstants; @@ -92,10 +92,10 @@ public class SplitRecoveryIT extends ConfigurableMacBase { } private void run(ServerContext c) throws Exception { - String zPath = c.getZooKeeperRoot() + "/testLock"; + var zPath = ServiceLock.path(c.getZooKeeperRoot() + "/testLock"); ZooReaderWriter zoo = c.getZooReaderWriter(); - zoo.putPersistentData(zPath, new byte[0], NodeExistsPolicy.OVERWRITE); - ZooLock zl = new ZooLock(c.getSiteConfiguration(), zPath, UUID.randomUUID()); + zoo.putPersistentData(zPath.toString(), new byte[0], NodeExistsPolicy.OVERWRITE); + ServiceLock zl = new ServiceLock(c.getSiteConfiguration(), zPath, UUID.randomUUID()); boolean gotLock = zl.tryLock(new LockWatcher() { @SuppressFBWarnings(value = "DM_EXIT", @@ -148,7 +148,7 @@ public class SplitRecoveryIT extends ConfigurableMacBase { } private void runSplitRecoveryTest(ServerContext context, int failPoint, String mr, - int extentToSplit, ZooLock zl, KeyExtent... extents) throws Exception { + int extentToSplit, ServiceLock zl, KeyExtent... extents) throws Exception { Text midRow = new Text(mr); @@ -196,7 +196,7 @@ public class SplitRecoveryIT extends ConfigurableMacBase { private void splitPartiallyAndRecover(ServerContext context, KeyExtent extent, KeyExtent high, KeyExtent low, double splitRatio, SortedMap<StoredTabletFile,DataFileValue> mapFiles, - Text midRow, String location, int steps, ZooLock zl) throws Exception { + Text midRow, String location, int steps, ServiceLock zl) throws Exception { SortedMap<StoredTabletFile,DataFileValue> lowDatafileSizes = new TreeMap<>(); SortedMap<StoredTabletFile,DataFileValue> highDatafileSizes = new TreeMap<>(); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/TabletStateChangeIteratorIT.java b/test/src/main/java/org/apache/accumulo/test/functional/TabletStateChangeIteratorIT.java index f7597fa..e981ab2 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/TabletStateChangeIteratorIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/TabletStateChangeIteratorIT.java @@ -61,8 +61,8 @@ import org.apache.accumulo.core.metadata.TServerInstance; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily; import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.fate.util.UtilWaitThread; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.server.manager.state.CurrentState; @@ -308,10 +308,10 @@ public class TabletStateChangeIteratorIT extends AccumuloClusterHarness { HashSet<TServerInstance> tservers = new HashSet<>(); for (String tserver : client.instanceOperations().getTabletServers()) { try { - String zPath = ZooUtil.getRoot(client.instanceOperations().getInstanceID()) - + Constants.ZTSERVERS + "/" + tserver; + var zPath = ServiceLock.path(ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + + Constants.ZTSERVERS + "/" + tserver); ClientInfo info = getClientInfo(); - long sessionId = ZooLock.getSessionId( + long sessionId = ServiceLock.getSessionId( new ZooCache(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()), zPath); tservers.add(new TServerInstance(tserver, sessionId)); } catch (Exception e) { diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java index 0ca819d..925441e 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/ZombieTServer.java @@ -39,9 +39,9 @@ import org.apache.accumulo.core.util.HostAndPort; import org.apache.accumulo.core.util.ServerServices; import org.apache.accumulo.core.util.ServerServices.Service; import org.apache.accumulo.core.util.threads.ThreadPools; -import org.apache.accumulo.fate.zookeeper.ZooLock; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason; -import org.apache.accumulo.fate.zookeeper.ZooLock.LockWatcher; +import org.apache.accumulo.fate.zookeeper.ServiceLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockLossReason; +import org.apache.accumulo.fate.zookeeper.ServiceLock.LockWatcher; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy; import org.apache.accumulo.server.ServerContext; @@ -116,11 +116,13 @@ public class ZombieTServer { null, -1, HostAndPort.fromParts("0.0.0.0", port)); String addressString = serverPort.address.toString(); - String zPath = context.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + addressString; + var zLockPath = + ServiceLock.path(context.getZooKeeperRoot() + Constants.ZTSERVERS + "/" + addressString); ZooReaderWriter zoo = context.getZooReaderWriter(); - zoo.putPersistentData(zPath, new byte[] {}, NodeExistsPolicy.SKIP); + zoo.putPersistentData(zLockPath.toString(), new byte[] {}, NodeExistsPolicy.SKIP); - ZooLock zlock = new ZooLock(context.getSiteConfiguration(), zPath, UUID.randomUUID()); + ServiceLock zlock = + new ServiceLock(context.getSiteConfiguration(), zLockPath, UUID.randomUUID()); LockWatcher lw = new LockWatcher() { diff --git a/test/src/main/java/org/apache/accumulo/test/replication/ReplicationIT.java b/test/src/main/java/org/apache/accumulo/test/replication/ReplicationIT.java index af50df8..0ba5cf7 100644 --- a/test/src/main/java/org/apache/accumulo/test/replication/ReplicationIT.java +++ b/test/src/main/java/org/apache/accumulo/test/replication/ReplicationIT.java @@ -83,9 +83,9 @@ import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.security.TablePermission; import org.apache.accumulo.core.tabletserver.log.LogEntry; import org.apache.accumulo.core.util.Pair; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooCache; import org.apache.accumulo.fate.zookeeper.ZooCacheFactory; -import org.apache.accumulo.fate.zookeeper.ZooLock; import org.apache.accumulo.fate.zookeeper.ZooUtil; import org.apache.accumulo.gc.SimpleGarbageCollector; import org.apache.accumulo.minicluster.ServerType; @@ -207,14 +207,14 @@ public class ReplicationIT extends ConfigurableMacBase { ZooCacheFactory zcf = new ZooCacheFactory(); ClientInfo info = ClientInfo.from(client.properties()); ZooCache zcache = zcf.getZooCache(info.getZooKeepers(), info.getZooKeepersSessionTimeOut()); - String zkPath = - ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK; + var zkPath = ServiceLock + .path(ZooUtil.getRoot(client.instanceOperations().getInstanceID()) + Constants.ZGC_LOCK); log.info("Looking for GC lock at {}", zkPath); - byte[] data = ZooLock.getLockData(zcache, zkPath, null); + byte[] data = ServiceLock.getLockData(zcache, zkPath, null); while (data == null) { log.info("Waiting for GC ZooKeeper lock to be acquired"); Thread.sleep(1000); - data = ZooLock.getLockData(zcache, zkPath, null); + data = ServiceLock.getLockData(zcache, zkPath, null); } } diff --git a/test/src/main/java/org/apache/accumulo/test/upgrade/GCUpgrade9to10TestIT.java b/test/src/main/java/org/apache/accumulo/test/upgrade/GCUpgrade9to10TestIT.java index fd75317..66e5718 100644 --- a/test/src/main/java/org/apache/accumulo/test/upgrade/GCUpgrade9to10TestIT.java +++ b/test/src/main/java/org/apache/accumulo/test/upgrade/GCUpgrade9to10TestIT.java @@ -43,7 +43,7 @@ import org.apache.accumulo.core.metadata.schema.Ample; import org.apache.accumulo.core.metadata.schema.MetadataSchema.DeletesSection; import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.security.TablePermission; -import org.apache.accumulo.fate.zookeeper.ZooLock; +import org.apache.accumulo.fate.zookeeper.ServiceLock; import org.apache.accumulo.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.manager.upgrade.Upgrader9to10; import org.apache.accumulo.minicluster.ServerType; @@ -83,10 +83,10 @@ public class GCUpgrade9to10TestIT extends ConfigurableMacBase { getCluster().killProcess(ServerType.GARBAGE_COLLECTOR, getCluster().getProcesses().get(ServerType.GARBAGE_COLLECTOR).iterator().next()); // delete lock in zookeeper if there, this will allow next GC to start quickly - String path = getServerContext().getZooKeeperRoot() + Constants.ZGC_LOCK; + var path = ServiceLock.path(getServerContext().getZooKeeperRoot() + Constants.ZGC_LOCK); ZooReaderWriter zk = new ZooReaderWriter(cluster.getZooKeepers(), 30000, OUR_SECRET); try { - ZooLock.deleteLock(zk, path); + ServiceLock.deleteLock(zk, path); } catch (IllegalStateException e) { log.error("Unable to delete ZooLock for mini accumulo-gc", e); }