Merge branch '2.1'

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/lock/ServiceLock.java

@@ -47,8 +47,27 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Preconditions;
+
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
+/**
+ * This class uses Sequential Ephemeral ZooKeeper Nodes
+ * (https://zookeeper.apache.org/doc/r3.9.5/zookeeperProgrammers.html#Sequence+Nodes+--+Unique+Naming)
+ * to implement locks for Accumulo server processes. This class will create an ephemeral sequential
+ * node under a base path using the prefix "zlock#" + UUID + "#". The ZooKeeper server will append a
+ * 10-digit zero-padded number to this prefix using a one-up counter. The base path could be an HA
+ * service like the Manager or a non-HA service like a TabletServer.
+ *
+ * When an instance of this class has the lowest counter number at the base path, then it has the
+ * lock. When an instance of this class does not have the lowest counter number, then it watches the
+ * node with the next lowest counter number. When the node that has the next lowest counter number
+ * is deleted, then this instance could acquire the lock.
+ *
+ * Instance of this class also place a Watcher on the base path node. If the base path Watcher
+ * receives a Session Expired event, then it calls lostLock which should end up halting the Accumulo
+ * server process.
+ */
 @SuppressFBWarnings(value = "CT_CONSTRUCTOR_THROW",
     justification = "Constructor validation is required for proper initialization")
 public class ServiceLock implements Watcher {
@@ -89,38 +108,70 @@ public interface AccumuloLockWatcher extends LockWatcher {
     void failedToAcquireLock(Exception e);
   }
 
+  // the base path
   private final ServiceLockPath path;
+
   protected final ZooSession zooKeeper;
+
+  // "zlock#" + UUID + "#"
   private final Prefix vmLockPrefix;
 
+  // A LockWatcher instance supplied to this class
+  // by the caller when trying to acquire the lock
+  // in ZooKeeper. This object does not represent
+  // a Watcher in ZooKeeper, but allows an instance
+  // of this class to communicate with the calling
+  // object by invoking callback methods.
   private LockWatcher lockWatcher;
-  private String lockNodeName;
+
+  // A variable which is initially null, then set
+  // to the createdNodeName when the lock is acquired.
+  // This variable is set to null when this instance
+  // loses the lock.
+  private volatile String lockNodeName;
+
+  // boolean to track if this instance has ever held the lock
   private volatile boolean lockWasAcquired;
-  private volatile boolean watchingParent;
 
-  private String createdNodeName;
-  private String watchingNodeName;
+  // boolean to track if there is a watcher on the base path.
+  private volatile boolean watchingBasePath;
+
+  // Represents the name of the ephemeral sequential node that
+  // the ZooKeeper server created for us with the unique one-up
+  // counter. This variable is set to null when this instance
+  // acquires the lock.
+  private volatile String createdNodeName;
+
+  // Represents the path of the ephemeral sequential node that
+  // has the next lowest counter value. An instance of this class
+  // will watch this node and will attempt to acquire the lock
+  // when this node is deleted.
+  private String watchingNodePath;
 
   public ServiceLock(ZooSession zookeeper, ServiceLockPath path, UUID uuid) {
     this.zooKeeper = requireNonNull(zookeeper);
     this.path = requireNonNull(path);
     try {
       zooKeeper.exists(path.toString(), this);
-      watchingParent = true;
+      watchingBasePath = true;
       this.vmLockPrefix = new Prefix(ZLOCK_PREFIX + uuid.toString() + "#");
     } catch (KeeperException | InterruptedException ex) {
       LOG.error("Error setting initial watch", ex);
       throw new IllegalStateException(ex);
     }
   }
 
-  private static class LockWatcherWrapper implements AccumuloLockWatcher {
+  // Watcher used in tryLock method that wraps the supplied LockWatcher
+  // so that the supplied LockWatcher.failedToAcquireLock method is not
+  // invoked. The supplied LockWatcher.failedToAcquireLock method may
+  // halt the server, but we don't want to do that in a tryLock invocation.
+  private static class TryLockWatcherWrapper implements AccumuloLockWatcher {
 
-    boolean acquiredLock = false;
-    final LockWatcher lw;
+    private boolean acquiredLock = false;
+    private final LockWatcher delegate;
 
-    public LockWatcherWrapper(LockWatcher lw2) {
-      this.lw = lw2;
+    public TryLockWatcherWrapper(LockWatcher lw2) {
+      this.delegate = lw2;
     }
 
     @Override
@@ -135,20 +186,20 @@ public void failedToAcquireLock(Exception e) {
 
     @Override
     public void lostLock(LockLossReason reason) {
-      lw.lostLock(reason);
+      delegate.lostLock(reason);
     }
 
     @Override
     public void unableToMonitorLockNode(Exception e) {
-      lw.unableToMonitorLockNode(e);
+      delegate.unableToMonitorLockNode(e);
     }
 
   }
 
   public synchronized boolean tryLock(LockWatcher lw, ServiceLockData lockData)
       throws KeeperException, InterruptedException {
 
-    LockWatcherWrapper lww = new LockWatcherWrapper(lw);
+    TryLockWatcherWrapper lww = new TryLockWatcherWrapper(lw);
 
     lock(lww, lockData);
 
@@ -263,48 +314,45 @@ public static String findLowestPrevPrefix(final List<String> children,
     return lowestPrevNode;
   }
 
-  private synchronized void determineLockOwnership(final String createdEphemeralNode,
-      final AccumuloLockWatcher lw) throws KeeperException, InterruptedException {
+  private synchronized void determineLockOwnership(final AccumuloLockWatcher lw)
+      throws KeeperException, InterruptedException {
 
-    if (createdNodeName == null) {
-      throw new IllegalStateException(
-          "Called determineLockOwnership() when ephemeralNodeName == null");
-    }
+    Preconditions.checkState(createdNodeName != null, "createdNodeName cannot be null");
 
     List<String> children = validateAndSort(path, zooKeeper.getChildren(path.toString(), null));
 
-    if (!children.contains(createdEphemeralNode)) {
-      LOG.error("Expected ephemeral node {} to be in the list of children {}", createdEphemeralNode,
+    if (!children.contains(createdNodeName)) {
+      LOG.error("Expected ephemeral node {} to be in the list of children {}", createdNodeName,
           children);
       throw new IllegalStateException(
-          "Lock attempt ephemeral node no longer exist " + createdEphemeralNode);
+          "Lock attempt ephemeral node no longer exist " + createdNodeName);
     }
 
-    if (children.get(0).equals(createdEphemeralNode)) {
+    if (children.get(0).equals(createdNodeName)) {
       LOG.debug("[{}] First candidate is my lock, acquiring...", vmLockPrefix);
-      if (!watchingParent) {
+      if (!watchingBasePath) {
         throw new IllegalStateException(
             "Can not acquire lock, no longer watching parent : " + path);
       }
       this.lockWatcher = lw;
-      this.lockNodeName = createdEphemeralNode;
+      this.lockNodeName = createdNodeName;
       createdNodeName = null;
       lockWasAcquired = true;
       lw.acquiredLock();
     } else {
       LOG.debug("[{}] Lock held by another process with ephemeral node: {}", vmLockPrefix,
           children.get(0));
 
-      String lowestPrevNode = findLowestPrevPrefix(children, createdEphemeralNode);
+      String lowestPrevNode = findLowestPrevPrefix(children, createdNodeName);
 
-      watchingNodeName = path + "/" + lowestPrevNode;
-      final String nodeToWatch = watchingNodeName;
+      watchingNodePath = path + "/" + lowestPrevNode;
+      final String nodeToWatch = watchingNodePath;
       LOG.debug("[{}] Establishing watch on prior node {}", vmLockPrefix, nodeToWatch);
       Watcher priorNodeWatcher = new Watcher() {
         @Override
         public void process(WatchedEvent event) {
-          if (LOG.isTraceEnabled()) {
-            LOG.trace("[{}] Processing {}", vmLockPrefix, event);
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("[{}] Processing {}", vmLockPrefix, event);
           }
           boolean renew = true;
           if (event.getType() == EventType.NodeDeleted && event.getPath().equals(nodeToWatch)) {
@@ -313,10 +361,10 @@ public void process(WatchedEvent event) {
             synchronized (ServiceLock.this) {
               try {
                 if (createdNodeName != null) {
-                  determineLockOwnership(createdEphemeralNode, lw);
+                  determineLockOwnership(lw);
                 } else if (LOG.isDebugEnabled()) {
                   LOG.debug("[{}] While waiting for another lock {}, {} was deleted; {}",
-                      vmLockPrefix, nodeToWatch, createdEphemeralNode, event);
+                      vmLockPrefix, nodeToWatch, createdNodeName, event);
                 }
               } catch (Exception e) {
                 if (lockNodeName == null) {
@@ -340,20 +388,24 @@ public void process(WatchedEvent event) {
             renew = false;
           }
           if (renew) {
-            try {
-              Stat restat = zooKeeper.exists(nodeToWatch, this);
-              if (restat == null) {
-                // if stat is null from the zookeeper.exists(path, Watcher) call, then we just
-                // created a Watcher on a node that does not exist. Delete the watcher we just
-                // created.
-                zooKeeper.removeWatches(nodeToWatch, this, WatcherType.Any, true);
-                determineLockOwnership(createdEphemeralNode, lw);
-              } else {
-                LOG.debug("[{}] Renewed watch on prior node  {}", vmLockPrefix, nodeToWatch);
+            synchronized (ServiceLock.this) {
+              if (createdNodeName != null) {
+                try {
+                  Stat restat = zooKeeper.exists(nodeToWatch, this);
+                  if (restat == null) {
+                    // if stat is null from the zookeeper.exists(path, Watcher) call, then we just
+                    // created a Watcher on a node that does not exist. Delete the watcher we just
+                    // created.
+                    zooKeeper.removeWatches(nodeToWatch, this, WatcherType.Any, true);
+                    determineLockOwnership(lw);
+                  } else {
+                    LOG.debug("[{}] Renewed watch on prior node {}", vmLockPrefix, nodeToWatch);
+                  }
+                } catch (KeeperException | InterruptedException e) {
+                  lw.failedToAcquireLock(
+                      new Exception("Failed to renew watch on prior node: " + nodeToWatch, e));
+                }
               }
-            } catch (KeeperException | InterruptedException e) {
-              lw.failedToAcquireLock(
-                  new Exception("Failed to renew watch on other manager node", e));
             }
           }
         }
@@ -365,7 +417,7 @@ public void process(WatchedEvent event) {
         // if stat is null from the zookeeper.exists(path, Watcher) call, then we just
         // created a Watcher on a node that does not exist. Delete the watcher we just created.
         zooKeeper.removeWatches(nodeToWatch, priorNodeWatcher, WatcherType.Any, true);
-        determineLockOwnership(createdEphemeralNode, lw);
+        determineLockOwnership(lw);
       }
     }
 
@@ -507,7 +559,7 @@ public void process(WatchedEvent event) {
       createdNodeName = pathForWatcher.substring(path.toString().length() + 1);
 
       // We have created a node, do we own the lock?
-      determineLockOwnership(createdNodeName, lw);
+      determineLockOwnership(lw);
 
     } catch (KeeperException | InterruptedException e) {
       lw.failedToAcquireLock(e);
@@ -571,7 +623,7 @@ public synchronized void unlock() throws InterruptedException, KeeperException {
    * @return path of node that this lock is watching
    */
   public synchronized String getWatching() {
-    return watchingNodeName;
+    return watchingNodePath;
   }
 
   public synchronized String getLockPath() {
@@ -628,15 +680,15 @@ public synchronized void process(WatchedEvent event) {
       LOG.debug("{}", event);
     }
 
-    watchingParent = false;
+    watchingBasePath = false;
 
     if (event.getState() == KeeperState.Expired && lockNodeName != null) {
       lostLock(LockLossReason.SESSION_EXPIRED);
     } else {
 
       try { // set the watch on the parent node again
         zooKeeper.exists(path.toString(), this);
-        watchingParent = true;
+        watchingBasePath = true;
       } catch (KeeperException.ConnectionLossException ex) {
         // we can't look at the lock because we aren't connected, but our session is still good
         LOG.warn("lost connection to zookeeper", ex);

test/src/main/java/org/apache/accumulo/test/lock/ServiceLockIT.java

@@ -653,7 +653,7 @@ public void testTryLock() throws Exception {
 
       // make sure still watching parent even though a lot of events occurred for the parent
       synchronized (zl) {
-        Field field = zl.getClass().getDeclaredField("watchingParent");
+        Field field = zl.getClass().getDeclaredField("watchingBasePath");
         field.setAccessible(true);
         assertTrue((Boolean) field.get(zl));
       }