Fix helix-lock regression (apache#2698)

Fix helix-lock regression by changing the default lock priority to 0

helix-lock/src/main/java/org/apache/helix/lock/LockInfo.java

@@ -168,7 +168,7 @@ public Long getTimeout() {
 
   /**
    * Get the value for PRIORITY attribute of the lock
-   * @return the priority of the lock, -1 if there is no priority set
+   * @return the priority of the lock, 0 if there is no priority set
    */
   public Integer getPriority() {
     return _record
@@ -204,7 +204,7 @@ public String getRequestorId() {
 
   /**
    * Get the value for REQUESTOR_PRIORITY attribute of the lock
-   * @return the requestor priority of the lock, -1 if there is no requestor priority set
+   * @return the requestor priority of the lock, 0 if there is no requestor priority set
    */
   public int getRequestorPriority() {
     return _record.getIntField(LockInfoAttribute.REQUESTOR_PRIORITY.name(),

helix-lock/src/main/java/org/apache/helix/lock/helix/LockConstants.java

@@ -27,7 +27,7 @@ public class LockConstants {
   public static final String DEFAULT_USER_ID = "NONE";
   public static final String DEFAULT_MESSAGE_TEXT = "NONE";
   public static final long DEFAULT_TIMEOUT_LONG = -1;
-  public static final int DEFAULT_PRIORITY_INT = -1;
+  public static final int DEFAULT_PRIORITY_INT = 0;
   public static final long DEFAULT_WAITING_TIMEOUT_LONG = -1;
   public static final long DEFAULT_CLEANUP_TIMEOUT_LONG = -1;
   public static final long DEFAULT_REQUESTING_TIMESTAMP_LONG = -1;

helix-lock/src/main/java/org/apache/helix/lock/helix/ZKDistributedNonblockingLock.java

@@ -312,7 +312,17 @@ public ZNRecord update(ZNRecord current) {
         return _record;
       }
 
-      // higher priority lock request will try to  preempt current lock owner
+      LockInfo unlockOrLockRequestLockInfo = new LockInfo(_record);
+      // Any unlock request from non-lock owners is blocked.
+      if (unlockOrLockRequestLockInfo.getOwner().equals(LockConstants.DEFAULT_USER_ID)) {
+        LOG.error("User {} is not the lock owner and cannot release lock at Lock path {}.", _userId,
+            _lockPath);
+        throw new HelixException(
+            String.format("User %s is not the lock owner and cannot release lock at Lock path %s.",
+                _userId, _lockPath));
+      }
+
+      // higher priority lock request will try to  preempt current lock owner.
       if (!isCurrentOwner(curLockInfo) && _priority > curLockInfo.getPriority()) {
         // if requestor field is empty, fill the field with requestor's id
         if (curLockInfo.getRequestorId().equals(LockConstants.DEFAULT_USER_ID)) {

helix-lock/src/test/java/org/apache/helix/lock/helix/TestZKHelixNonblockingLock.java

@@ -141,6 +141,46 @@ public void testAcquireLockWhenExistingLockExpired() {
     Assert.assertFalse(_lock.isCurrentOwner());
   }
 
+  @Test
+  public void testNonLockOwnerUnlockNoPriorityLock() {
+    // Fake condition when the lock owner is not current user
+    String fakeUserID = UUID.randomUUID().toString();
+    ZNRecord fakeRecord = new ZNRecord(fakeUserID);
+    fakeRecord.setSimpleField(LockInfo.LockInfoAttribute.OWNER.name(), fakeUserID);
+    fakeRecord
+        .setSimpleField(LockInfo.LockInfoAttribute.TIMEOUT.name(), String.valueOf(Long.MAX_VALUE));
+    _gZkClient.create(_lockPath, fakeRecord, CreateMode.PERSISTENT);
+
+    // Verify the current user is not a lock owner
+    Assert.assertFalse(_lock.isCurrentOwner());
+    // trylock and unlock will return false since both users have default priority of 0
+    Assert.assertFalse(_lock.tryLock());
+    Assert.assertFalse(_lock.unlock());
+  }
+
+  @Test
+  public void testNonLockOwnerUnlockNoPriorityLockFail() {
+    // Fake condition when the lock owner is not current user
+    String fakeUserID = UUID.randomUUID().toString();
+    ZNRecord fakeRecord = new ZNRecord(fakeUserID);
+    fakeRecord.setSimpleField(LockInfo.LockInfoAttribute.OWNER.name(), fakeUserID);
+    fakeRecord
+        .setSimpleField(LockInfo.LockInfoAttribute.TIMEOUT.name(), String.valueOf(Long.MAX_VALUE));
+    _gZkClient.create(_lockPath, fakeRecord, CreateMode.PERSISTENT);
+
+    ZKDistributedNonblockingLock.Builder lockBuilder = new ZKDistributedNonblockingLock.Builder();
+    lockBuilder.setLockScope(_participantScope).setZkAddress(ZK_ADDR).setTimeout(3600000L)
+        .setLockMsg("higher priority lock").setUserId("user1").setPriority(5)
+        .setWaitingTimeout(30000).setCleanupTimeout(10000).setIsForceful(false);
+    ZKDistributedNonblockingLock lock = lockBuilder.build();
+    // Verify the current user is not a lock owner
+    Assert.assertFalse(lock.isCurrentOwner());
+    // Since _lock with _userId is not the locker owner anymore, its unlock() should fail.
+    Assert.assertFalse(lock.unlock());
+    lock.close();
+  }
+
+
   @Test
   public void testSimultaneousAcquire() {
     List<Callable<Boolean>> threads = new ArrayList<>();

helix-lock/src/test/java/org/apache/helix/lock/helix/TestZKHelixNonblockingLockWithPriority.java

@@ -81,6 +81,46 @@ public void afterSuite() throws IOException {
     super.afterSuite();
   }
 
+  @Test
+  public void testNonLockOwnerUnlockFail() throws Exception {
+    ZKDistributedNonblockingLock.Builder lockBuilder = new ZKDistributedNonblockingLock.Builder();
+    lockBuilder.setLockScope(_participantScope).setZkAddress(ZK_ADDR).setTimeout(3600000L)
+        .setLockMsg("higher priority lock").setUserId("user1").setPriority(0)
+        .setWaitingTimeout(30000).setCleanupTimeout(10000).setIsForceful(false)
+        .setLockListener(createLockListener());
+    ZKDistributedNonblockingLock lock = lockBuilder.build();
+
+    Thread t = new Thread() {
+      @Override
+      public void run() {
+        lock.tryLock();
+      }
+    };
+
+    t.start();
+    t.join();
+    Assert.assertTrue(lock.isCurrentOwner());
+
+    lockBuilder.setUserId("user2").setPriority(5);
+    ZKDistributedNonblockingLock lock2 = lockBuilder.build();
+    // unlock should fail because even if user2 has higher priority because user2 set can unlock
+    // not owned lock to false.
+    Assert.assertFalse(lock2.unlock());
+    t = new Thread() {
+      @Override
+      public void run() {
+        lock2.tryLock();
+      }
+    };
+    t.start();
+    t.join();
+    Assert.assertTrue(lock2.isCurrentOwner());
+    lock2.unlock();
+    lock2.close();
+    lock.close();
+  }
+
+
   @Test
   public void testLowerPriorityRequestRejected() throws Exception {
     ZKDistributedNonblockingLock lock = createLockWithConfig();