You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2020/04/14 02:30:20 UTC
[helix] 07/19: Added unit tests for Helix nonblocking lock
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch distributed-lock
in repository https://gitbox.apache.org/repos/asf/helix.git
commit d58fbe6b2a87b6416330a4468c0a94710616602f
Author: Molly Gao <mg...@mgao-mn1.linkedin.biz>
AuthorDate: Mon Feb 3 14:26:41 2020 -0800
Added unit tests for Helix nonblocking lock
---
.../apache/helix/lock/ZKHelixNonblockingLock.java | 50 ++++--
.../helix/lock/ZKHelixNonblockingLockInfo.java | 27 +++-
.../helix/lock/TestZKHelixNonblockingLock.java | 176 +++++++++++++++++++++
3 files changed, 236 insertions(+), 17 deletions(-)
diff --git a/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLock.java b/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLock.java
index 75dd0e5..5be84bb 100644
--- a/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLock.java
+++ b/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLock.java
@@ -26,6 +26,7 @@ import java.util.UUID;
import org.apache.helix.AccessOption;
import org.apache.helix.BaseDataAccessor;
import org.apache.helix.ZNRecord;
+import org.apache.helix.ZNRecordUpdater;
import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.manager.zk.ZkClient;
@@ -33,6 +34,7 @@ import org.apache.helix.manager.zk.client.HelixZkClient;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.util.ZNRecordUtil;
import org.apache.log4j.Logger;
+import org.apache.zookeeper.data.Stat;
/**
@@ -80,32 +82,44 @@ public class ZKHelixNonblockingLock implements HelixLock {
}
@Override
- public boolean acquireLock() {
+ /**
+ * Blocking call to acquire a lock
+ * @return true if the lock was successfully acquired,
+ * false if the lock could not be acquired
+ */ public boolean acquireLock() {
// Set lock information fields
ZNRecord lockInfo = new ZNRecord(_userID);
- lockInfo.setSimpleField("Owner", _userID);
- lockInfo.setSimpleField("message", _lockMsg);
+ lockInfo.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name(), _userID);
+ lockInfo.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.MESSAGE.name(), _lockMsg);
long timeout = System.currentTimeMillis() + _timeout;
- lockInfo.setSimpleField("timeout", String.valueOf(timeout));
+ lockInfo
+ .setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.TIMEOUT.name(), String.valueOf(timeout));
// Try to create the lock node
boolean success = _baseDataAccessor.create(_lockPath, lockInfo, AccessOption.PERSISTENT);
- // If fail to create the lock node, compare the timeout timestamp of current lock node with current time, if already passes the timeout, delete current lock node and try to create lock node again
+ // If fail to create the lock node (acquire the lock), compare the timeout timestamp of current lock node with current time, if already passes the timeout, release current lock and try to acquire the lock again
if (!success) {
- ZNRecord curLock = _baseDataAccessor.get(_lockPath, null, AccessOption.PERSISTENT);
- long curTimeout = Long.parseLong(curLock.getSimpleField("timeout"));
+ Stat stat = new Stat();
+ ZNRecord curLock = _baseDataAccessor.get(_lockPath, stat, AccessOption.PERSISTENT);
+ long curTimeout =
+ Long.parseLong(curLock.getSimpleField(ZKHelixNonblockingLockInfo.InfoKey.TIMEOUT.name()));
if (System.currentTimeMillis() >= curTimeout) {
- _baseDataAccessor.remove(_lockPath, AccessOption.PERSISTENT);
- return _baseDataAccessor.create(_lockPath, lockInfo, AccessOption.PERSISTENT);
+ success =
+ _baseDataAccessor.set(_lockPath, lockInfo, stat.getVersion(), AccessOption.PERSISTENT);
}
}
return success;
}
@Override
- public boolean releaseLock() {
+ /**
+ * Blocking call to release a lock
+ * @return true if the lock was successfully released,
+ * false if the locked is not locked or is not locked by the user,
+ * or the lock could not be released
+ */ public boolean releaseLock() {
if (isOwner()) {
return _baseDataAccessor.remove(_lockPath, AccessOption.PERSISTENT);
}
@@ -113,7 +127,13 @@ public class ZKHelixNonblockingLock implements HelixLock {
}
@Override
- public LockInfo<String> getLockInfo() {
+ /**
+ * Retrieve the lock information, e.g. lock timeout, lock message, etc.
+ * @return lock metadata information, return null if there is no lock node for the path provided
+ */ public LockInfo<String> getLockInfo() {
+ if (!_baseDataAccessor.exists(_lockPath, AccessOption.PERSISTENT)) {
+ return null;
+ }
ZKHelixNonblockingLockInfo<String> lockInfo = new ZKHelixNonblockingLockInfo<>();
ZNRecord curLockInfo = _baseDataAccessor.get(_lockPath, null, AccessOption.PERSISTENT);
lockInfo.setLockInfoFields(curLockInfo);
@@ -121,12 +141,16 @@ public class ZKHelixNonblockingLock implements HelixLock {
}
@Override
- public boolean isOwner() {
+ /**
+ * Check if the user is current lock owner
+ * @return true if the user is the lock owner,
+ * false if the user is not the lock owner or the lock doesn't have a owner
+ */ public boolean isOwner() {
ZNRecord curLockInfo = _baseDataAccessor.get(_lockPath, null, AccessOption.PERSISTENT);
if (curLockInfo == null) {
return false;
}
- String ownerID = curLockInfo.getSimpleField("owner");
+ String ownerID = curLockInfo.getSimpleField(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name());
if (ownerID == null) {
return false;
}
diff --git a/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLockInfo.java b/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLockInfo.java
index d54c386..fb86187 100644
--- a/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLockInfo.java
+++ b/helix-lock/src/main/java/org/apache/helix/lock/ZKHelixNonblockingLockInfo.java
@@ -27,23 +27,42 @@ import org.apache.helix.ZNRecord;
public class ZKHelixNonblockingLockInfo<T extends String> implements LockInfo<T> {
- private Map<String, String> lockInfo;
+ private Map<String, String> lockInfo;
+
+ enum InfoKey {
+ OWNER, MESSAGE, TIMEOUT
+ }
public ZKHelixNonblockingLockInfo() {
lockInfo = new HashMap<>();
}
@Override
- public void setInfoValue(String infoKey, String infoValue) {
+ /**
+ * Create a single filed of LockInfo, or update the value of the field if it already exists
+ * @param infoKey the key of the LockInfo field
+ * @param infoValue the value of the LockInfo field
+ */ public void setInfoValue(String infoKey, String infoValue) {
lockInfo.put(infoKey, infoValue);
}
@Override
- public T getInfoValue(String infoKey) {
- return (T)lockInfo.get(infoKey);
+ /**
+ * Get the value of a field in LockInfo
+ * @param infoKey the key of the LockInfo field
+ * @return the value of the field or null if this key does not exist
+ */ public T getInfoValue(String infoKey) {
+ return (T) lockInfo.get(infoKey);
}
+ /**
+ * Update the lock info with information in a ZNRecord
+ * @param record Information about the lock that stored as ZNRecord format
+ */
public void setLockInfoFields(ZNRecord record) {
+ if (record == null) {
+ return;
+ }
Map<String, String> recordSimpleFields = record.getSimpleFields();
lockInfo.putAll(recordSimpleFields);
}
diff --git a/helix-lock/src/test/java/org/apache/helix/lock/TestZKHelixNonblockingLock.java b/helix-lock/src/test/java/org/apache/helix/lock/TestZKHelixNonblockingLock.java
new file mode 100644
index 0000000..95c41ac
--- /dev/null
+++ b/helix-lock/src/test/java/org/apache/helix/lock/TestZKHelixNonblockingLock.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.helix.lock;
+
+import java.util.Date;
+import java.util.UUID;
+
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.model.HelixConfigScope;
+import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
+import org.apache.helix.model.builder.HelixConfigScopeBuilder;
+import org.apache.zookeeper.CreateMode;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class TestZKHelixNonblockingLock extends ZkUnitTestBase {
+
+ private final String _lockPath = "/testLockPath";
+ private final String _className = TestHelper.getTestClassName();
+ private final String _methodName = TestHelper.getTestMethodName();
+ private final String _clusterName = _className + "_" + _methodName;
+ private final String _lockMessage = "Test";
+
+ @BeforeClass
+ public void beforeClass() throws Exception {
+
+ System.out.println("START " + _clusterName + " at " + new Date(System.currentTimeMillis()));
+
+ TestHelper.setupCluster(_clusterName, ZK_ADDR, 12918, "localhost", "TestDB", 1, 10, 5, 3,
+ "MasterSlave", true);
+ }
+
+ @Test
+ public void testAcquireLockWithParticipantScope() {
+
+ // Initialize lock with participant config scope
+ HelixConfigScope participantScope =
+ new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT).forCluster(_clusterName)
+ .forParticipant("localhost_12918").build();
+
+ String userID = UUID.randomUUID().toString();
+
+ ZKHelixNonblockingLock lock =
+ new ZKHelixNonblockingLock(_clusterName, participantScope, ZK_ADDR, Long.MAX_VALUE,
+ _lockMessage, userID);
+
+ // Acquire lock
+ lock.acquireLock();
+ String lockPath = "/" + _clusterName + '/' + "LOCKS" + '/' + participantScope;
+ Assert.assertTrue(_gZkClient.exists(lockPath));
+
+ // Get lock information
+ LockInfo<String> record = lock.getLockInfo();
+ Assert
+ .assertEquals(record.getInfoValue(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name()), userID);
+ Assert.assertEquals(record.getInfoValue(ZKHelixNonblockingLockInfo.InfoKey.MESSAGE.name()),
+ _lockMessage);
+
+ // Check if the user is lock owner
+ Assert.assertTrue(lock.isOwner());
+
+ // Release lock
+ lock.releaseLock();
+ Assert.assertFalse(_gZkClient.exists(lockPath));
+ }
+
+ @Test
+ public void testAcquireLockWithUserProvidedPath() {
+
+ // Initialize lock with user provided path
+ String userID = UUID.randomUUID().toString();
+
+ ZKHelixNonblockingLock lock =
+ new ZKHelixNonblockingLock(_lockPath, ZK_ADDR, Long.MAX_VALUE, _lockMessage, userID);
+
+ //Acquire lock
+ lock.acquireLock();
+ Assert.assertTrue(_gZkClient.exists(_lockPath));
+
+ // Get lock information
+ LockInfo<String> record = lock.getLockInfo();
+ Assert
+ .assertEquals(record.getInfoValue(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name()), userID);
+ Assert.assertEquals(record.getInfoValue(ZKHelixNonblockingLockInfo.InfoKey.MESSAGE.name()),
+ _lockMessage);
+
+ // Check if the user is lock owner
+ Assert.assertTrue(lock.isOwner());
+
+ // Release lock
+ lock.releaseLock();
+ Assert.assertFalse(_gZkClient.exists(_lockPath));
+ }
+
+ @Test
+ public void testAcquireLockWhenExistingLockNotExpired() {
+
+ // Initialize lock with user provided path
+ String ownerID = UUID.randomUUID().toString();
+
+ ZKHelixNonblockingLock lock =
+ new ZKHelixNonblockingLock(_lockPath, ZK_ADDR, 0L, _lockMessage, ownerID);
+
+ // Fake condition when the lock owner is not current user
+ String fakeUserID = UUID.randomUUID().toString();
+ ZNRecord fakeRecord = new ZNRecord(fakeUserID);
+ fakeRecord.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name(), fakeUserID);
+ fakeRecord.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.TIMEOUT.name(),
+ String.valueOf(Long.MAX_VALUE));
+ _gZkClient.create(_lockPath, fakeRecord, CreateMode.PERSISTENT);
+
+ // Check if the user is lock owner
+ Assert.assertFalse(lock.isOwner());
+
+ // Acquire lock
+ Assert.assertFalse(lock.acquireLock());
+ Assert.assertFalse(lock.isOwner());
+
+ // Release lock
+ Assert.assertFalse(lock.releaseLock());
+ Assert.assertTrue(_gZkClient.exists(_lockPath));
+
+ _gZkClient.delete(_lockPath);
+ }
+
+ @Test
+ public void testAcquireLockWhenExistingLockExpired() {
+
+ // Initialize lock with user provided path
+ String ownerID = UUID.randomUUID().toString();
+
+ ZKHelixNonblockingLock lock =
+ new ZKHelixNonblockingLock(_lockPath, ZK_ADDR, Long.MAX_VALUE, _lockMessage, ownerID);
+
+ // Fake condition when the current lock already expired
+ String fakeUserID = UUID.randomUUID().toString();
+ ZNRecord fakeRecord = new ZNRecord(fakeUserID);
+ fakeRecord.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.OWNER.name(), fakeUserID);
+ fakeRecord.setSimpleField(ZKHelixNonblockingLockInfo.InfoKey.TIMEOUT.name(),
+ String.valueOf(System.currentTimeMillis()));
+ _gZkClient.create(_lockPath, fakeRecord, CreateMode.PERSISTENT);
+
+ // Acquire lock
+ Assert.assertTrue(lock.acquireLock());
+ Assert.assertTrue(_gZkClient.exists(_lockPath));
+
+ // Check if the current user is the lock owner
+ Assert.assertTrue(lock.isOwner());
+
+ // Release lock
+ Assert.assertTrue(lock.releaseLock());
+ Assert.assertFalse(_gZkClient.exists(_lockPath));
+ }
+}
+