You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by me...@apache.org on 2026/03/24 15:00:30 UTC
(shardingsphere) branch master updated: Refactor DistributedLockHolder to remove the lock cache (#38544)
This is an automated email from the ASF dual-hosted git repository.
menghaoran 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 348ce3bf9bd Refactor DistributedLockHolder to remove the lock cache (#38544)
348ce3bf9bd is described below
commit 348ce3bf9bdc4b2600a043ac7306206b69413bbe
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Tue Mar 24 23:00:23 2026 +0800
Refactor DistributedLockHolder to remove the lock cache (#38544)
* Refactor DistributedLockHolder to remove the lock cache
* Fix spotless
---
...eratorContext.java => ExclusiveLockHandle.java} | 21 +--
.../mode/exclusive/ExclusiveOperatorContext.java | 13 +-
.../mode/exclusive/ExclusiveOperatorEngine.java | 8 +-
.../exclusive/ExclusiveOperatorEngineTest.java | 51 ++++---
.../exclusive/ClusterExclusiveOperatorContext.java | 51 ++++++-
.../ClusterExclusiveOperatorContextTest.java | 169 ++++++++++++++++++---
.../cluster/core/lock/DistributedLockHolder.java | 8 -
.../core/lock/DistributedLockHolderTest.java | 135 ++++++++++++----
.../StandaloneExclusiveOperatorContext.java | 13 +-
.../StandaloneExclusiveOperatorContextTest.java | 37 +++--
10 files changed, 368 insertions(+), 138 deletions(-)
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveLockHandle.java
similarity index 65%
copy from mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java
copy to mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveLockHandle.java
index 670f25dea76..4474252d12f 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveLockHandle.java
@@ -18,23 +18,10 @@
package org.apache.shardingsphere.mode.exclusive;
/**
- * Exclusive operator context.
+ * Exclusive lock handle.
*/
-public interface ExclusiveOperatorContext {
+public interface ExclusiveLockHandle extends AutoCloseable {
- /**
- * Start exclusive operation.
- *
- * @param operationKey operation key
- * @param timeoutMillis timeout milliseconds
- * @return is started or not
- */
- boolean start(String operationKey, long timeoutMillis);
-
- /**
- * Stop exclusive operation.
- *
- * @param operationKey operation key
- */
- void stop(String operationKey);
+ @Override
+ void close();
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java
index 670f25dea76..6414f203c0c 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorContext.java
@@ -17,6 +17,8 @@
package org.apache.shardingsphere.mode.exclusive;
+import java.util.Optional;
+
/**
* Exclusive operator context.
*/
@@ -27,14 +29,7 @@ public interface ExclusiveOperatorContext {
*
* @param operationKey operation key
* @param timeoutMillis timeout milliseconds
- * @return is started or not
- */
- boolean start(String operationKey, long timeoutMillis);
-
- /**
- * Stop exclusive operation.
- *
- * @param operationKey operation key
+ * @return exclusive lock handle
*/
- void stop(String operationKey);
+ Optional<ExclusiveLockHandle> start(String operationKey, long timeoutMillis);
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngine.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngine.java
index decc05bbb2c..3745d84d015 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngine.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngine.java
@@ -24,6 +24,7 @@ import org.apache.shardingsphere.mode.node.path.engine.generator.NodePathGenerat
import org.apache.shardingsphere.mode.node.path.type.exclusive.ExclusiveOperationNodePath;
import java.sql.SQLException;
+import java.util.Optional;
/**
* Exclusive operator engine.
@@ -60,11 +61,10 @@ public final class ExclusiveOperatorEngine {
*/
public <T> T operateWithResult(final ExclusiveOperation operation, final long timeoutMillis, final ExclusiveOperationCallback<T> callback) throws SQLException {
String operationKey = NodePathGenerator.toPath(new ExclusiveOperationNodePath(operation.getName()));
- if (context.start(operationKey, timeoutMillis)) {
- try {
+ Optional<ExclusiveLockHandle> lockHandle = context.start(operationKey, timeoutMillis);
+ if (lockHandle.isPresent()) {
+ try (ExclusiveLockHandle ignored = lockHandle.get()) {
return callback.execute();
- } finally {
- context.stop(operationKey);
}
}
return null;
diff --git a/mode/core/src/test/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngineTest.java b/mode/core/src/test/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngineTest.java
index ad6157f8727..2d1979db1cd 100644
--- a/mode/core/src/test/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngineTest.java
+++ b/mode/core/src/test/java/org/apache/shardingsphere/mode/exclusive/ExclusiveOperatorEngineTest.java
@@ -20,53 +20,68 @@ package org.apache.shardingsphere.mode.exclusive;
import org.apache.shardingsphere.mode.exclusive.callback.ExclusiveOperationCallback;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
import java.sql.SQLException;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
-@ExtendWith(MockitoExtension.class)
class ExclusiveOperatorEngineTest {
- @Mock
- private ExclusiveOperatorContext context;
+ private StubExclusiveOperatorContext context;
private ExclusiveOperatorEngine engine;
@BeforeEach
void setUp() {
+ context = new StubExclusiveOperatorContext();
engine = new ExclusiveOperatorEngine(context);
}
@Test
void assertOperate() throws SQLException {
long timeoutMillis = 100L;
- when(context.start("/exclusive_operation/foo_operation", timeoutMillis)).thenReturn(true);
AtomicBoolean callbackExecuted = new AtomicBoolean(false);
+ context.startResult = Optional.of(() -> context.closed.set(true));
engine.operate(() -> "foo_operation", timeoutMillis, () -> callbackExecuted.set(true));
assertTrue(callbackExecuted.get());
- verify(context).start("/exclusive_operation/foo_operation", timeoutMillis);
- verify(context).stop("/exclusive_operation/foo_operation");
+ assertThat(context.actualOperationKey, is("/exclusive_operation/foo_operation"));
+ assertThat(context.actualTimeoutMillis, is(timeoutMillis));
+ assertTrue(context.closed.get());
}
@Test
void assertOperateWithResult() throws SQLException {
@SuppressWarnings("unchecked")
- ExclusiveOperationCallback<String> callback = mock(ExclusiveOperationCallback.class);
+ ExclusiveOperationCallback<String> callback = () -> "ignored";
long timeoutMillis = 100L;
+ context.startResult = Optional.empty();
assertNull(engine.operateWithResult(() -> "foo_operation", timeoutMillis, callback));
- verify(context).start("/exclusive_operation/foo_operation", timeoutMillis);
- verify(context, never()).stop(anyString());
- verify(callback, never()).execute();
+ assertThat(context.actualOperationKey, is("/exclusive_operation/foo_operation"));
+ assertThat(context.actualTimeoutMillis, is(timeoutMillis));
+ assertFalse(context.closed.get());
+ }
+
+ private static final class StubExclusiveOperatorContext implements ExclusiveOperatorContext {
+
+ private String actualOperationKey;
+
+ private long actualTimeoutMillis;
+
+ private Optional<ExclusiveLockHandle> startResult = Optional.empty();
+
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ @Override
+ public Optional<ExclusiveLockHandle> start(final String operationKey, final long timeoutMillis) {
+ actualOperationKey = operationKey;
+ actualTimeoutMillis = timeoutMillis;
+ return startResult;
+ }
}
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
index d2e86d07bf4..ef1fe710c4e 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
@@ -18,9 +18,16 @@
package org.apache.shardingsphere.mode.manager.cluster.exclusive;
import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.apache.shardingsphere.mode.exclusive.ExclusiveOperatorContext;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import org.apache.shardingsphere.mode.repository.cluster.core.lock.DistributedLockHolder;
+import org.apache.shardingsphere.mode.repository.cluster.lock.DistributedLock;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Cluster exclusive operator context.
@@ -30,13 +37,47 @@ public final class ClusterExclusiveOperatorContext implements ExclusiveOperatorC
private final ClusterPersistRepository repository;
+ private final Collection<String> exclusiveOperationKeys = new CopyOnWriteArraySet<>();
+
@Override
- public boolean start(final String operationKey, final long timeoutMillis) {
- return DistributedLockHolder.getDistributedLock(operationKey, repository).tryLock(timeoutMillis);
+ public Optional<ExclusiveLockHandle> start(final String operationKey, final long timeoutMillis) {
+ if (!exclusiveOperationKeys.add(operationKey)) {
+ return Optional.empty();
+ }
+ DistributedLock distributedLock = DistributedLockHolder.getDistributedLock(operationKey, repository);
+ if (!distributedLock.tryLock(timeoutMillis)) {
+ exclusiveOperationKeys.remove(operationKey);
+ return Optional.empty();
+ }
+ return Optional.of(new ClusterExclusiveLockHandle(operationKey, distributedLock, exclusiveOperationKeys));
}
- @Override
- public void stop(final String operationKey) {
- DistributedLockHolder.getDistributedLock(operationKey, repository).unlock();
+ private static final class ClusterExclusiveLockHandle implements ExclusiveLockHandle {
+
+ private final String operationKey;
+
+ private final DistributedLock distributedLock;
+
+ private final Collection<String> exclusiveOperationKeys;
+
+ private final AtomicBoolean closed = new AtomicBoolean();
+
+ private ClusterExclusiveLockHandle(final String operationKey, final DistributedLock distributedLock, final Collection<String> exclusiveOperationKeys) {
+ this.operationKey = operationKey;
+ this.distributedLock = distributedLock;
+ this.exclusiveOperationKeys = exclusiveOperationKeys;
+ }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ distributedLock.unlock();
+ } finally {
+ exclusiveOperationKeys.remove(operationKey);
+ }
+ }
}
}
diff --git a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
index f375e6d3ce8..e879ee5a7b7 100644
--- a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
+++ b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
@@ -17,41 +17,166 @@
package org.apache.shardingsphere.mode.manager.cluster.exclusive;
+import org.apache.shardingsphere.infra.instance.ComputeNodeInstanceContext;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
-import org.apache.shardingsphere.mode.repository.cluster.core.lock.DistributedLockHolder;
+import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
+import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEventListener;
import org.apache.shardingsphere.mode.repository.cluster.lock.DistributedLock;
-import org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
-import org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-@ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(DistributedLockHolder.class)
class ClusterExclusiveOperatorContextTest {
- @Mock
- private ClusterPersistRepository repository;
+ @Test
+ void assertStartReturnsHandle() {
+ StubDistributedLock distributedLock = new StubDistributedLock(true);
+ assertTrue(new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock)).start("op", 50L).isPresent());
+ assertThat(distributedLock.tryLockCallCount, is(1));
+ }
+
+ @Test
+ void assertStartReturnsEmptyWhenKeyExists() {
+ StubDistributedLock distributedLock = new StubDistributedLock(true);
+ ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock));
+ ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ assertFalse(context.start("op", 50L).isPresent());
+ assertThat(distributedLock.tryLockCallCount, is(1));
+ lockHandle.close();
+ }
- @Mock
- private DistributedLock distributedLock;
+ @Test
+ void assertCloseUnlock() {
+ StubDistributedLock distributedLock = new StubDistributedLock(true);
+ new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock)).start("op", 50L).orElseThrow(AssertionError::new).close();
+ assertThat(distributedLock.unlockCallCount, is(1));
+ }
@Test
- void assertStartTryLock() {
- when(DistributedLockHolder.getDistributedLock("op", repository)).thenReturn(distributedLock);
- when(distributedLock.tryLock(50L)).thenReturn(true);
- assertTrue(new ClusterExclusiveOperatorContext(repository).start("op", 50L));
- verify(distributedLock).tryLock(50L);
+ void assertStartReturnsHandleAfterClose() {
+ StubDistributedLock distributedLock = new StubDistributedLock(true, true);
+ ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock));
+ ExclusiveLockHandle firstHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ firstHandle.close();
+ Optional<ExclusiveLockHandle> actual = context.start("op", 50L);
+ assertTrue(actual.isPresent());
+ assertThat(actual.orElseThrow(AssertionError::new), not(firstHandle));
}
@Test
- void assertStopUnlock() {
- when(DistributedLockHolder.getDistributedLock("op", repository)).thenReturn(distributedLock);
- new ClusterExclusiveOperatorContext(repository).stop("op");
- verify(distributedLock).unlock();
+ void assertStartRemovesOperationKeyAfterTryLockFailure() {
+ StubDistributedLock distributedLock = new StubDistributedLock(false, true);
+ ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock));
+ assertFalse(context.start("op", 50L).isPresent());
+ assertTrue(context.start("op", 50L).isPresent());
+ assertThat(distributedLock.tryLockCallCount, is(2));
+ }
+
+ private static final class StubDistributedLock implements DistributedLock {
+
+ private final List<Boolean> tryLockResults;
+
+ private int tryLockCallCount;
+
+ private int unlockCallCount;
+
+ private StubDistributedLock(final boolean... tryLockResults) {
+ this.tryLockResults = new LinkedList<>();
+ for (boolean each : tryLockResults) {
+ this.tryLockResults.add(each);
+ }
+ }
+
+ @Override
+ public boolean tryLock(final long timeoutMillis) {
+ tryLockCallCount++;
+ return tryLockResults.remove(0);
+ }
+
+ @Override
+ public void unlock() {
+ unlockCallCount++;
+ }
+ }
+
+ private static final class StubClusterPersistRepository implements ClusterPersistRepository {
+
+ private final DistributedLock distributedLock;
+
+ private StubClusterPersistRepository(final DistributedLock distributedLock) {
+ this.distributedLock = distributedLock;
+ }
+
+ @Override
+ public void init(final ClusterPersistRepositoryConfiguration config, final ComputeNodeInstanceContext computeNodeInstanceContext) {
+ }
+
+ @Override
+ public String query(final String key) {
+ return null;
+ }
+
+ @Override
+ public List<String> getChildrenKeys(final String key) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isExisted(final String key) {
+ return false;
+ }
+
+ @Override
+ public void persist(final String key, final String value) {
+ }
+
+ @Override
+ public void update(final String key, final String value) {
+ }
+
+ @Override
+ public void persistEphemeral(final String key, final String value) {
+ }
+
+ @Override
+ public boolean persistExclusiveEphemeral(final String key, final String value) {
+ return false;
+ }
+
+ @Override
+ public Optional<DistributedLock> getDistributedLock(final String lockKey) {
+ return Optional.of(distributedLock);
+ }
+
+ @Override
+ public void delete(final String key) {
+ }
+
+ @Override
+ public void watch(final String key, final DataChangedEventListener listener) {
+ }
+
+ @Override
+ public void removeDataListener(final String key) {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String getType() {
+ return "STUB";
+ }
}
}
diff --git a/mode/type/cluster/repository/core/src/main/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolder.java b/mode/type/cluster/repository/core/src/main/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolder.java
index 4115a6aeeca..46daa3d8b65 100644
--- a/mode/type/cluster/repository/core/src/main/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolder.java
+++ b/mode/type/cluster/repository/core/src/main/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolder.java
@@ -24,9 +24,7 @@ import org.apache.shardingsphere.mode.repository.cluster.core.lock.type.DefaultD
import org.apache.shardingsphere.mode.repository.cluster.core.lock.type.props.DefaultLockTypedProperties;
import org.apache.shardingsphere.mode.repository.cluster.lock.DistributedLock;
-import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
/**
* Distributed lock holder.
@@ -34,8 +32,6 @@ import java.util.concurrent.ConcurrentHashMap;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class DistributedLockHolder {
- private static final Map<String, DistributedLock> LOCKS = new ConcurrentHashMap<>();
-
/**
* Get distributed lock.
*
@@ -44,10 +40,6 @@ public final class DistributedLockHolder {
* @return got distributed lock
*/
public static DistributedLock getDistributedLock(final String lockKey, final ClusterPersistRepository repository) {
- return LOCKS.computeIfAbsent(lockKey, key -> loadDistributedLock(lockKey, repository));
- }
-
- private static DistributedLock loadDistributedLock(final String lockKey, final ClusterPersistRepository repository) {
return repository.getDistributedLock(lockKey).orElseGet(() -> new DefaultDistributedLock(lockKey, repository, new DefaultLockTypedProperties(new Properties())));
}
}
diff --git a/mode/type/cluster/repository/core/src/test/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolderTest.java b/mode/type/cluster/repository/core/src/test/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolderTest.java
index 193211303b6..527109694f4 100644
--- a/mode/type/cluster/repository/core/src/test/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolderTest.java
+++ b/mode/type/cluster/repository/core/src/test/java/org/apache/shardingsphere/mode/repository/cluster/core/lock/DistributedLockHolderTest.java
@@ -17,52 +17,131 @@
package org.apache.shardingsphere.mode.repository.cluster.core.lock;
-import lombok.SneakyThrows;
+import org.apache.shardingsphere.infra.instance.ComputeNodeInstanceContext;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
import org.apache.shardingsphere.mode.repository.cluster.core.lock.type.DefaultDistributedLock;
+import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEventListener;
import org.apache.shardingsphere.mode.repository.cluster.lock.DistributedLock;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.internal.configuration.plugins.Plugins;
-import org.mockito.junit.jupiter.MockitoExtension;
-import java.util.Map;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
import java.util.Optional;
-import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.isA;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.hamcrest.Matchers.not;
-@ExtendWith(MockitoExtension.class)
class DistributedLockHolderTest {
- @Mock
- private ClusterPersistRepository repository;
-
- @BeforeEach
- @AfterEach
- @SuppressWarnings("unchecked")
- @SneakyThrows(ReflectiveOperationException.class)
- void clearLocks() {
- ((Map<String, DistributedLock>) Plugins.getMemberAccessor().get(DistributedLockHolder.class.getDeclaredField("LOCKS"), null)).clear();
- }
-
@Test
void assertGetDistributedLockReturnsRepositoryLock() {
- DistributedLock distributedLock = mock(DistributedLock.class);
- when(repository.getDistributedLock("lock-key")).thenReturn(Optional.of(distributedLock));
- assertThat(DistributedLockHolder.getDistributedLock("lock-key", repository), is(distributedLock));
+ DistributedLock firstLock = new StubDistributedLock();
+ DistributedLock secondLock = new StubDistributedLock();
+ StubClusterPersistRepository repository = new StubClusterPersistRepository(Optional.of(firstLock), Optional.of(secondLock));
+ assertThat(DistributedLockHolder.getDistributedLock("lock-key", repository), is(firstLock));
+ assertThat(DistributedLockHolder.getDistributedLock("lock-key", repository), is(secondLock));
}
@Test
- void assertGetDistributedLockCreatesDefaultLock() {
- when(repository.getDistributedLock("lock-key")).thenReturn(Optional.empty());
+ void assertGetDistributedLockCreatesDefaultLockEachTime() {
+ StubClusterPersistRepository repository = new StubClusterPersistRepository(Optional.empty(), Optional.empty());
DistributedLock actual = DistributedLockHolder.getDistributedLock("lock-key", repository);
+ DistributedLock another = DistributedLockHolder.getDistributedLock("lock-key", repository);
assertThat(actual, isA(DefaultDistributedLock.class));
+ assertThat(another, isA(DefaultDistributedLock.class));
+ assertThat(another, not(actual));
+ }
+
+ private static final class StubDistributedLock implements DistributedLock {
+
+ @Override
+ public boolean tryLock(final long timeoutMillis) {
+ return true;
+ }
+
+ @Override
+ public void unlock() {
+ }
+ }
+
+ private static final class StubClusterPersistRepository implements ClusterPersistRepository {
+
+ private final List<Optional<DistributedLock>> distributedLocks;
+
+ private int index;
+
+ private StubClusterPersistRepository(final Optional<DistributedLock> firstLock, final Optional<DistributedLock> secondLock) {
+ distributedLocks = Arrays.asList(firstLock, secondLock);
+ }
+
+ @Override
+ public void init(final ClusterPersistRepositoryConfiguration config, final ComputeNodeInstanceContext computeNodeInstanceContext) {
+ }
+
+ @Override
+ public String query(final String key) {
+ return null;
+ }
+
+ @Override
+ public List<String> getChildrenKeys(final String key) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isExisted(final String key) {
+ return false;
+ }
+
+ @Override
+ public void persist(final String key, final String value) {
+ }
+
+ @Override
+ public void update(final String key, final String value) {
+ }
+
+ @Override
+ public void persistEphemeral(final String key, final String value) {
+ }
+
+ @Override
+ public boolean persistExclusiveEphemeral(final String key, final String value) {
+ return false;
+ }
+
+ @Override
+ public Optional<DistributedLock> getDistributedLock(final String lockKey) {
+ Optional<DistributedLock> result = distributedLocks.get(index);
+ if (index < distributedLocks.size() - 1) {
+ index++;
+ }
+ return result;
+ }
+
+ @Override
+ public void delete(final String key) {
+ }
+
+ @Override
+ public void watch(final String key, final DataChangedEventListener listener) {
+ }
+
+ @Override
+ public void removeDataListener(final String key) {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String getType() {
+ return "STUB";
+ }
}
}
diff --git a/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java b/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
index 680f3aa685e..c82770cc023 100644
--- a/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
+++ b/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
@@ -17,10 +17,12 @@
package org.apache.shardingsphere.mode.manager.standalone.exclusive;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.apache.shardingsphere.mode.exclusive.ExclusiveOperatorContext;
import org.apache.shardingsphere.mode.retry.RetryExecutor;
import java.util.Collection;
+import java.util.Optional;
import java.util.concurrent.CopyOnWriteArraySet;
/**
@@ -31,12 +33,9 @@ public final class StandaloneExclusiveOperatorContext implements ExclusiveOperat
private final Collection<String> exclusiveOperationKeys = new CopyOnWriteArraySet<>();
@Override
- public boolean start(final String operationKey, final long timeoutMillis) {
- return new RetryExecutor(timeoutMillis, 50L).execute(arg -> exclusiveOperationKeys.add(operationKey), null);
- }
-
- @Override
- public void stop(final String operationKey) {
- exclusiveOperationKeys.remove(operationKey);
+ public Optional<ExclusiveLockHandle> start(final String operationKey, final long timeoutMillis) {
+ return new RetryExecutor(timeoutMillis, 50L).execute(arg -> exclusiveOperationKeys.add(operationKey), null)
+ ? Optional.of(() -> exclusiveOperationKeys.remove(operationKey))
+ : Optional.empty();
}
}
diff --git a/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java b/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
index 7216b885de9..24b51aba163 100644
--- a/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
+++ b/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
@@ -17,42 +17,39 @@
package org.apache.shardingsphere.mode.manager.standalone.exclusive;
-import lombok.SneakyThrows;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.junit.jupiter.api.Test;
-import org.mockito.internal.configuration.plugins.Plugins;
-import java.util.Collection;
+import java.util.Optional;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class StandaloneExclusiveOperatorContextTest {
@Test
- void assertStartReturnsTrueWhenKeyAbsent() {
- assertTrue(new StandaloneExclusiveOperatorContext().start("operation-key", 50L));
+ void assertStartReturnsHandleWhenKeyAbsent() {
+ Optional<ExclusiveLockHandle> actual = new StandaloneExclusiveOperatorContext().start("operation-key", 50L);
+ assertTrue(actual.isPresent());
}
@Test
- void assertStartReturnsFalseWhenKeyExists() {
+ void assertStartReturnsEmptyWhenKeyExists() {
StandaloneExclusiveOperatorContext context = new StandaloneExclusiveOperatorContext();
- Collection<String> existingKeys = getExclusiveOperationKeys(context);
- existingKeys.add("duplicated-key");
- assertFalse(context.start("duplicated-key", 0L));
+ ExclusiveLockHandle firstHandle = context.start("duplicated-key", 0L).orElseThrow(AssertionError::new);
+ assertFalse(context.start("duplicated-key", 0L).isPresent());
+ firstHandle.close();
}
@Test
- void assertStopRemovesOperationKey() {
+ void assertCloseReleasesOperationKey() {
StandaloneExclusiveOperatorContext context = new StandaloneExclusiveOperatorContext();
- Collection<String> existingKeys = getExclusiveOperationKeys(context);
- existingKeys.add("stopped-key");
- context.stop("stopped-key");
- assertTrue(existingKeys.isEmpty());
- }
-
- @SuppressWarnings("unchecked")
- @SneakyThrows(ReflectiveOperationException.class)
- private Collection<String> getExclusiveOperationKeys(final StandaloneExclusiveOperatorContext context) {
- return (Collection<String>) Plugins.getMemberAccessor().get(StandaloneExclusiveOperatorContext.class.getDeclaredField("exclusiveOperationKeys"), context);
+ ExclusiveLockHandle firstHandle = context.start("stopped-key", 0L).orElseThrow(AssertionError::new);
+ firstHandle.close();
+ Optional<ExclusiveLockHandle> actual = context.start("stopped-key", 0L);
+ assertTrue(actual.isPresent());
+ assertThat(actual.orElseThrow(AssertionError::new), not(firstHandle));
}
}