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/04/02 06:02:30 UTC
(shardingsphere) branch master updated: Refactor ClusterExclusiveOperatorContext (#38584)
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 6f8388063d4 Refactor ClusterExclusiveOperatorContext (#38584)
6f8388063d4 is described below
commit 6f8388063d40fb45fd8968fe98a0221ec10e1dd6
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Thu Apr 2 14:02:23 2026 +0800
Refactor ClusterExclusiveOperatorContext (#38584)
* Refactor ClusterExclusiveOperatorContext
* Refactor ClusterExclusiveOperatorContext
---
.../exclusive/ClusterExclusiveOperatorContext.java | 17 +-
.../ClusterExclusiveOperatorContextTest.java | 243 +++++++++++++++++----
2 files changed, 215 insertions(+), 45 deletions(-)
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 ef1fe710c4e..d7d1779448c 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
@@ -20,6 +20,7 @@ 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.retry.RetryExecutor;
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;
@@ -35,23 +36,33 @@ import java.util.concurrent.atomic.AtomicBoolean;
@RequiredArgsConstructor
public final class ClusterExclusiveOperatorContext implements ExclusiveOperatorContext {
+ private static final long RETRY_INTERVAL_MILLIS = 50L;
+
private final ClusterPersistRepository repository;
private final Collection<String> exclusiveOperationKeys = new CopyOnWriteArraySet<>();
@Override
public Optional<ExclusiveLockHandle> start(final String operationKey, final long timeoutMillis) {
- if (!exclusiveOperationKeys.add(operationKey)) {
+ final long startTimeMillis = System.currentTimeMillis();
+ if (!new RetryExecutor(timeoutMillis, RETRY_INTERVAL_MILLIS).execute(arg -> exclusiveOperationKeys.add(operationKey), null)) {
return Optional.empty();
}
- DistributedLock distributedLock = DistributedLockHolder.getDistributedLock(operationKey, repository);
- if (!distributedLock.tryLock(timeoutMillis)) {
+ final DistributedLock distributedLock = DistributedLockHolder.getDistributedLock(operationKey, repository);
+ if (!distributedLock.tryLock(getRemainingTimeoutMillis(startTimeMillis, timeoutMillis))) {
exclusiveOperationKeys.remove(operationKey);
return Optional.empty();
}
return Optional.of(new ClusterExclusiveLockHandle(operationKey, distributedLock, exclusiveOperationKeys));
}
+ private long getRemainingTimeoutMillis(final long startTimeMillis, final long timeoutMillis) {
+ if (timeoutMillis < 0L) {
+ return timeoutMillis;
+ }
+ return Math.max(0L, timeoutMillis - (System.currentTimeMillis() - startTimeMillis));
+ }
+
private static final class ClusterExclusiveLockHandle implements ExclusiveLockHandle {
private final String 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 e879ee5a7b7..3c076051f4b 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
@@ -29,6 +29,15 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
@@ -39,83 +48,230 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
class ClusterExclusiveOperatorContextTest {
@Test
- void assertStartReturnsHandle() {
- StubDistributedLock distributedLock = new StubDistributedLock(true);
- assertTrue(new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock)).start("op", 50L).isPresent());
- assertThat(distributedLock.tryLockCallCount, is(1));
+ void assertStart() {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true));
+ assertTrue(context.start("op", 50L).isPresent());
+ assertThat(sharedState.tryLockCallCount.get(), 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();
+ void assertStartWhenOperationKeyExists() {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true));
+ final ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ try {
+ assertFalse(context.start("op", 1L).isPresent());
+ assertThat(sharedState.tryLockCallCount.get(), is(1));
+ } finally {
+ lockHandle.close();
+ }
+ }
+
+ @Test
+ void assertStartWaitsUntilOperationKeyReleased() throws InterruptedException, ExecutionException, TimeoutException {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true, true));
+ final ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ final ExecutorService executorService = Executors.newSingleThreadExecutor();
+ try {
+ final Future<Optional<ExclusiveLockHandle>> future = executorService.submit(() -> context.start("op", 500L));
+ TimeUnit.MILLISECONDS.sleep(100L);
+ assertFalse(future.isDone());
+ assertThat(sharedState.tryLockCallCount.get(), is(1));
+ lockHandle.close();
+ final Optional<ExclusiveLockHandle> actual = future.get(1L, TimeUnit.SECONDS);
+ assertTrue(actual.isPresent());
+ assertThat(sharedState.tryLockCallCount.get(), is(2));
+ actual.orElseThrow(AssertionError::new).close();
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void assertClose() {
+ final SharedState sharedState = new SharedState();
+ new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true)).start("op", 50L).orElseThrow(AssertionError::new).close();
+ assertThat(sharedState.unlockCallCount.get(), is(1));
}
@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));
+ void assertCloseOnlyUnlockOnce() {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true));
+ final ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ lockHandle.close();
+ lockHandle.close();
+ assertThat(sharedState.unlockCallCount.get(), is(1));
}
@Test
- void assertStartReturnsHandleAfterClose() {
- StubDistributedLock distributedLock = new StubDistributedLock(true, true);
- ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock));
- ExclusiveLockHandle firstHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ void assertStartAfterClose() {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, true, true));
+ final ExclusiveLockHandle firstHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
firstHandle.close();
- Optional<ExclusiveLockHandle> actual = context.start("op", 50L);
+ final Optional<ExclusiveLockHandle> actual = context.start("op", 50L);
assertTrue(actual.isPresent());
assertThat(actual.orElseThrow(AssertionError::new), not(firstHandle));
}
@Test
- void assertStartRemovesOperationKeyAfterTryLockFailure() {
- StubDistributedLock distributedLock = new StubDistributedLock(false, true);
- ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new StubClusterPersistRepository(distributedLock));
+ void assertStartAfterTryLockFailure() {
+ final SharedState sharedState = new SharedState();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(new FreshLockClusterPersistRepository(sharedState, false, true));
assertFalse(context.start("op", 50L).isPresent());
assertTrue(context.start("op", 50L).isPresent());
- assertThat(distributedLock.tryLockCallCount, is(2));
+ assertThat(sharedState.tryLockCallCount.get(), is(2));
+ }
+
+ @Test
+ void assertStartWithFreshLockInstances() {
+ final SharedState sharedState = new SharedState();
+ final FreshLockClusterPersistRepository repository = new FreshLockClusterPersistRepository(sharedState, true, true);
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(repository);
+ final ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ try {
+ assertFalse(context.start("op", 1L).isPresent());
+ assertThat(repository.getDistributedLockCallCount.get(), is(1));
+ } finally {
+ lockHandle.close();
+ }
+ context.start("op", 50L).orElseThrow(AssertionError::new).close();
+ assertThat(repository.getDistributedLockCallCount.get(), is(2));
+ }
+
+ @Test
+ void assertStartWithDefaultDistributedLockFallback() {
+ final DefaultLockClusterPersistRepository repository = new DefaultLockClusterPersistRepository();
+ final ClusterExclusiveOperatorContext context = new ClusterExclusiveOperatorContext(repository);
+ final ExclusiveLockHandle lockHandle = context.start("op", 50L).orElseThrow(AssertionError::new);
+ try {
+ assertFalse(context.start("op", 1L).isPresent());
+ assertThat(repository.persistExclusiveEphemeralCallCount.get(), is(1));
+ } finally {
+ lockHandle.close();
+ }
+ context.start("op", 50L).orElseThrow(AssertionError::new).close();
+ assertThat(repository.persistExclusiveEphemeralCallCount.get(), is(2));
}
- private static final class StubDistributedLock implements DistributedLock {
+ private static final class SharedState {
+
+ private final AtomicInteger tryLockCallCount = new AtomicInteger();
+
+ private final AtomicInteger unlockCallCount = new AtomicInteger();
+
+ private final List<Boolean> tryLockResults = new LinkedList<>();
- private final List<Boolean> tryLockResults;
+ private void addTryLockResult(final boolean tryLockResult) {
+ tryLockResults.add(tryLockResult);
+ }
+ }
+
+ private static final class FreshLockClusterPersistRepository implements ClusterPersistRepository {
- private int tryLockCallCount;
+ private final SharedState sharedState;
- private int unlockCallCount;
+ private final AtomicInteger getDistributedLockCallCount = new AtomicInteger();
- private StubDistributedLock(final boolean... tryLockResults) {
- this.tryLockResults = new LinkedList<>();
+ private FreshLockClusterPersistRepository(final SharedState sharedState, final boolean... tryLockResults) {
+ this.sharedState = sharedState;
for (boolean each : tryLockResults) {
- this.tryLockResults.add(each);
+ sharedState.addTryLockResult(each);
}
}
+ @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) {
+ getDistributedLockCallCount.incrementAndGet();
+ return Optional.of(new FreshDistributedLock(sharedState));
+ }
+
+ @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";
+ }
+ }
+
+ private static final class FreshDistributedLock implements DistributedLock {
+
+ private final SharedState sharedState;
+
+ private FreshDistributedLock(final SharedState sharedState) {
+ this.sharedState = sharedState;
+ }
+
@Override
public boolean tryLock(final long timeoutMillis) {
- tryLockCallCount++;
- return tryLockResults.remove(0);
+ sharedState.tryLockCallCount.incrementAndGet();
+ return sharedState.tryLockResults.remove(0);
}
@Override
public void unlock() {
- unlockCallCount++;
+ sharedState.unlockCallCount.incrementAndGet();
}
}
- private static final class StubClusterPersistRepository implements ClusterPersistRepository {
+ private static final class DefaultLockClusterPersistRepository implements ClusterPersistRepository {
- private final DistributedLock distributedLock;
+ private final Set<String> ephemeralKeys = ConcurrentHashMap.newKeySet();
- private StubClusterPersistRepository(final DistributedLock distributedLock) {
- this.distributedLock = distributedLock;
- }
+ private final AtomicInteger persistExclusiveEphemeralCallCount = new AtomicInteger();
@Override
public void init(final ClusterPersistRepositoryConfiguration config, final ComputeNodeInstanceContext computeNodeInstanceContext) {
@@ -133,7 +289,7 @@ class ClusterExclusiveOperatorContextTest {
@Override
public boolean isExisted(final String key) {
- return false;
+ return ephemeralKeys.contains(key);
}
@Override
@@ -146,20 +302,23 @@ class ClusterExclusiveOperatorContextTest {
@Override
public void persistEphemeral(final String key, final String value) {
+ ephemeralKeys.add(key);
}
@Override
public boolean persistExclusiveEphemeral(final String key, final String value) {
- return false;
+ persistExclusiveEphemeralCallCount.incrementAndGet();
+ return ephemeralKeys.add(key);
}
@Override
public Optional<DistributedLock> getDistributedLock(final String lockKey) {
- return Optional.of(distributedLock);
+ return Optional.empty();
}
@Override
public void delete(final String key) {
+ ephemeralKeys.remove(key);
}
@Override