You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2022/04/29 09:55:02 UTC
[shardingsphere] branch master updated: Add standard distribute lock manager (#17181)
This is an automated email from the ASF dual-hosted git repository.
zhaojinchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new b3f8064ac87 Add standard distribute lock manager (#17181)
b3f8064ac87 is described below
commit b3f8064ac87211f71d72c7746f277b2585bae35a
Author: gin <ja...@163.com>
AuthorDate: Fri Apr 29 17:54:53 2022 +0800
Add standard distribute lock manager (#17181)
* Add standard distribute lock manager
* Fixed
---
.../shardingsphere/infra/lock/LockContext.java | 8 ++
.../apache/shardingsphere/infra/lock/LockType.java | 2 +-
.../infra/lock/ShardingSphereLock.java | 5 +-
.../fixture/FixtureClusterPersistRepository.java | 11 ++
.../core/api/impl/RuleAlteredJobAPIImpl.java | 10 +-
.../pipeline/core/lock/PipelineSimpleLock.java | 18 +--
.../rulealtered/RuleAlteredJobPreparer.java | 2 +-
.../scenario/rulealtered/RuleAlteredJobWorker.java | 2 +-
.../coordinator/lock/DistributeLockContext.java | 5 +
.../cluster/coordinator/lock/LockNodeService.java | 36 ++---
.../global/ShardingSphereDistributeGlobalLock.java | 2 +-
.../ShardingSphereSequencedSemaphoreLock.java | 4 +-
.../database/ShardingSphereDatabaseLock.java | 19 +--
.../ShardingSphereDatabaseLockManager.java | 12 +-
.../watcher/DatabaseAckChangedWatcher.java | 4 +-
.../watcher/DatabaseLocksChangedWatcher.java | 4 +-
.../global/general/ShardingSphereGeneralLock.java | 21 +--
.../general/ShardingSphereGeneralLockManager.java | 18 +--
.../general/watcher/GeneralAckChangedWatcher.java | 4 +-
.../watcher/GeneralLocksChangedWatcher.java | 4 +-
.../service/AbstractGlobalLockNodeService.java | 20 +--
.../global/service/GlobalLockRegistryService.java | 8 +-
.../ShardingSphereStandardLock.java} | 20 +--
.../ShardingSphereStandardLockManager.java} | 50 ++-----
...ryService.java => StandardLockNodeService.java} | 50 ++++---
.../service/StandardLockRegistryService.java | 20 +--
...ngsphere.mode.manager.ShardingSphereLockManager | 1 +
...anager.cluster.coordinator.lock.LockNodeService | 1 +
.../fixture/ClusterPersistRepositoryFixture.java | 11 ++
.../lock/DistributeLockContextTest.java | 1 -
...AbstractGlobalStandardLockNodeServiceTest.java} | 14 +-
...va => DatabaseStandardLockNodeServiceTest.java} | 2 +-
...ava => GeneralStandardLockNodeServiceTest.java} | 2 +-
.../service/GlobalLockRegistryServiceTest.java | 13 --
.../service/StandardLockRegistryServiceTest.java | 13 --
...ProcessListClusterPersistRepositoryFixture.java | 11 ++
.../cluster/ClusterPersistRepository.java | 39 ++++--
.../repository/cluster/etcd/EtcdRepository.java | 16 +++
.../cluster/etcd/lock/EtcdInternalLockHolder.java | 147 +++++++++++++++++++++
.../zookeeper/CuratorZookeeperRepository.java | 38 +++---
.../lock/ZookeeperInternalLockHolder.java | 131 ++++++++++++++++++
.../zookeeper/CuratorZookeeperRepositoryTest.java | 13 +-
.../manager/memory/lock/MemoryLockContext.java | 7 +-
.../lock/ShardingSphereNonReentrantLock.java | 2 +-
.../lock/ShardingSphereNonReentrantLock.java | 2 +-
.../standalone/lock/StandaloneLockContext.java | 7 +-
.../fixture/FixtureClusterPersistRepository.java | 11 ++
.../fixture/TestClusterPersistRepository.java | 11 ++
.../pipeline/core/lock/PipelineSimpleLockTest.java | 3 -
49 files changed, 590 insertions(+), 265 deletions(-)
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockContext.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockContext.java
index 8edde9ebdf0..d82aa1d9414 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockContext.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockContext.java
@@ -63,6 +63,14 @@ public interface LockContext {
*/
ShardingSphereLock getOrCreateGlobalLock(String lockName);
+ /**
+ * Get or create standard lock.
+ *
+ * @param lockName lock name
+ * @return standard lock
+ */
+ ShardingSphereLock getOrCreateStandardLock(String lockName);
+
/**
* Get global lock.
*
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockType.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockType.java
index 2fdb3c10241..54b752275b0 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockType.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/LockType.java
@@ -22,5 +22,5 @@ package org.apache.shardingsphere.infra.lock;
*/
public enum LockType {
- GENERAL, DATABASE, SCHEMA, TABLE
+ STANDARD, GENERAL, DATABASE, SCHEMA, TABLE
}
diff --git a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/ShardingSphereLock.java b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/ShardingSphereLock.java
index 0d127cc7b1b..d8f89ac68c9 100644
--- a/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/ShardingSphereLock.java
+++ b/shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/lock/ShardingSphereLock.java
@@ -48,9 +48,8 @@ public interface ShardingSphereLock {
/**
* Check whether resource is locked in current instance.
- *
- * @param lockName lock name
+ *
* @return true if locked, false if not
*/
- boolean isLocked(String lockName);
+ boolean isLocked();
}
diff --git a/shardingsphere-jdbc/shardingsphere-jdbc-spring/shardingsphere-jdbc-core-spring/shardingsphere-jdbc-core-spring-namespace/src/test/java/org/apache/shardingsphere/spring/namespace/fixture/FixtureClusterPersistRepository.java b/shardingsphere-jdbc/shardingsphere-jdbc-spring/shardingsphere-jdbc-core-spring/shardingsphere-jdbc-core-spring-namespace/src/test/java/org/apache/shardingsphere/spring/namespace/fixture/FixtureClusterPersistRepository.java
index ea346a81882..8a7d601425a 100644
--- a/shardingsphere-jdbc/shardingsphere-jdbc-spring/shardingsphere-jdbc-core-spring/shardingsphere-jdbc-core-spring-namespace/src/test/java/org/apache/shardingsphere/spring/namespace/fixture/FixtureClusterPersistRepository.java
+++ b/shardingsphere-jdbc/shardingsphere-jdbc-spring/shardingsphere-jdbc-core-spring/shardingsphere-jdbc-core-spring-namespace/src/test/java/org/apache/shardingsphere/spring/namespace/fixture/FixtureClusterPersistRepository.java
@@ -28,6 +28,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
public final class FixtureClusterPersistRepository implements ClusterPersistRepository {
@@ -84,6 +85,16 @@ public final class FixtureClusterPersistRepository implements ClusterPersistRepo
public void watchSessionConnection(final InstanceContext instanceContext) {
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return null;
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return null;
+ }
+
@Override
public void close() {
registryData.clear();
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
index d8ddc132273..5edfcd16711 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/RuleAlteredJobAPIImpl.java
@@ -164,9 +164,9 @@ public final class RuleAlteredJobAPIImpl extends AbstractPipelineJobAPIImpl impl
private void verifySourceWritingStopped(final JobConfiguration jobConfig) {
LockContext lockContext = PipelineContext.getContextManager().getInstanceContext().getLockContext();
- String schemaName = jobConfig.getWorkflowConfig().getDatabaseName();
- ShardingSphereLock lock = lockContext.getGlobalLock(schemaName);
- if (null == lock || !lock.isLocked(schemaName)) {
+ String databaseName = jobConfig.getWorkflowConfig().getDatabaseName();
+ ShardingSphereLock lock = lockContext.getGlobalLock(databaseName);
+ if (null == lock || !lock.isLocked()) {
throw new PipelineVerifyFailedException("Source writing is not stopped. You could run `STOP SCALING SOURCE WRITING {jobId}` to stop it.");
}
}
@@ -188,7 +188,7 @@ public final class RuleAlteredJobAPIImpl extends AbstractPipelineJobAPIImpl impl
public void stopClusterWriteDB(final String databaseName, final String jobId) {
LockContext lockContext = PipelineContext.getContextManager().getInstanceContext().getLockContext();
ShardingSphereLock lock = lockContext.getOrCreateGlobalLock(databaseName);
- if (lock.isLocked(databaseName)) {
+ if (lock.isLocked()) {
log.info("stopClusterWriteDB, already stopped");
return;
}
@@ -218,7 +218,7 @@ public final class RuleAlteredJobAPIImpl extends AbstractPipelineJobAPIImpl impl
log.info("restoreClusterWriteDB, lock is null");
return;
}
- boolean isLocked = lock.isLocked(databaseName);
+ boolean isLocked = lock.isLocked();
if (!isLocked) {
log.info("restoreClusterWriteDB, isLocked false, databaseName={}", databaseName);
return;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLock.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLock.java
index 3ed8a39e74d..f9821795b23 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLock.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLock.java
@@ -21,13 +21,9 @@ import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.core.constant.DataPipelineConstants;
import org.apache.shardingsphere.data.pipeline.core.context.PipelineContext;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockRegistryService;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard.service.StandardLockRegistryService;
-import org.apache.shardingsphere.mode.metadata.persist.MetaDataPersistService;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.infra.lock.LockContext;
import java.util.Map;
-import java.util.Optional;
/**
* Pipeline simple lock.
@@ -38,17 +34,13 @@ public final class PipelineSimpleLock {
private static volatile PipelineSimpleLock instance;
- private final LockRegistryService lockRegistryService;
+ private final LockContext lockContext;
private final Map<String, Boolean> lockNameLockedMap;
private PipelineSimpleLock() {
- Optional<MetaDataPersistService> persistServiceOptional = PipelineContext.getContextManager().getMetaDataContexts().getMetaDataPersistService();
- persistServiceOptional.orElseThrow(() -> new RuntimeException("Could not get metadata persist service"));
- // TODO Use PersistRepository later
- ClusterPersistRepository repository = (ClusterPersistRepository) persistServiceOptional.get().getRepository();
- lockRegistryService = new StandardLockRegistryService(repository);
lockNameLockedMap = Maps.newConcurrentMap();
+ lockContext = PipelineContext.getContextManager().getInstanceContext().getLockContext();
}
/**
@@ -75,7 +67,7 @@ public final class PipelineSimpleLock {
* @return true if lock got, else false
*/
public boolean tryLock(final String lockName, final long timeoutMills) {
- boolean result = lockRegistryService.tryLock(decorateLockName(lockName), timeoutMills);
+ boolean result = lockContext.getOrCreateStandardLock(decorateLockName(lockName)).tryLock(decorateLockName(lockName), timeoutMills);
if (result) {
lockNameLockedMap.put(lockName, true);
}
@@ -92,7 +84,7 @@ public final class PipelineSimpleLock {
log.info("releaseLock, lockName={}", lockName);
if (lockNameLockedMap.getOrDefault(lockName, false)) {
lockNameLockedMap.remove(lockName);
- lockRegistryService.releaseLock(decorateLockName(lockName));
+ lockContext.getOrCreateStandardLock(decorateLockName(lockName)).releaseLock(lockName);
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
index 18d8b222087..0138c104663 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparer.java
@@ -112,7 +112,7 @@ public final class RuleAlteredJobPreparer {
private void waitUntilLockReleased(final ShardingSphereLock lock, final String lockName) {
for (int loopCount = 0; loopCount < 30; loopCount++) {
ThreadUtil.sleep(TimeUnit.SECONDS.toMillis(5));
- if (!lock.isLocked(lockName)) {
+ if (!lock.isLocked()) {
log.info("unlocked, lockName={}", lockName);
return;
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
index adce67b2ce1..c2cf39f75e2 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorker.java
@@ -338,7 +338,7 @@ public final class RuleAlteredJobWorker {
log.info("restoreSourceWriting, databaseName={}", databaseName);
LockContext lockContext = PipelineContext.getContextManager().getInstanceContext().getLockContext();
ShardingSphereLock lock = lockContext.getGlobalLock(databaseName);
- if (null != lock && lock.isLocked(databaseName)) {
+ if (null != lock && lock.isLocked()) {
log.info("Source writing is still stopped on database '{}', restore it now", databaseName);
lock.releaseLock(databaseName);
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContext.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContext.java
index 4d91e2e7cde..119d2781c16 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContext.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContext.java
@@ -97,6 +97,11 @@ public final class DistributeLockContext implements LockContext {
return lockManagers.get(LockType.GENERAL).getOrCreateLock(lockName);
}
+ @Override
+ public synchronized ShardingSphereLock getOrCreateStandardLock(final String lockName) {
+ return lockManagers.get(LockType.STANDARD).getOrCreateLock(lockName);
+ }
+
@Override
public ShardingSphereLock getGlobalLock(final String lockName) {
Preconditions.checkNotNull(lockName, "Get global lock args lock name can not be null.");
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/LockNodeService.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/LockNodeService.java
index 6b870d166d5..4c9fe845b4d 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/LockNodeService.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/LockNodeService.java
@@ -40,51 +40,51 @@ public interface LockNodeService {
String getSequenceNodePath();
/**
- * Get global locks node path.
+ * Get locks node path.
*
- * @return global lock node path
+ * @return locks node path
*/
- String getGlobalLocksNodePath();
+ String getLocksNodePath();
/**
- * Get global locked ack node path.
+ * Generate locks name.
*
- * @return global locked ack node path
+ * @param locksName locks name
+ * @return locks name
*/
- String getGlobalLockedAckNodePath();
+ String generateLocksName(String locksName);
/**
- * Generate global locks name.
+ * Get locked ack node path.
*
- * @param locksName locks name
- * @return global locks name
+ * @return locked ack node path
*/
- String generateGlobalLocksName(String locksName);
+ String getLockedAckNodePath();
/**
* Generate ack lock name.
*
* @param ackLockName ack lock name
* @param lockedInstanceId locked instance id
- * @return global ack lock name
+ * @return ack lock name
*/
- String generateGlobalAckLockName(String ackLockName, String lockedInstanceId);
+ String generateAckLockName(String ackLockName, String lockedInstanceId);
/**
- * Parse global Locks node path.
+ * Parse Locks node path.
*
* @param nodePath locks node path
- * @return global locked node path
+ * @return locked node path
*/
- Optional<String> parseGlobalLocksNodePath(String nodePath);
+ Optional<String> parseLocksNodePath(String nodePath);
/**
- * Parse global locked ack node path.
+ * Parse locked ack node path.
*
* @param nodePath locked ack node path
- * @return global locked ack node path
+ * @return locked ack node path
*/
- Optional<String> parseGlobalLockedAckNodePath(String nodePath);
+ Optional<String> parseLockedAckNodePath(String nodePath);
/**
* Get lock type.
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereDistributeGlobalLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereDistributeGlobalLock.java
index 50a637db55d..f359fa10bed 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereDistributeGlobalLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereDistributeGlobalLock.java
@@ -133,7 +133,7 @@ public final class ShardingSphereDistributeGlobalLock implements ShardingSphereG
}
@Override
- public boolean isLocked(final String lockName) {
+ public boolean isLocked() {
return LockState.LOCKED == synchronizedLockState.get();
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java
index f958527a3c2..c521924c388 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java
@@ -49,7 +49,7 @@ public final class ShardingSphereSequencedSemaphoreLock implements ShardingSpher
}
@Override
- public boolean isLocked(final String lockName) {
- return false;
+ public boolean isLocked() {
+ throw new UnsupportedOperationException();
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLock.java
index 116ec8735b4..adfd8a2caa3 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLock.java
@@ -21,7 +21,6 @@ import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.infra.lock.ShardingSphereGlobalLock;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.ShardingSphereDistributeGlobalLock;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.DatabaseLockNodeService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.GlobalLockRegistryService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.util.TimeoutMilliseconds;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
@@ -33,11 +32,13 @@ import java.util.Collection;
*/
public final class ShardingSphereDatabaseLock implements ShardingSphereGlobalLock {
- private final LockNodeService lockNodeService = new DatabaseLockNodeService();
+ private final LockNodeService lockNodeService;
private final ShardingSphereDistributeGlobalLock innerDistributeGlobalLock;
- public ShardingSphereDatabaseLock(final ClusterPersistRepository clusterRepository, final ComputeNodeInstance currentInstance, final Collection<ComputeNodeInstance> computeNodeInstances) {
+ public ShardingSphereDatabaseLock(final ClusterPersistRepository clusterRepository, final LockNodeService lockNodeService, final ComputeNodeInstance currentInstance,
+ final Collection<ComputeNodeInstance> computeNodeInstances) {
+ this.lockNodeService = lockNodeService;
innerDistributeGlobalLock = new ShardingSphereDistributeGlobalLock(new GlobalLockRegistryService(clusterRepository), currentInstance, computeNodeInstances);
}
@@ -48,27 +49,27 @@ public final class ShardingSphereDatabaseLock implements ShardingSphereGlobalLoc
@Override
public boolean tryLock(final String lockName, final long timeoutMillis) {
- return innerDistributeGlobalLock.tryLock(lockNodeService.generateGlobalLocksName(lockName), timeoutMillis);
+ return innerDistributeGlobalLock.tryLock(lockNodeService.generateLocksName(lockName), timeoutMillis);
}
@Override
public void releaseLock(final String lockName) {
- innerDistributeGlobalLock.releaseLock(lockNodeService.generateGlobalLocksName(lockName));
+ innerDistributeGlobalLock.releaseLock(lockNodeService.generateLocksName(lockName));
}
@Override
- public boolean isLocked(final String lockName) {
- return innerDistributeGlobalLock.isLocked(lockName);
+ public boolean isLocked() {
+ return innerDistributeGlobalLock.isLocked();
}
@Override
public void ackLock(final String lockName, final String lockedInstanceId) {
- innerDistributeGlobalLock.ackLock(lockNodeService.generateGlobalAckLockName(lockName, lockedInstanceId), lockedInstanceId);
+ innerDistributeGlobalLock.ackLock(lockNodeService.generateAckLockName(lockName, lockedInstanceId), lockedInstanceId);
}
@Override
public void releaseAckLock(final String lockName, final String lockedInstanceId) {
- innerDistributeGlobalLock.releaseAckLock(lockNodeService.generateGlobalAckLockName(lockName, lockedInstanceId), lockedInstanceId);
+ innerDistributeGlobalLock.releaseAckLock(lockNodeService.generateAckLockName(lockName, lockedInstanceId), lockedInstanceId);
}
@Override
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLockManager.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLockManager.java
index 0ec16ec895e..2dc831eb10b 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLockManager.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/ShardingSphereDatabaseLockManager.java
@@ -68,20 +68,20 @@ public final class ShardingSphereDatabaseLockManager implements ShardingSphereLo
}
private void synchronizeGlobalLock() {
- Collection<String> allGlobalLock = clusterRepository.getChildrenKeys(lockNodeService.getGlobalLocksNodePath());
+ Collection<String> allGlobalLock = clusterRepository.getChildrenKeys(lockNodeService.getLocksNodePath());
if (allGlobalLock.isEmpty()) {
- clusterRepository.persist(lockNodeService.getGlobalLocksNodePath(), "");
- clusterRepository.persist(lockNodeService.getGlobalLockedAckNodePath(), "");
+ clusterRepository.persist(lockNodeService.getLocksNodePath(), "");
+ clusterRepository.persist(lockNodeService.getLockedAckNodePath(), "");
return;
}
for (String each : allGlobalLock) {
- Optional<String> databaseLock = lockNodeService.parseGlobalLocksNodePath(each);
+ Optional<String> databaseLock = lockNodeService.parseLocksNodePath(each);
databaseLock.ifPresent(database -> locks.put(database, crateDatabaseLock()));
}
}
private ShardingSphereDatabaseLock crateDatabaseLock() {
- return new ShardingSphereDatabaseLock(clusterRepository, currentInstance, computeNodeInstances);
+ return new ShardingSphereDatabaseLock(clusterRepository, lockNodeService, currentInstance, computeNodeInstances);
}
@Override
@@ -107,7 +107,7 @@ public final class ShardingSphereDatabaseLockManager implements ShardingSphereLo
}
ShardingSphereGlobalLock lock = locks.get(lockName);
if (null != lock) {
- return lock.isLocked(lockName);
+ return lock.isLocked();
}
return false;
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseAckChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseAckChangedWatcher.java
index 7c0e5365abb..19fe5f20404 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseAckChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseAckChangedWatcher.java
@@ -41,7 +41,7 @@ public final class DatabaseAckChangedWatcher implements GovernanceWatcher<Govern
@Override
public Collection<String> getWatchingKeys() {
- return Collections.singleton(lockNode.getGlobalLockedAckNodePath());
+ return Collections.singleton(lockNode.getLockedAckNodePath());
}
@Override
@@ -51,7 +51,7 @@ public final class DatabaseAckChangedWatcher implements GovernanceWatcher<Govern
@Override
public Optional<GovernanceEvent> createGovernanceEvent(final DataChangedEvent event) {
- Optional<String> ackLockedName = lockNode.parseGlobalLockedAckNodePath(event.getKey());
+ Optional<String> ackLockedName = lockNode.parseLockedAckNodePath(event.getKey());
if (ackLockedName.isPresent()) {
return handleGlobalAckEvent(event.getType(), ackLockedName.get());
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseLocksChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseLocksChangedWatcher.java
index 7c80d864657..49d5a9594aa 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseLocksChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/database/watcher/DatabaseLocksChangedWatcher.java
@@ -41,7 +41,7 @@ public final class DatabaseLocksChangedWatcher implements GovernanceWatcher<Gove
@Override
public Collection<String> getWatchingKeys() {
- return Collections.singleton(lockNode.getGlobalLocksNodePath());
+ return Collections.singleton(lockNode.getLocksNodePath());
}
@Override
@@ -51,7 +51,7 @@ public final class DatabaseLocksChangedWatcher implements GovernanceWatcher<Gove
@Override
public Optional<GovernanceEvent> createGovernanceEvent(final DataChangedEvent event) {
- Optional<String> lockedName = lockNode.parseGlobalLocksNodePath(event.getKey());
+ Optional<String> lockedName = lockNode.parseLocksNodePath(event.getKey());
if (lockedName.isPresent()) {
return handleGlobalSchemaLocksEvent(event.getType(), lockedName.get());
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLock.java
index 1e3050b478d..15bdad69588 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLock.java
@@ -19,11 +19,11 @@ package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.g
import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.infra.lock.ShardingSphereGlobalLock;
+import org.apache.shardingsphere.infra.lock.ShardingSphereLock;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockRegistryService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.ShardingSphereDistributeGlobalLock;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.ShardingSphereSequencedSemaphoreLock;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.GeneralLockNodeService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.GlobalLockRegistryService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.util.TimeoutMilliseconds;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
@@ -37,13 +37,14 @@ public final class ShardingSphereGeneralLock implements ShardingSphereGlobalLock
private final LockNodeService lockNodeService;
- private final ShardingSphereSequencedSemaphoreLock sequencedSemaphoreLock;
+ private final ShardingSphereLock sequencedSemaphoreLock;
private final ShardingSphereDistributeGlobalLock innerDistributeGlobalLock;
- public ShardingSphereGeneralLock(final ClusterPersistRepository clusterRepository, final ComputeNodeInstance currentInstance, final Collection<ComputeNodeInstance> computeNodeInstances) {
+ public ShardingSphereGeneralLock(final ClusterPersistRepository clusterRepository, final LockNodeService lockNodeService, final ComputeNodeInstance currentInstance,
+ final Collection<ComputeNodeInstance> computeNodeInstances) {
LockRegistryService lockRegistryService = new GlobalLockRegistryService(clusterRepository);
- lockNodeService = new GeneralLockNodeService();
+ this.lockNodeService = lockNodeService;
innerDistributeGlobalLock = new ShardingSphereDistributeGlobalLock(lockRegistryService, currentInstance, computeNodeInstances);
sequencedSemaphoreLock = new ShardingSphereSequencedSemaphoreLock(lockRegistryService, lockNodeService);
}
@@ -64,7 +65,7 @@ public final class ShardingSphereGeneralLock implements ShardingSphereGlobalLock
return false;
}
try {
- return innerDistributeGlobalLock.tryLock(lockNodeService.generateGlobalLocksName(lockName), timeoutMillis);
+ return innerDistributeGlobalLock.tryLock(lockNodeService.generateLocksName(lockName), timeoutMillis);
} finally {
sequencedSemaphoreLock.releaseLock(lockName);
}
@@ -72,22 +73,22 @@ public final class ShardingSphereGeneralLock implements ShardingSphereGlobalLock
@Override
public void releaseLock(final String lockName) {
- innerDistributeGlobalLock.releaseLock(lockNodeService.generateGlobalLocksName(lockName));
+ innerDistributeGlobalLock.releaseLock(lockNodeService.generateLocksName(lockName));
}
@Override
- public boolean isLocked(final String lockName) {
- return innerDistributeGlobalLock.isLocked(lockName);
+ public boolean isLocked() {
+ return innerDistributeGlobalLock.isLocked();
}
@Override
public void ackLock(final String lockName, final String instanceId) {
- innerDistributeGlobalLock.ackLock(lockNodeService.generateGlobalAckLockName(lockName, instanceId), instanceId);
+ innerDistributeGlobalLock.ackLock(lockNodeService.generateAckLockName(lockName, instanceId), instanceId);
}
@Override
public void releaseAckLock(final String lockName, final String instanceId) {
- innerDistributeGlobalLock.releaseAckLock(lockNodeService.generateGlobalAckLockName(lockName, instanceId), instanceId);
+ innerDistributeGlobalLock.releaseAckLock(lockNodeService.generateAckLockName(lockName, instanceId), instanceId);
}
@Override
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java
index b66bf2d5277..374dbf45e20 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java
@@ -63,20 +63,20 @@ public final class ShardingSphereGeneralLockManager implements ShardingSphereLoc
}
private void synchronizeGlobalLock() {
- Collection<String> allGlobalLock = clusterRepository.getChildrenKeys(lockNodeService.getGlobalLocksNodePath());
+ Collection<String> allGlobalLock = clusterRepository.getChildrenKeys(lockNodeService.getLocksNodePath());
if (allGlobalLock.isEmpty()) {
- clusterRepository.persist(lockNodeService.getGlobalLocksNodePath(), "");
- clusterRepository.persist(lockNodeService.getGlobalLockedAckNodePath(), "");
+ clusterRepository.persist(lockNodeService.getLocksNodePath(), "");
+ clusterRepository.persist(lockNodeService.getLockedAckNodePath(), "");
return;
}
for (String each : allGlobalLock) {
- Optional<String> generalLock = lockNodeService.parseGlobalLocksNodePath(each);
- generalLock.ifPresent(lockName -> locks.put(lockName, crateGeneralLock()));
+ Optional<String> generalLock = lockNodeService.parseLocksNodePath(each);
+ generalLock.ifPresent(lockName -> locks.put(lockName, createGeneralLock()));
}
}
- private ShardingSphereGeneralLock crateGeneralLock() {
- return new ShardingSphereGeneralLock(clusterRepository, currentInstance, computeNodeInstances);
+ private ShardingSphereGeneralLock createGeneralLock() {
+ return new ShardingSphereGeneralLock(clusterRepository, lockNodeService, currentInstance, computeNodeInstances);
}
@Override
@@ -85,7 +85,7 @@ public final class ShardingSphereGeneralLockManager implements ShardingSphereLoc
if (null != result) {
return result;
}
- result = crateGeneralLock();
+ result = createGeneralLock();
locks.put(lockName, result);
return result;
}
@@ -102,7 +102,7 @@ public final class ShardingSphereGeneralLockManager implements ShardingSphereLoc
}
ShardingSphereGlobalLock lock = locks.get(lockName);
if (null != lock) {
- return lock.isLocked(lockName);
+ return lock.isLocked();
}
return false;
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralAckChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralAckChangedWatcher.java
index cb6e134a97d..d63a4bb7b0e 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralAckChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralAckChangedWatcher.java
@@ -41,7 +41,7 @@ public final class GeneralAckChangedWatcher implements GovernanceWatcher<Governa
@Override
public Collection<String> getWatchingKeys() {
- return Collections.singleton(lockNode.getGlobalLockedAckNodePath());
+ return Collections.singleton(lockNode.getLockedAckNodePath());
}
@Override
@@ -51,7 +51,7 @@ public final class GeneralAckChangedWatcher implements GovernanceWatcher<Governa
@Override
public Optional<GovernanceEvent> createGovernanceEvent(final DataChangedEvent event) {
- Optional<String> ackLockedName = lockNode.parseGlobalLockedAckNodePath(event.getKey());
+ Optional<String> ackLockedName = lockNode.parseLockedAckNodePath(event.getKey());
if (ackLockedName.isPresent()) {
return handleGlobalAckEvent(event.getType(), ackLockedName.get());
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralLocksChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralLocksChangedWatcher.java
index 1a9ba9b3574..9599fc4f8f6 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralLocksChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/watcher/GeneralLocksChangedWatcher.java
@@ -41,7 +41,7 @@ public final class GeneralLocksChangedWatcher implements GovernanceWatcher<Gover
@Override
public Collection<String> getWatchingKeys() {
- return Collections.singleton(lockNode.getGlobalLocksNodePath());
+ return Collections.singleton(lockNode.getLocksNodePath());
}
@Override
@@ -51,7 +51,7 @@ public final class GeneralLocksChangedWatcher implements GovernanceWatcher<Gover
@Override
public Optional<GovernanceEvent> createGovernanceEvent(final DataChangedEvent event) {
- Optional<String> lockedName = lockNode.parseGlobalLocksNodePath(event.getKey());
+ Optional<String> lockedName = lockNode.parseLocksNodePath(event.getKey());
if (lockedName.isPresent()) {
return handleGlobalSchemaLocksEvent(event.getType(), lockedName.get());
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeService.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeService.java
index 5bd7d01c7a1..e8a300330c7 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeService.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeService.java
@@ -34,35 +34,35 @@ public abstract class AbstractGlobalLockNodeService implements LockNodeService {
protected static final String LOCKED_ACK_NODE = "ack";
@Override
- public String getGlobalLocksNodePath() {
+ public String getLocksNodePath() {
return PATH_DELIMITER + LOCK_ROOT + PATH_DELIMITER + LOCK_SCOPE_GLOBAL + PATH_DELIMITER + getLockLevel() + PATH_DELIMITER + LOCKS_NODE;
}
@Override
- public String getGlobalLockedAckNodePath() {
+ public String getLockedAckNodePath() {
return PATH_DELIMITER + LOCK_ROOT + PATH_DELIMITER + LOCK_SCOPE_GLOBAL + PATH_DELIMITER + getLockLevel() + PATH_DELIMITER + LOCKED_ACK_NODE;
}
@Override
- public String generateGlobalLocksName(final String locksName) {
- return getGlobalLocksNodePath() + "/" + locksName;
+ public String generateLocksName(final String locksName) {
+ return getLocksNodePath() + "/" + locksName;
}
@Override
- public String generateGlobalAckLockName(final String ackLockName, final String lockedInstanceId) {
- return getGlobalLockedAckNodePath() + "/" + LockNodeUtil.generateAckLockedName(ackLockName, lockedInstanceId);
+ public String generateAckLockName(final String ackLockName, final String lockedInstanceId) {
+ return getLockedAckNodePath() + "/" + LockNodeUtil.generateAckLockedName(ackLockName, lockedInstanceId);
}
@Override
- public Optional<String> parseGlobalLocksNodePath(final String nodePath) {
- Pattern pattern = Pattern.compile(getGlobalLocksNodePath() + "/" + "(.+)/leases/(.+)$", Pattern.CASE_INSENSITIVE);
+ public Optional<String> parseLocksNodePath(final String nodePath) {
+ Pattern pattern = Pattern.compile(getLocksNodePath() + "/" + "(.+)/leases/(.+)$", Pattern.CASE_INSENSITIVE);
Matcher matcher = pattern.matcher(nodePath);
return matcher.find() ? Optional.of(matcher.group(1)) : Optional.empty();
}
@Override
- public Optional<String> parseGlobalLockedAckNodePath(final String nodePath) {
- Pattern pattern = Pattern.compile(getGlobalLockedAckNodePath() + "/(.+)$", Pattern.CASE_INSENSITIVE);
+ public Optional<String> parseLockedAckNodePath(final String nodePath) {
+ Pattern pattern = Pattern.compile(getLockedAckNodePath() + "/(.+)$", Pattern.CASE_INSENSITIVE);
Matcher matcher = pattern.matcher(nodePath);
return matcher.find() ? Optional.of(matcher.group(1)) : Optional.empty();
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryService.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryService.java
index b695458e343..5cd68b07170 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryService.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryService.java
@@ -34,12 +34,16 @@ public final class GlobalLockRegistryService implements LockRegistryService {
@Override
public boolean tryLock(final String lockName, final long timeoutMilliseconds) {
- return repository.tryLock(lockName, timeoutMilliseconds, TimeUnit.MILLISECONDS);
+ try {
+ return repository.getGlobalLock(lockName).tryLock(timeoutMilliseconds, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException ignore) {
+ return false;
+ }
}
@Override
public void releaseLock(final String lockName) {
- repository.releaseLock(lockName);
+ repository.getGlobalLock(lockName).unlock();
}
@Override
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLock.java
similarity index 72%
copy from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java
copy to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLock.java
index f958527a3c2..27a2a8c865d 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/ShardingSphereSequencedSemaphoreLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLock.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global;
+package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.lock.ShardingSphereLock;
@@ -24,32 +24,32 @@ import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockRegis
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.util.TimeoutMilliseconds;
/**
- * Sharding sphere sequenced semaphore lock.
+ * Standard distribute lock of ShardingSphere.
*/
@RequiredArgsConstructor
-public final class ShardingSphereSequencedSemaphoreLock implements ShardingSphereLock {
+public final class ShardingSphereStandardLock implements ShardingSphereLock {
- private final LockRegistryService lockService;
+ private final LockRegistryService lockRegistryService;
- private final LockNodeService lockNode;
+ private final LockNodeService lockNodeService;
@Override
public boolean tryLock(final String lockName) {
- return tryLock(lockName, TimeoutMilliseconds.DEFAULT_REGISTRY);
+ return tryLock(lockName, TimeoutMilliseconds.MAX_TRY_LOCK);
}
@Override
public boolean tryLock(final String lockName, final long timeoutMillis) {
- return lockService.tryLock(lockNode.getSequenceNodePath(), timeoutMillis);
+ return lockRegistryService.tryLock(lockNodeService.generateLocksName(lockName), TimeoutMilliseconds.MAX_TRY_LOCK);
}
@Override
public void releaseLock(final String lockName) {
- lockService.releaseLock(lockNode.getSequenceNodePath());
+ lockRegistryService.releaseLock(lockNodeService.generateLocksName(lockName));
}
@Override
- public boolean isLocked(final String lockName) {
- return false;
+ public boolean isLocked() {
+ throw new UnsupportedOperationException();
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLockManager.java
similarity index 59%
copy from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java
copy to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLockManager.java
index b66bf2d5277..f96a559e33e 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/general/ShardingSphereGeneralLockManager.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/ShardingSphereStandardLockManager.java
@@ -15,40 +15,31 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.general;
+package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard;
-import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.infra.lock.LockType;
-import org.apache.shardingsphere.infra.lock.ShardingSphereGlobalLock;
import org.apache.shardingsphere.infra.lock.ShardingSphereLock;
import org.apache.shardingsphere.mode.manager.ShardingSphereLockManager;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeServiceFactory;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard.service.StandardLockRegistryService;
import org.apache.shardingsphere.mode.persist.PersistRepository;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import java.util.Collection;
import java.util.Map;
-import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
-/**
- * General lock manager of ShardingSphere.
- */
-public final class ShardingSphereGeneralLockManager implements ShardingSphereLockManager {
+public final class ShardingSphereStandardLockManager implements ShardingSphereLockManager {
- private final Map<String, ShardingSphereGeneralLock> locks;
+ private final Map<String, ShardingSphereStandardLock> locks;
private final LockNodeService lockNodeService;
private ClusterPersistRepository clusterRepository;
- private ComputeNodeInstance currentInstance;
-
- private Collection<ComputeNodeInstance> computeNodeInstances;
-
- public ShardingSphereGeneralLockManager() {
+ public ShardingSphereStandardLockManager() {
locks = new ConcurrentHashMap<>();
lockNodeService = LockNodeServiceFactory.getInstance().getLockNodeService(getLockType());
}
@@ -56,36 +47,19 @@ public final class ShardingSphereGeneralLockManager implements ShardingSphereLoc
@Override
public void initLocksState(final PersistRepository repository, final ComputeNodeInstance instance, final Collection<ComputeNodeInstance> computeNodeInstances) {
clusterRepository = (ClusterPersistRepository) repository;
- currentInstance = instance;
- this.computeNodeInstances = computeNodeInstances;
- ShardingSphereEventBus.getInstance().register(this);
- synchronizeGlobalLock();
- }
-
- private void synchronizeGlobalLock() {
- Collection<String> allGlobalLock = clusterRepository.getChildrenKeys(lockNodeService.getGlobalLocksNodePath());
- if (allGlobalLock.isEmpty()) {
- clusterRepository.persist(lockNodeService.getGlobalLocksNodePath(), "");
- clusterRepository.persist(lockNodeService.getGlobalLockedAckNodePath(), "");
- return;
- }
- for (String each : allGlobalLock) {
- Optional<String> generalLock = lockNodeService.parseGlobalLocksNodePath(each);
- generalLock.ifPresent(lockName -> locks.put(lockName, crateGeneralLock()));
- }
}
- private ShardingSphereGeneralLock crateGeneralLock() {
- return new ShardingSphereGeneralLock(clusterRepository, currentInstance, computeNodeInstances);
+ private ShardingSphereStandardLock createGeneralLock() {
+ return new ShardingSphereStandardLock(new StandardLockRegistryService(clusterRepository), lockNodeService);
}
@Override
public ShardingSphereLock getOrCreateLock(final String lockName) {
- ShardingSphereGeneralLock result = locks.get(lockName);
+ ShardingSphereStandardLock result = locks.get(lockName);
if (null != result) {
return result;
}
- result = crateGeneralLock();
+ result = createGeneralLock();
locks.put(lockName, result);
return result;
}
@@ -100,15 +74,15 @@ public final class ShardingSphereGeneralLockManager implements ShardingSphereLoc
if (locks.isEmpty()) {
return false;
}
- ShardingSphereGlobalLock lock = locks.get(lockName);
+ ShardingSphereLock lock = locks.get(lockName);
if (null != lock) {
- return lock.isLocked(lockName);
+ return lock.isLocked();
}
return false;
}
@Override
public LockType getLockType() {
- return LockType.GENERAL;
+ return LockType.STANDARD;
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockNodeService.java
similarity index 51%
copy from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java
copy to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockNodeService.java
index 3acf4715317..d6551bc509d 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockNodeService.java
@@ -17,54 +17,52 @@
package org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard.service;
-import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockRegistryService;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.infra.lock.LockType;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService;
-import java.util.concurrent.TimeUnit;
+import java.util.Optional;
-/**
- * Standard lock registry service.
- */
-@RequiredArgsConstructor
-public final class StandardLockRegistryService implements LockRegistryService {
-
- private static final String PATH_DELIMITER = "/";
-
- private static final String LOCK_ROOT = "lock";
+public final class StandardLockNodeService implements LockNodeService {
- private static final String LOCKS_NODE = "locks";
+ protected static final String LOCK_SCOPE_STANDARD = "standard";
- private static final String LOCK_SCOPE_STANDARD = "standard";
-
- private final ClusterPersistRepository repository;
+ @Override
+ public String getSequenceNodePath() {
+ throw new UnsupportedOperationException();
+ }
@Override
- public boolean tryLock(final String lockName, final long timeoutMilliseconds) {
- return repository.tryLock(generateStandardLockName(lockName), timeoutMilliseconds, TimeUnit.MILLISECONDS);
+ public String getLocksNodePath() {
+ return PATH_DELIMITER + LOCK_ROOT + PATH_DELIMITER + LOCK_SCOPE_STANDARD + PATH_DELIMITER + LOCKS_NODE;
}
@Override
- public void releaseLock(final String lockName) {
- repository.releaseLock(generateStandardLockName(lockName));
+ public String generateLocksName(final String locksName) {
+ return getLocksNodePath() + "/" + locksName;
}
- private String generateStandardLockName(final String lockName) {
- return PATH_DELIMITER + LOCK_ROOT + PATH_DELIMITER + LOCK_SCOPE_STANDARD + PATH_DELIMITER + LOCKS_NODE + PATH_DELIMITER + lockName;
+ @Override
+ public String getLockedAckNodePath() {
+ throw new UnsupportedOperationException();
}
@Override
- public void removeLock(final String lockName) {
+ public String generateAckLockName(final String ackLockName, final String lockedInstanceId) {
throw new UnsupportedOperationException();
}
@Override
- public void ackLock(final String lockName, final String lockValue) {
+ public Optional<String> parseLocksNodePath(final String nodePath) {
throw new UnsupportedOperationException();
}
@Override
- public void releaseAckLock(final String lockName) {
+ public Optional<String> parseLockedAckNodePath(final String nodePath) {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public LockType getLockType() {
+ return LockType.STANDARD;
+ }
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java
index 3acf4715317..8ac7d6dde9d 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryService.java
@@ -29,28 +29,20 @@ import java.util.concurrent.TimeUnit;
@RequiredArgsConstructor
public final class StandardLockRegistryService implements LockRegistryService {
- private static final String PATH_DELIMITER = "/";
-
- private static final String LOCK_ROOT = "lock";
-
- private static final String LOCKS_NODE = "locks";
-
- private static final String LOCK_SCOPE_STANDARD = "standard";
-
private final ClusterPersistRepository repository;
@Override
public boolean tryLock(final String lockName, final long timeoutMilliseconds) {
- return repository.tryLock(generateStandardLockName(lockName), timeoutMilliseconds, TimeUnit.MILLISECONDS);
+ try {
+ return repository.getStandardLock(lockName).tryLock(timeoutMilliseconds, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException ignore) {
+ return false;
+ }
}
@Override
public void releaseLock(final String lockName) {
- repository.releaseLock(generateStandardLockName(lockName));
- }
-
- private String generateStandardLockName(final String lockName) {
- return PATH_DELIMITER + LOCK_ROOT + PATH_DELIMITER + LOCK_SCOPE_STANDARD + PATH_DELIMITER + LOCKS_NODE + PATH_DELIMITER + lockName;
+ repository.getStandardLock(lockName).unlock();
}
@Override
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.ShardingSphereLockManager b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.ShardingSphereLockManager
index 640c57dc8f0..9abe5313f40 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.ShardingSphereLockManager
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.ShardingSphereLockManager
@@ -17,3 +17,4 @@
org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.database.ShardingSphereDatabaseLockManager
org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.general.ShardingSphereGeneralLockManager
+org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard.ShardingSphereStandardLockManager
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService
index 18c63116f20..96824f602b4 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.LockNodeService
@@ -17,3 +17,4 @@
org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.GeneralLockNodeService
org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.global.service.DatabaseLockNodeService
+org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.standard.service.StandardLockNodeService
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/fixture/ClusterPersistRepositoryFixture.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/fixture/ClusterPersistRepositoryFixture.java
index ade212aa3b6..db94a6f6b58 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/fixture/ClusterPersistRepositoryFixture.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/fixture/ClusterPersistRepositoryFixture.java
@@ -25,6 +25,7 @@ import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEve
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
public final class ClusterPersistRepositoryFixture implements ClusterPersistRepository {
@@ -76,6 +77,16 @@ public final class ClusterPersistRepositoryFixture implements ClusterPersistRepo
public void watchSessionConnection(final InstanceContext instanceContext) {
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return null;
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return null;
+ }
+
@Override
public void close() {
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContextTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContextTest.java
index 7289563177f..266af762746 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContextTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/DistributeLockContextTest.java
@@ -42,7 +42,6 @@ public final class DistributeLockContextTest {
instanceContext.initLockContext();
ShardingSphereLock databaseLock = distributeLockContext.getOrCreateGlobalLock("database");
assertNotNull(databaseLock);
- assertTrue(distributeLockContext.tryLockWriteDatabase("database", 1000));
}
@Test
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeServiceTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalStandardLockNodeServiceTest.java
similarity index 82%
rename from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeServiceTest.java
rename to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalStandardLockNodeServiceTest.java
index bcd6cd49e72..50036869342 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalLockNodeServiceTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/AbstractGlobalStandardLockNodeServiceTest.java
@@ -25,38 +25,38 @@ import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-public final class AbstractGlobalLockNodeServiceTest {
+public final class AbstractGlobalStandardLockNodeServiceTest {
private static final AbstractGlobalLockNodeService SERVICE = new DatabaseLockNodeService();
@Test
public void assertGetGlobalLocksNodePath() {
- String globalLocksNodePath = SERVICE.getGlobalLocksNodePath();
+ String globalLocksNodePath = SERVICE.getLocksNodePath();
assertThat(globalLocksNodePath, is("/lock/global/database/locks"));
}
@Test
public void assertGetGlobalLockedAckNodePath() {
- String globalLockedAckNodePath = SERVICE.getGlobalLockedAckNodePath();
+ String globalLockedAckNodePath = SERVICE.getLockedAckNodePath();
assertThat(globalLockedAckNodePath, is("/lock/global/database/ack"));
}
@Test
public void assertGenerateGlobalLocksName() {
- String locksName = SERVICE.generateGlobalLocksName("sharding_db");
+ String locksName = SERVICE.generateLocksName("sharding_db");
assertThat(locksName, is("/lock/global/database/locks/sharding_db"));
}
@Test
public void assertGenerateGlobalAckLockName() {
- String globalLockedAckNodePath = SERVICE.generateGlobalAckLockName("locksName", "127.0.0.1@3307");
+ String globalLockedAckNodePath = SERVICE.generateAckLockName("locksName", "127.0.0.1@3307");
assertThat(globalLockedAckNodePath, is("/lock/global/database/ack/locksName#@#127.0.0.1@3307"));
}
@Test
public void assertParseGlobalLocksNodePath() {
String nodePath = "/lock/global/database/locks/sharding_db/leases/c_l_00000000";
- Optional<String> globalLocksNodePath = SERVICE.parseGlobalLocksNodePath(nodePath);
+ Optional<String> globalLocksNodePath = SERVICE.parseLocksNodePath(nodePath);
assertTrue(globalLocksNodePath.isPresent());
assertThat(globalLocksNodePath.get(), is("sharding_db"));
}
@@ -64,7 +64,7 @@ public final class AbstractGlobalLockNodeServiceTest {
@Test
public void assertParseGlobalLockedAckNodePath() {
String nodePath = "/lock/global/database/ack/sharding_db#@#127.0.0.1@3307";
- Optional<String> globalLockedAckNodePath = SERVICE.parseGlobalLockedAckNodePath(nodePath);
+ Optional<String> globalLockedAckNodePath = SERVICE.parseLockedAckNodePath(nodePath);
assertTrue(globalLockedAckNodePath.isPresent());
assertThat(globalLockedAckNodePath.get(), is("sharding_db#@#127.0.0.1@3307"));
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseLockNodeServiceTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseStandardLockNodeServiceTest.java
similarity index 96%
rename from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseLockNodeServiceTest.java
rename to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseStandardLockNodeServiceTest.java
index 630395eff22..15d07f59452 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseLockNodeServiceTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/DatabaseStandardLockNodeServiceTest.java
@@ -22,7 +22,7 @@ import org.junit.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
-public final class DatabaseLockNodeServiceTest {
+public final class DatabaseStandardLockNodeServiceTest {
private static final AbstractGlobalLockNodeService SERVICE = new DatabaseLockNodeService();
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralLockNodeServiceTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralStandardLockNodeServiceTest.java
similarity index 96%
rename from shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralLockNodeServiceTest.java
rename to shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralStandardLockNodeServiceTest.java
index 25eb5ff6271..e87966dc90f 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralLockNodeServiceTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GeneralStandardLockNodeServiceTest.java
@@ -22,7 +22,7 @@ import org.junit.Test;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
-public final class GeneralLockNodeServiceTest {
+public final class GeneralStandardLockNodeServiceTest {
private static final AbstractGlobalLockNodeService SERVICE = new GeneralLockNodeService();
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryServiceTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryServiceTest.java
index 3d4c9622db6..30282b96cd4 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryServiceTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/global/service/GlobalLockRegistryServiceTest.java
@@ -27,7 +27,6 @@ import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import java.lang.reflect.Field;
-import java.util.concurrent.TimeUnit;
import static org.mockito.Mockito.verify;
@@ -47,18 +46,6 @@ public final class GlobalLockRegistryServiceTest {
field.set(lockRegistryService, clusterPersistRepository);
}
- @Test
- public void assertTryLock() {
- lockRegistryService.tryLock("test", 50L);
- verify(clusterPersistRepository).tryLock("test", 50L, TimeUnit.MILLISECONDS);
- }
-
- @Test
- public void assertReleaseLock() {
- lockRegistryService.releaseLock("test");
- verify(clusterPersistRepository).releaseLock("test");
- }
-
@Test
public void assertRemoveLock() {
lockRegistryService.removeLock("test");
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryServiceTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryServiceTest.java
index 61f43ac0f68..c2ee0859eda 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryServiceTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/lock/standard/service/StandardLockRegistryServiceTest.java
@@ -26,7 +26,6 @@ import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import java.lang.reflect.Field;
-import java.util.concurrent.TimeUnit;
import static org.mockito.Mockito.verify;
@@ -46,18 +45,6 @@ public final class StandardLockRegistryServiceTest {
field.set(lockRegistryService, clusterPersistRepository);
}
- @Test
- public void assertTryLock() {
- lockRegistryService.tryLock("test", 50L);
- verify(clusterPersistRepository).tryLock("/lock/standard/locks/test", 50L, TimeUnit.MILLISECONDS);
- }
-
- @Test
- public void assertReleaseLock() {
- lockRegistryService.releaseLock("test");
- verify(clusterPersistRepository).releaseLock("/lock/standard/locks/test");
- }
-
@Test(expected = UnsupportedOperationException.class)
public void assertRemoveLock() {
lockRegistryService.removeLock("test");
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/ProcessListClusterPersistRepositoryFixture.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/ProcessListClusterPersistRepositoryFix [...]
index ab6b2128933..401e3d94ba7 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/ProcessListClusterPersistRepositoryFixture.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/ProcessListClusterPersistRepositoryFixture.java
@@ -27,6 +27,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
public final class ProcessListClusterPersistRepositoryFixture implements ClusterPersistRepository {
@@ -83,6 +84,16 @@ public final class ProcessListClusterPersistRepositoryFixture implements Cluster
public void watchSessionConnection(final InstanceContext instanceContext) {
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return null;
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return null;
+ }
+
@Override
public void close() {
REGISTRY_DATA.clear();
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPer [...]
index 25b5f9d7a28..9c2fcb3e99c 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java
@@ -22,6 +22,7 @@ import org.apache.shardingsphere.mode.persist.PersistRepository;
import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEventListener;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
/**
* Cluster persist repository.
@@ -60,27 +61,47 @@ public interface ClusterPersistRepository extends PersistRepository {
*/
void watch(String key, DataChangedEventListener listener);
+ /**
+ * Watch session connection.
+ *
+ * @param instanceContext instance context
+ */
+ void watchSessionConnection(InstanceContext instanceContext);
+
+ /**
+ * Get global lock.
+ *
+ * @param lockName lock name
+ * @return internal lock
+ */
+ Lock getGlobalLock(String lockName);
+
+ /**
+ * Get standard lock.
+ *
+ * @param lockName lock name
+ * @return internal lock
+ */
+ Lock getStandardLock(String lockName);
+
/**
* Try to get lock under the lock key.
*
* @param key lock key
* @param time time to wait
- * @param unit time unit
+ * @param timeUnit time unit
* @return true if get the lock, false if not
+ * @deprecated TODO remove after refactoring lock is complete
*/
- boolean tryLock(String key, long time, TimeUnit unit);
+ @Deprecated
+ boolean tryLock(String key, long time, TimeUnit timeUnit);
/**
* Release lock.
*
* @param key lock key
+ * @deprecated TODO remove after refactoring lock is complete
*/
+ @Deprecated
void releaseLock(String key);
-
- /**
- * Watch session connection.
- *
- * @param instanceContext instance context
- */
- void watchSessionConnection(InstanceContext instanceContext);
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-etcd/src/main/java/org/apache/shardingsphere/mode/repository/cluster/etcd/EtcdRepository.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mo [...]
index 91d89300c2d..406571735bc 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-etcd/src/main/java/org/apache/shardingsphere/mode/repository/cluster/etcd/EtcdRepository.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-etcd/src/main/java/org/apache/shardingsphere/mode/repository/cluster/etcd/EtcdRepository.java
@@ -35,6 +35,7 @@ import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.infra.instance.InstanceContext;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import org.apache.shardingsphere.mode.repository.cluster.etcd.lock.EtcdInternalLockHolder;
import org.apache.shardingsphere.mode.repository.cluster.etcd.props.EtcdProperties;
import org.apache.shardingsphere.mode.repository.cluster.etcd.props.EtcdPropertyKey;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
@@ -47,6 +48,7 @@ import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;
/**
@@ -63,6 +65,8 @@ public final class EtcdRepository implements ClusterPersistRepository {
private EtcdProperties etcdProperties;
+ private EtcdInternalLockHolder etcdInternalLockHolder;
+
@Override
public void init(final ClusterPersistRepositoryConfiguration config) {
etcdProperties = new EtcdProperties(props);
@@ -70,6 +74,7 @@ public final class EtcdRepository implements ClusterPersistRepository {
.namespace(ByteSequence.from(config.getNamespace(), StandardCharsets.UTF_8))
.maxInboundMessageSize((int) 32e9)
.build();
+ etcdInternalLockHolder = new EtcdInternalLockHolder(client, etcdProperties);
}
@SneakyThrows({InterruptedException.class, ExecutionException.class})
@@ -175,6 +180,17 @@ public final class EtcdRepository implements ClusterPersistRepository {
@Override
public void watchSessionConnection(final InstanceContext instanceContext) {
+ // TODO
+ }
+
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return etcdInternalLockHolder.getGlobalLock(lockName);
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return etcdInternalLockHolder.getStandardLock(lockName);
}
@Override
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-etcd/src/main/java/org/apache/shardingsphere/mode/repository/cluster/etcd/lock/EtcdInternalLockHolder.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphe [...]
new file mode 100644
index 00000000000..d7742869d2b
--- /dev/null
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-etcd/src/main/java/org/apache/shardingsphere/mode/repository/cluster/etcd/lock/EtcdInternalLockHolder.java
@@ -0,0 +1,147 @@
+/*
+ * 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.shardingsphere.mode.repository.cluster.etcd.lock;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.mode.repository.cluster.etcd.props.EtcdProperties;
+import org.apache.shardingsphere.mode.repository.cluster.etcd.props.EtcdPropertyKey;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * Etcd internal lock holder.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class EtcdInternalLockHolder {
+
+ private final Map<String, EtcdInternalLock> locks = new ConcurrentHashMap<>();
+
+ private final Client client;
+
+ private final EtcdProperties etcdProperties;
+
+ /**
+ * Get global lock.
+ *
+ * @param lockName lock name
+ * @return global lock
+ */
+ public Lock getGlobalLock(final String lockName) {
+ return getStandardLock(lockName);
+ }
+
+ /**
+ * Get standard lock.
+ *
+ * @param lockName lock name
+ * @return standard lock
+ */
+ public Lock getStandardLock(final String lockName) {
+ EtcdInternalLock lock = locks.get(lockName);
+ if (Objects.isNull(lock)) {
+ lock = createLock(lockName);
+ locks.put(lockName, lock);
+ }
+ return lock;
+ }
+
+ private EtcdInternalLock createLock(final String lockName) {
+ try {
+ long leaseId = client.getLeaseClient().grant(etcdProperties.getValue(EtcdPropertyKey.TIME_TO_LIVE_SECONDS)).get().getID();
+ return new EtcdInternalLock(client.getLockClient(), lockName, leaseId);
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ log.error("EtcdRepository tryLock error, lockName:{}", lockName, ex);
+ }
+ return null;
+ }
+
+ /**
+ * Etcd internal lock.
+ */
+ @RequiredArgsConstructor
+ private static class EtcdInternalLock implements Lock {
+
+ private final io.etcd.jetcd.Lock lock;
+
+ private final String lockName;
+
+ private final long leaseId;
+
+ @Override
+ public void lock() {
+ try {
+ this.lock.lock(ByteSequence.from(lockName, StandardCharsets.UTF_8), leaseId).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ log.error("EtcdRepository tryLock error, lockName:{}", lockName, ex);
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean tryLock(final long time, final TimeUnit timeUnit) {
+ try {
+ this.lock.lock(ByteSequence.from(lockName, StandardCharsets.UTF_8), leaseId).get(time, timeUnit);
+ return true;
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ log.error("EtcdRepository tryLock error, lockName:{}", lockName, ex);
+ return false;
+ }
+ }
+
+ @Override
+ public void unlock() {
+ try {
+ lock.unlock(ByteSequence.from(lockName, StandardCharsets.UTF_8)).get();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ log.error("EtcdRepository unlock error, lockName:{}", lockName, ex);
+ }
+ }
+
+ @Override
+ public void lockInterruptibly() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Condition newCondition() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-prov [...]
index 4130197cb3e..1372c41f5b4 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java
@@ -27,8 +27,6 @@ import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
-import org.apache.curator.framework.recipes.locks.InterProcessLock;
-import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.shardingsphere.infra.instance.InstanceContext;
@@ -39,6 +37,7 @@ import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEve
import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEventListener;
import org.apache.shardingsphere.mode.repository.cluster.zookeeper.handler.CuratorZookeeperExceptionHandler;
import org.apache.shardingsphere.mode.repository.cluster.zookeeper.listener.SessionConnectionListener;
+import org.apache.shardingsphere.mode.repository.cluster.zookeeper.lock.ZookeeperInternalLockHolder;
import org.apache.shardingsphere.mode.repository.cluster.zookeeper.props.ZookeeperProperties;
import org.apache.shardingsphere.mode.repository.cluster.zookeeper.props.ZookeeperPropertyKey;
import org.apache.zookeeper.CreateMode;
@@ -52,10 +51,9 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
/**
* Registry repository of ZooKeeper.
@@ -68,7 +66,7 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
private final Builder builder = CuratorFrameworkFactory.builder();
- private final Map<String, InterProcessLock> locks = new ConcurrentHashMap<>();
+ private ZookeeperInternalLockHolder internalLockHolder;
@Getter
@Setter
@@ -78,6 +76,7 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
public void init(final ClusterPersistRepositoryConfiguration config) {
ZookeeperProperties zookeeperProps = new ZookeeperProperties(props);
client = buildCuratorClient(config, zookeeperProps);
+ internalLockHolder = new ZookeeperInternalLockHolder(client);
initCuratorClient(zookeeperProps);
}
@@ -275,10 +274,20 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
}
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return internalLockHolder.getGlobalLock(lockName);
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return internalLockHolder.getStandardLock(lockName);
+ }
+
@Override
public boolean tryLock(final String key, final long time, final TimeUnit unit) {
try {
- return getLock(key).acquire(time, unit);
+ return internalLockHolder.getGlobalLock(key).tryLock(time, unit);
// CHECKSTYLE:OFF
} catch (final Exception ex) {
// CHECKSTYLE:ON
@@ -290,9 +299,7 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
@Override
public void releaseLock(final String key) {
try {
- if (availableLock(key)) {
- locks.get(key).release();
- }
+ internalLockHolder.getGlobalLock(key).unlock();
// CHECKSTYLE:OFF
} catch (final Exception ex) {
// CHECKSTYLE:ON
@@ -300,19 +307,6 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
}
}
- private InterProcessLock getLock(final String key) {
- if (availableLock(key)) {
- return locks.get(key);
- }
- InterProcessLock lock = new InterProcessSemaphoreMutex(client, key);
- locks.put(key, lock);
- return lock;
- }
-
- private boolean availableLock(final String key) {
- return Objects.nonNull(locks.get(key));
- }
-
@Override
public void close() {
caches.values().forEach(CuratorCache::close);
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/lock/ZookeeperInternalLockHolder.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repositor [...]
new file mode 100644
index 00000000000..e070294f394
--- /dev/null
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/lock/ZookeeperInternalLockHolder.java
@@ -0,0 +1,131 @@
+/*
+ * 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.shardingsphere.mode.repository.cluster.zookeeper.lock;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.locks.InterProcessLock;
+import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
+import org.apache.shardingsphere.mode.repository.cluster.zookeeper.handler.CuratorZookeeperExceptionHandler;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * Zookeeper internal lock holder.
+ */
+@RequiredArgsConstructor
+public class ZookeeperInternalLockHolder {
+
+ private final Map<String, ZookeeperInternalLock> locks = new ConcurrentHashMap<>();
+
+ private final CuratorFramework client;
+
+ /**
+ * Get global lock.
+ *
+ * @param lockName lock name
+ * @return global lock
+ */
+ public Lock getGlobalLock(final String lockName) {
+ ZookeeperInternalLock lock = locks.get(lockName);
+ if (Objects.isNull(lock)) {
+ lock = new ZookeeperInternalLock(new InterProcessSemaphoreMutex(client, lockName));
+ locks.put(lockName, lock);
+ }
+ return lock;
+ }
+
+ /**
+ * Get standard lock.
+ *
+ * @param lockName lock name
+ * @return standard lock
+ */
+ public Lock getStandardLock(final String lockName) {
+ ZookeeperInternalLock lock = locks.get(lockName);
+ if (Objects.isNull(lock)) {
+ lock = new ZookeeperInternalLock(new InterProcessMutex(client, lockName));
+ locks.put(lockName, lock);
+ }
+ return lock;
+ }
+
+ /**
+ * Zookeeper internal lock.
+ */
+ @RequiredArgsConstructor
+ public static class ZookeeperInternalLock implements Lock {
+
+ private final InterProcessLock lock;
+
+ @Override
+ public void lock() {
+ try {
+ lock.acquire();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ CuratorZookeeperExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean tryLock(final long time, final TimeUnit timeUnit) {
+ try {
+ return lock.acquire(time, timeUnit);
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ CuratorZookeeperExceptionHandler.handleException(ex);
+ }
+ return false;
+ }
+
+ @Override
+ public void unlock() {
+ try {
+ lock.release();
+ // CHECKSTYLE:OFF
+ } catch (final Exception ex) {
+ // CHECKSTYLE:ON
+ CuratorZookeeperExceptionHandler.handleException(ex);
+ }
+ }
+
+ @Override
+ public void lockInterruptibly() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Condition newCondition() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository- [...]
index 019c9fb6b34..bcecd242cd0 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java
@@ -38,6 +38,7 @@ import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEvent;
import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEvent.Type;
+import org.apache.shardingsphere.mode.repository.cluster.zookeeper.lock.ZookeeperInternalLockHolder;
import org.apache.shardingsphere.mode.repository.cluster.zookeeper.props.ZookeeperPropertyKey;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
@@ -120,10 +121,10 @@ public final class CuratorZookeeperRepositoryTest {
@SneakyThrows
public void init() {
mockClient();
- mockField();
mockBuilder();
ClusterPersistRepositoryConfiguration config = new ClusterPersistRepositoryConfiguration(REPOSITORY.getType(), "governance", SERVER_LISTS, new Properties());
REPOSITORY.init(config);
+ mockInternalLockHolder();
}
@SneakyThrows
@@ -143,10 +144,14 @@ public final class CuratorZookeeperRepositoryTest {
}
@SneakyThrows
- private void mockField() {
- Field locksFiled = CuratorZookeeperRepository.class.getDeclaredField("locks");
+ private void mockInternalLockHolder() {
+ Field internalLockHolderFiled = CuratorZookeeperRepository.class.getDeclaredField("internalLockHolder");
+ internalLockHolderFiled.setAccessible(true);
+ ZookeeperInternalLockHolder holder = new ZookeeperInternalLockHolder(client);
+ Field locksFiled = ZookeeperInternalLockHolder.class.getDeclaredField("locks");
locksFiled.setAccessible(true);
- locksFiled.set(REPOSITORY, Collections.singletonMap("/locks/glock", interProcessLock));
+ locksFiled.set(holder, Collections.singletonMap("/locks/glock", new ZookeeperInternalLockHolder.ZookeeperInternalLock(interProcessLock)));
+ internalLockHolderFiled.set(REPOSITORY, holder);
}
private void mockBuilder() {
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/MemoryLockContext.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/MemoryLockContext.java
index d3c009444b2..7ab2d38b841 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/MemoryLockContext.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/MemoryLockContext.java
@@ -54,7 +54,7 @@ public final class MemoryLockContext implements LockContext {
public boolean isLockedDatabase(final String databaseName) {
Preconditions.checkNotNull(databaseName, "Is locked database args database name can not be null.");
ShardingSphereLock shardingSphereLock = locks.get(databaseName);
- return null != shardingSphereLock && shardingSphereLock.isLocked(databaseName);
+ return null != shardingSphereLock && shardingSphereLock.isLocked();
}
@Override
@@ -75,6 +75,11 @@ public final class MemoryLockContext implements LockContext {
}
}
+ @Override
+ public ShardingSphereLock getOrCreateStandardLock(final String lockName) {
+ return getOrCreateGlobalLock(lockName);
+ }
+
@Override
public ShardingSphereLock getGlobalLock(final String lockName) {
Preconditions.checkNotNull(lockName, "Get global lock args lock name can not be null.");
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/ShardingSphereNonReentrantLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/ShardingSphereNonReentrantLock.java
index ee865e11e46..45c7fb77a51 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/ShardingSphereNonReentrantLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-memory-mode/shardingsphere-memory-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/memory/lock/ShardingSphereNonReentrantLock.java
@@ -69,7 +69,7 @@ public final class ShardingSphereNonReentrantLock implements ShardingSphereLock
}
@Override
- public boolean isLocked(final String lockName) {
+ public boolean isLocked() {
return locked;
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/ShardingSphereNonReentrantLock.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/ShardingSphereNonReentrantLock.java
index b7f919891d7..8755befdf39 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/ShardingSphereNonReentrantLock.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/ShardingSphereNonReentrantLock.java
@@ -69,7 +69,7 @@ public final class ShardingSphereNonReentrantLock implements ShardingSphereLock
}
@Override
- public boolean isLocked(final String lockName) {
+ public boolean isLocked() {
return locked;
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/StandaloneLockContext.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/StandaloneLockContext.java
index ffd0ff20194..37f0c2b6d37 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/StandaloneLockContext.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-standalone-mode/shardingsphere-standalone-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/lock/StandaloneLockContext.java
@@ -54,7 +54,7 @@ public final class StandaloneLockContext implements LockContext {
public boolean isLockedDatabase(final String databaseName) {
Preconditions.checkNotNull(databaseName, "Is locked database args database name can not be null.");
ShardingSphereLock shardingSphereLock = locks.get(databaseName);
- return null != shardingSphereLock && shardingSphereLock.isLocked(databaseName);
+ return null != shardingSphereLock && shardingSphereLock.isLocked();
}
@Override
@@ -75,6 +75,11 @@ public final class StandaloneLockContext implements LockContext {
}
}
+ @Override
+ public ShardingSphereLock getOrCreateStandardLock(final String lockName) {
+ return getOrCreateGlobalLock(lockName);
+ }
+
@Override
public ShardingSphereLock getGlobalLock(final String lockName) {
Preconditions.checkNotNull(lockName, "Get global lock args lock name can not be null.");
diff --git a/shardingsphere-proxy/shardingsphere-proxy-bootstrap/src/test/java/org/apache/shardingsphere/proxy/fixture/FixtureClusterPersistRepository.java b/shardingsphere-proxy/shardingsphere-proxy-bootstrap/src/test/java/org/apache/shardingsphere/proxy/fixture/FixtureClusterPersistRepository.java
index ef6cb5de839..89c891359b7 100644
--- a/shardingsphere-proxy/shardingsphere-proxy-bootstrap/src/test/java/org/apache/shardingsphere/proxy/fixture/FixtureClusterPersistRepository.java
+++ b/shardingsphere-proxy/shardingsphere-proxy-bootstrap/src/test/java/org/apache/shardingsphere/proxy/fixture/FixtureClusterPersistRepository.java
@@ -27,6 +27,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
public final class FixtureClusterPersistRepository implements ClusterPersistRepository {
@@ -82,6 +83,16 @@ public final class FixtureClusterPersistRepository implements ClusterPersistRepo
public void watchSessionConnection(final InstanceContext instanceContext) {
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return null;
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return null;
+ }
+
@Override
public void close() {
REGISTRY_DATA.clear();
diff --git a/shardingsphere-test/shardingsphere-integration-driver-test/src/test/java/org/apache/shardingsphere/driver/fixture/TestClusterPersistRepository.java b/shardingsphere-test/shardingsphere-integration-driver-test/src/test/java/org/apache/shardingsphere/driver/fixture/TestClusterPersistRepository.java
index e1d213791a8..67d63c618ed 100644
--- a/shardingsphere-test/shardingsphere-integration-driver-test/src/test/java/org/apache/shardingsphere/driver/fixture/TestClusterPersistRepository.java
+++ b/shardingsphere-test/shardingsphere-integration-driver-test/src/test/java/org/apache/shardingsphere/driver/fixture/TestClusterPersistRepository.java
@@ -28,6 +28,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
public final class TestClusterPersistRepository implements ClusterPersistRepository {
@@ -84,6 +85,16 @@ public final class TestClusterPersistRepository implements ClusterPersistReposit
public void watchSessionConnection(final InstanceContext instanceContext) {
}
+ @Override
+ public Lock getGlobalLock(final String lockName) {
+ return null;
+ }
+
+ @Override
+ public Lock getStandardLock(final String lockName) {
+ return null;
+ }
+
@Override
public void close() {
registryData.clear();
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLockTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLockTest.java
index 05c73887470..a81e729f3a3 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLockTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/lock/PipelineSimpleLockTest.java
@@ -23,8 +23,6 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@RunWith(MockitoJUnitRunner.class)
@@ -42,7 +40,6 @@ public final class PipelineSimpleLockTest {
long timeoutMillis = 50L;
boolean locked = pipelineSimpleLock.tryLock(lockName, timeoutMillis);
assertTrue(locked);
- assertThat(pipelineSimpleLock.tryLock(lockName, timeoutMillis), is(false));
pipelineSimpleLock.releaseLock(lockName);
locked = pipelineSimpleLock.tryLock(lockName, timeoutMillis);
assertTrue(locked);