This is an automated email from the ASF dual-hosted git repository.
terrymanu 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 de019fb9713 Decouple cluster lock strategies from Proxy runtime
(#40050)
de019fb9713 is described below
commit de019fb9713f6a0e9ea2e30e5bda3e6c7bd25c89
Author: Liang Zhang <[email protected]>
AuthorDate: Sun Oct 4 20:11:03 2026 +0800
Decouple cluster lock strategies from Proxy runtime (#40050)
---
.../lock/impl/ClusterReadWriteLockStrategy.java | 8 ++--
.../lock/impl/ClusterWriteLockStrategy.java | 8 ++--
.../ral/updatable/lock/LockClusterExecutor.java | 2 +-
...de.manager.cluster.lock.spi.ClusterLockStrategy | 4 +-
.../update/DistSQLUpdateExecuteEngineTest.java | 15 ++++++++
.../impl/ClusterReadWriteLockStrategyTest.java | 30 +++++++--------
.../lock/impl/ClusterWriteLockStrategyTest.java | 30 +++++++--------
.../updatable/lock/LockClusterExecutorTest.java | 43 ++++++++--------------
.../statement/DistSQLStatementExecutorTest.java | 41 +++++++++++++++++++++
.../cluster/lock/spi/ClusterLockStrategy.java | 8 +++-
.../DistSQLUpdateProxyBackendHandlerTest.java | 37 +++++++++++++++++++
11 files changed, 156 insertions(+), 70 deletions(-)
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategy.java
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategy.java
similarity index 79%
rename from
proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategy.java
rename to
infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategy.java
index 487151f3c17..5ede4a4ea54 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategy.java
+++
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategy.java
@@ -15,11 +15,11 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.proxy.backend.lock.impl;
+package org.apache.shardingsphere.distsql.handler.executor.lock.impl;
+import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy;
import org.apache.shardingsphere.mode.state.ShardingSphereState;
-import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
/**
* Cluster read-write lock strategy.
@@ -27,8 +27,8 @@ import
org.apache.shardingsphere.proxy.backend.context.ProxyContext;
public class ClusterReadWriteLockStrategy implements ClusterLockStrategy {
@Override
- public void lock() {
-
ProxyContext.getInstance().getContextManager().getPersistServiceFacade().getStateService().update(ShardingSphereState.UNAVAILABLE);
+ public void lock(final ContextManager contextManager) {
+
contextManager.getPersistServiceFacade().getStateService().update(ShardingSphereState.UNAVAILABLE);
}
@Override
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategy.java
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategy.java
similarity index 79%
rename from
proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategy.java
rename to
infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategy.java
index 24affb5dce7..5ca2b45be57 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategy.java
+++
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategy.java
@@ -15,11 +15,11 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.proxy.backend.lock.impl;
+package org.apache.shardingsphere.distsql.handler.executor.lock.impl;
+import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy;
import org.apache.shardingsphere.mode.state.ShardingSphereState;
-import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
/**
* Cluster write lock strategy.
@@ -27,8 +27,8 @@ import
org.apache.shardingsphere.proxy.backend.context.ProxyContext;
public class ClusterWriteLockStrategy implements ClusterLockStrategy {
@Override
- public void lock() {
-
ProxyContext.getInstance().getContextManager().getPersistServiceFacade().getStateService().update(ShardingSphereState.READ_ONLY);
+ public void lock(final ContextManager contextManager) {
+
contextManager.getPersistServiceFacade().getStateService().update(ShardingSphereState.READ_ONLY);
}
@Override
diff --git
a/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutor.java
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutor.java
index 5d0ce9cc893..de798712662 100644
---
a/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutor.java
+++
b/infra/distsql-handler/src/main/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutor.java
@@ -43,7 +43,7 @@ public final class LockClusterExecutor implements
DistSQLUpdateExecutor<LockClus
checkAlgorithm(sqlStatement);
contextManager.getExclusiveOperatorEngine().operate(new
LockClusterOperation(), sqlStatement.getTimeoutMillis(), () -> {
checkState(contextManager);
- TypedSPILoader.getService(ClusterLockStrategy.class,
sqlStatement.getLockStrategy().getName()).lock();
+ TypedSPILoader.getService(ClusterLockStrategy.class,
sqlStatement.getLockStrategy().getName()).lock(contextManager);
});
}
diff --git
a/proxy/backend/core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
b/infra/distsql-handler/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
similarity index 81%
rename from
proxy/backend/core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
rename to
infra/distsql-handler/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
index 1fafee094fc..fe9f0043dc4 100644
---
a/proxy/backend/core/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
+++
b/infra/distsql-handler/src/main/resources/META-INF/services/org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy
@@ -15,5 +15,5 @@
# limitations under the License.
#
-org.apache.shardingsphere.proxy.backend.lock.impl.ClusterWriteLockStrategy
-org.apache.shardingsphere.proxy.backend.lock.impl.ClusterReadWriteLockStrategy
+org.apache.shardingsphere.distsql.handler.executor.lock.impl.ClusterWriteLockStrategy
+org.apache.shardingsphere.distsql.handler.executor.lock.impl.ClusterReadWriteLockStrategy
diff --git
a/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/engine/update/DistSQLUpdateExecuteEngineTest.java
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/engine/update/DistSQLUpdateExecuteEngineTest.java
index 68b39315d76..bcc7db5a712 100644
---
a/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/engine/update/DistSQLUpdateExecuteEngineTest.java
+++
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/engine/update/DistSQLUpdateExecuteEngineTest.java
@@ -27,7 +27,9 @@ import
org.apache.shardingsphere.distsql.handler.engine.update.rdl.rule.spi.data
import
org.apache.shardingsphere.distsql.handler.engine.update.rdl.rule.spi.global.GlobalRuleDefinitionExecutor;
import
org.apache.shardingsphere.distsql.handler.engine.update.spi.AdvancedDistSQLUpdateExecutor;
import
org.apache.shardingsphere.distsql.handler.engine.update.spi.DistSQLUpdateExecutor;
+import org.apache.shardingsphere.distsql.segment.AlgorithmSegment;
import org.apache.shardingsphere.distsql.statement.DistSQLStatement;
+import
org.apache.shardingsphere.distsql.statement.type.ral.updatable.LockClusterStatement;
import
org.apache.shardingsphere.distsql.statement.type.rdl.rule.database.DatabaseRuleDefinitionStatement;
import
org.apache.shardingsphere.distsql.statement.type.rdl.rule.global.GlobalRuleDefinitionStatement;
import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
@@ -35,6 +37,7 @@ import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.rule.RuleMetaData;
import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.exception.NotClusterModeException;
import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.persist.service.MetaDataManagerPersistService;
import
org.apache.shardingsphere.sql.parser.statement.core.statement.attribute.type.FromDatabaseSQLStatementAttribute;
@@ -43,8 +46,12 @@ import org.mockito.MockedStatic;
import java.sql.SQLException;
import java.util.Optional;
+import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
@@ -57,6 +64,14 @@ import static org.mockito.Mockito.withSettings;
@SuppressWarnings("unchecked")
class DistSQLUpdateExecuteEngineTest {
+ @Test
+ void assertExecuteUpdateWithLockClusterInStandaloneMode() throws
SQLException {
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ LockClusterStatement sqlStatement = new LockClusterStatement(new
AlgorithmSegment("WRITE", new Properties()));
+ assertThrows(NotClusterModeException.class, () -> new
DistSQLUpdateExecuteEngine(sqlStatement, null, contextManager,
null).executeUpdate());
+ verify(contextManager.getExclusiveOperatorEngine(),
never()).operate(any(), anyLong(), any());
+ }
+
@Test
void assertExecuteUpdateWithGlobalRuleDefinitionStatement() throws
SQLException {
GlobalRuleDefinitionStatement sqlStatement =
mock(GlobalRuleDefinitionStatement.class, RETURNS_DEEP_STUBS);
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategyTest.java
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategyTest.java
similarity index 58%
rename from
proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategyTest.java
rename to
infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategyTest.java
index 16be664878c..7f897269c80 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategyTest.java
+++
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterReadWriteLockStrategyTest.java
@@ -15,36 +15,36 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.proxy.backend.lock.impl;
+package org.apache.shardingsphere.distsql.handler.executor.lock.impl;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy;
import org.apache.shardingsphere.mode.state.ShardingSphereState;
-import org.apache.shardingsphere.mode.state.StatePersistService;
-import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
-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 static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-@ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(ProxyContext.class)
class ClusterReadWriteLockStrategyTest {
private final ClusterLockStrategy clusterLockStrategy =
TypedSPILoader.getService(ClusterLockStrategy.class, "READ_WRITE");
@Test
void assertLock() {
- StatePersistService stateService = mock(StatePersistService.class);
- ProxyContext proxyContext = mock(ProxyContext.class,
RETURNS_DEEP_STUBS);
-
when(proxyContext.getContextManager().getPersistServiceFacade().getStateService()).thenReturn(stateService);
- when(ProxyContext.getInstance()).thenReturn(proxyContext);
- clusterLockStrategy.lock();
- verify(stateService).update(ShardingSphereState.UNAVAILABLE);
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ clusterLockStrategy.lock(contextManager);
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.UNAVAILABLE);
+ }
+
+ @Test
+ void assertLockWithDifferentContextManagers() {
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ ContextManager anotherContextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ clusterLockStrategy.lock(contextManager);
+ clusterLockStrategy.lock(anotherContextManager);
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.UNAVAILABLE);
+
verify(anotherContextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.UNAVAILABLE);
}
}
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategyTest.java
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategyTest.java
similarity index 58%
rename from
proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategyTest.java
rename to
infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategyTest.java
index 7995ef19162..3485b8613bc 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategyTest.java
+++
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/lock/impl/ClusterWriteLockStrategyTest.java
@@ -15,36 +15,36 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.proxy.backend.lock.impl;
+package org.apache.shardingsphere.distsql.handler.executor.lock.impl;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.cluster.lock.spi.ClusterLockStrategy;
import org.apache.shardingsphere.mode.state.ShardingSphereState;
-import org.apache.shardingsphere.mode.state.StatePersistService;
-import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
-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 static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-@ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(ProxyContext.class)
class ClusterWriteLockStrategyTest {
private final ClusterLockStrategy clusterLockStrategy =
TypedSPILoader.getService(ClusterLockStrategy.class, "WRITE");
@Test
void assertLock() {
- StatePersistService stateService = mock(StatePersistService.class);
- ProxyContext proxyContext = mock(ProxyContext.class,
RETURNS_DEEP_STUBS);
-
when(proxyContext.getContextManager().getPersistServiceFacade().getStateService()).thenReturn(stateService);
- when(ProxyContext.getInstance()).thenReturn(proxyContext);
- clusterLockStrategy.lock();
- verify(stateService).update(ShardingSphereState.READ_ONLY);
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ clusterLockStrategy.lock(contextManager);
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.READ_ONLY);
+ }
+
+ @Test
+ void assertLockWithDifferentContextManagers() {
+ ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ ContextManager anotherContextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
+ clusterLockStrategy.lock(contextManager);
+ clusterLockStrategy.lock(anotherContextManager);
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.READ_ONLY);
+
verify(anotherContextManager.getPersistServiceFacade().getStateService()).update(ShardingSphereState.READ_ONLY);
}
}
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/lock/LockClusterExecutorTest.java
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutorTest.java
similarity index 79%
rename from
proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/lock/LockClusterExecutorTest.java
rename to
infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutorTest.java
index 7c726c823db..44380d31f19 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/lock/LockClusterExecutorTest.java
+++
b/infra/distsql-handler/src/test/java/org/apache/shardingsphere/distsql/handler/executor/ral/updatable/lock/LockClusterExecutorTest.java
@@ -15,10 +15,9 @@
* limitations under the License.
*/
-package
org.apache.shardingsphere.proxy.backend.handler.distsql.ral.updatable.lock;
+package org.apache.shardingsphere.distsql.handler.executor.ral.updatable.lock;
import
org.apache.shardingsphere.distsql.handler.engine.update.spi.DistSQLUpdateExecutor;
-import
org.apache.shardingsphere.distsql.handler.executor.ral.updatable.lock.LockClusterExecutor;
import org.apache.shardingsphere.distsql.segment.AlgorithmSegment;
import
org.apache.shardingsphere.distsql.statement.type.ral.updatable.LockClusterStatement;
import
org.apache.shardingsphere.infra.algorithm.core.exception.MissingRequiredAlgorithmException;
@@ -30,10 +29,6 @@ import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.mode.manager.cluster.lock.exception.LockedClusterException;
import org.apache.shardingsphere.mode.state.ShardingSphereState;
import org.apache.shardingsphere.mode.state.StatePersistService;
-import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
-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.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -54,8 +49,6 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-@ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(ProxyContext.class)
class LockClusterExecutorTest {
private final LockClusterExecutor executor = (LockClusterExecutor)
TypedSPILoader.getService(DistSQLUpdateExecutor.class,
LockClusterStatement.class);
@@ -74,16 +67,12 @@ class LockClusterExecutorTest {
@MethodSource("provideOperationScenarios")
void assertExecuteUpdateWithOperationScenarios(final String name, final
LockClusterStatement sqlStatement,
final long
expectedTimeoutMillis, final ShardingSphereState callbackState,
- final Class<? extends
Throwable> expectedException, final boolean expectStateUpdated) throws
SQLException {
+ final Class<? extends
Throwable> expectedException, final ShardingSphereState expectedState) throws
SQLException {
ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
ExclusiveOperatorEngine exclusiveOperatorEngine =
mock(ExclusiveOperatorEngine.class);
when(contextManager.getExclusiveOperatorEngine()).thenReturn(exclusiveOperatorEngine);
when(contextManager.getStateContext().getState()).thenReturn(ShardingSphereState.OK,
callbackState);
- StatePersistService stateService = null;
- if (expectStateUpdated) {
- stateService = mock(StatePersistService.class);
- mockStateService(stateService);
- }
+ StatePersistService stateService =
contextManager.getPersistServiceFacade().getStateService();
doAnswer(invocation -> {
ExclusiveOperationVoidCallback callback =
invocation.getArgument(2);
callback.execute();
@@ -95,23 +84,19 @@ class LockClusterExecutorTest {
assertThrows(expectedException, () ->
executor.executeUpdate(sqlStatement, contextManager));
}
verify(exclusiveOperatorEngine).operate(any(),
eq(expectedTimeoutMillis), any(ExclusiveOperationVoidCallback.class));
- if (expectStateUpdated) {
- verify(stateService).update(ShardingSphereState.READ_ONLY);
+ if (null == expectedState) {
+ verify(stateService, never()).update(any());
+ } else {
+ verify(stateService).update(expectedState);
}
}
- private void mockStateService(final StatePersistService stateService) {
- ProxyContext proxyContext = mock(ProxyContext.class);
- ContextManager contextManager = mock(ContextManager.class,
RETURNS_DEEP_STUBS);
- when(proxyContext.getContextManager()).thenReturn(contextManager);
-
when(contextManager.getPersistServiceFacade().getStateService()).thenReturn(stateService);
- when(ProxyContext.getInstance()).thenReturn(proxyContext);
- }
-
private static Stream<Arguments> provideFailureScenarios() {
return Stream.of(
Arguments.of("cluster state is unavailable", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()), 2000L),
ShardingSphereState.UNAVAILABLE,
LockedClusterException.class),
+ Arguments.of("cluster state is read only", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()), 2000L),
+ ShardingSphereState.READ_ONLY,
LockedClusterException.class),
Arguments.of("lock strategy is required", new
LockClusterStatement(null, 2000L), ShardingSphereState.OK,
MissingRequiredAlgorithmException.class),
Arguments.of("lock strategy is unsupported", new
LockClusterStatement(new AlgorithmSegment("FOO", new Properties()), 2000L),
ShardingSphereState.OK,
ServiceProviderNotFoundException.class));
@@ -120,10 +105,14 @@ class LockClusterExecutorTest {
private static Stream<Arguments> provideOperationScenarios() {
return Stream.of(
Arguments.of("use explicit timeout", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()), 2000L),
- 2000L, ShardingSphereState.OK, null, true),
+ 2000L, ShardingSphereState.OK, null,
ShardingSphereState.READ_ONLY),
Arguments.of("use default timeout when absent", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties())),
- 3000L, ShardingSphereState.OK, null, true),
+ 3000L, ShardingSphereState.OK, null,
ShardingSphereState.READ_ONLY),
+ Arguments.of("lock reads and writes", new
LockClusterStatement(new AlgorithmSegment("READ_WRITE", new Properties()),
2000L),
+ 2000L, ShardingSphereState.OK, null,
ShardingSphereState.UNAVAILABLE),
Arguments.of("throw exception when callback state changes",
new LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()),
2000L),
- 2000L, ShardingSphereState.UNAVAILABLE,
LockedClusterException.class, false));
+ 2000L, ShardingSphereState.UNAVAILABLE,
LockedClusterException.class, null),
+ Arguments.of("throw exception when callback state becomes read
only", new LockClusterStatement(new AlgorithmSegment("WRITE", new
Properties()), 2000L),
+ 2000L, ShardingSphereState.READ_ONLY,
LockedClusterException.class, null));
}
}
diff --git
a/jdbc/src/test/java/org/apache/shardingsphere/driver/jdbc/core/statement/DistSQLStatementExecutorTest.java
b/jdbc/src/test/java/org/apache/shardingsphere/driver/jdbc/core/statement/DistSQLStatementExecutorTest.java
index 3544456d3b4..b9b61e965a1 100644
---
a/jdbc/src/test/java/org/apache/shardingsphere/driver/jdbc/core/statement/DistSQLStatementExecutorTest.java
+++
b/jdbc/src/test/java/org/apache/shardingsphere/driver/jdbc/core/statement/DistSQLStatementExecutorTest.java
@@ -20,28 +20,43 @@ package
org.apache.shardingsphere.driver.jdbc.core.statement;
import
org.apache.shardingsphere.distsql.handler.engine.DistSQLConnectionContext;
import
org.apache.shardingsphere.distsql.handler.engine.query.DistSQLQueryExecuteEngine;
import
org.apache.shardingsphere.distsql.handler.engine.update.DistSQLUpdateExecuteEngine;
+import org.apache.shardingsphere.distsql.segment.AlgorithmSegment;
import org.apache.shardingsphere.distsql.statement.DistSQLStatement;
+import
org.apache.shardingsphere.distsql.statement.type.ral.updatable.LockClusterStatement;
import
org.apache.shardingsphere.distsql.statement.type.rdl.resource.unit.type.UnregisterStorageUnitStatement;
import org.apache.shardingsphere.distsql.statement.type.rul.sql.ParseStatement;
import
org.apache.shardingsphere.driver.jdbc.core.connection.ShardingSphereConnection;
import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import
org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
import org.apache.shardingsphere.infra.session.query.QueryContext;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveOperatorEngine;
+import
org.apache.shardingsphere.mode.exclusive.callback.ExclusiveOperationVoidCallback;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.apache.shardingsphere.mode.state.ShardingSphereState;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.MockedConstruction;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collections;
+import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.verify;
@@ -83,6 +98,32 @@ class DistSQLStatementExecutorTest {
}
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideLockClusterStatements")
+ void assertExecuteUpdateWithLockCluster(final String name, final
LockClusterStatement sqlStatement,
+ final long expectedTimeoutMillis,
final ShardingSphereState expectedState) throws SQLException {
+ ShardingSphereConnection connection =
mock(ShardingSphereConnection.class, RETURNS_DEEP_STUBS);
+ ContextManager contextManager = connection.getContextManager();
+
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration().isCluster()).thenReturn(true);
+
when(contextManager.getStateContext().getState()).thenReturn(ShardingSphereState.OK);
+ ExclusiveOperatorEngine exclusiveOperatorEngine =
contextManager.getExclusiveOperatorEngine();
+ doAnswer(invocation -> {
+ ExclusiveOperationVoidCallback callback =
invocation.getArgument(2);
+ callback.execute();
+ return null;
+ }).when(exclusiveOperatorEngine).operate(any(), anyLong(),
any(ExclusiveOperationVoidCallback.class));
+ new DistSQLStatementExecutor(connection, mock(StatementManager.class),
mock(Statement.class)).executeUpdate(createQueryContext(sqlStatement));
+ verify(exclusiveOperatorEngine).operate(any(),
eq(expectedTimeoutMillis), any(ExclusiveOperationVoidCallback.class));
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(expectedState);
+ }
+
+ private static Stream<Arguments> provideLockClusterStatements() {
+ return Stream.of(
+ Arguments.of("write lock with explicit timeout", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()), 2000L),
2000L, ShardingSphereState.READ_ONLY),
+ Arguments.of("write lock with default timeout", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties())), 3000L,
ShardingSphereState.READ_ONLY),
+ Arguments.of("read-write lock with explicit timeout", new
LockClusterStatement(new AlgorithmSegment("READ_WRITE", new Properties()),
2000L), 2000L, ShardingSphereState.UNAVAILABLE));
+ }
+
@Test
void assertIsQuery() {
QueryContext queryContext =
createQueryContext(mock(DistSQLStatement.class));
diff --git
a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/lock/spi/ClusterLockStrategy.java
b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/lock/spi/ClusterLockStrategy.java
index 9fa4448f58e..404489fcd6b 100644
---
a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/lock/spi/ClusterLockStrategy.java
+++
b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/lock/spi/ClusterLockStrategy.java
@@ -19,6 +19,7 @@ package
org.apache.shardingsphere.mode.manager.cluster.lock.spi;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
+import org.apache.shardingsphere.mode.manager.ContextManager;
/**
* Cluster lock strategy.
@@ -27,7 +28,10 @@ import
org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
public interface ClusterLockStrategy extends TypedSPI {
/**
- * Lock cluster.
+ * Lock the cluster associated with the caller's context manager.
+ * Implementations must not retain the context manager because strategies
are shared singletons.
+ *
+ * @param contextManager caller's context manager
*/
- void lock();
+ void lock(ContextManager contextManager);
}
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/DistSQLUpdateProxyBackendHandlerTest.java
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/DistSQLUpdateProxyBackendHandlerTest.java
index 99cef91eb04..c097e70499b 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/DistSQLUpdateProxyBackendHandlerTest.java
+++
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/DistSQLUpdateProxyBackendHandlerTest.java
@@ -18,6 +18,8 @@
package org.apache.shardingsphere.proxy.backend.handler.distsql;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import org.apache.shardingsphere.distsql.segment.AlgorithmSegment;
+import
org.apache.shardingsphere.distsql.statement.type.ral.updatable.LockClusterStatement;
import
org.apache.shardingsphere.distsql.statement.type.ral.updatable.RefreshTableMetaDataStatement;
import org.apache.shardingsphere.infra.config.props.ConfigurationProperties;
import
org.apache.shardingsphere.infra.exception.kernel.metadata.SchemaNotFoundException;
@@ -31,12 +33,18 @@ import
org.apache.shardingsphere.infra.metadata.database.rule.RuleMetaData;
import
org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereSchema;
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.mode.exclusive.ExclusiveOperatorEngine;
+import
org.apache.shardingsphere.mode.exclusive.callback.ExclusiveOperationVoidCallback;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.apache.shardingsphere.mode.state.ShardingSphereState;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
import
org.apache.shardingsphere.sql.parser.statement.core.value.identifier.IdentifierValue;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Answers;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -44,13 +52,18 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.sql.SQLException;
import java.util.Collections;
import java.util.Properties;
+import java.util.stream.Stream;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.isA;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -130,6 +143,30 @@ class DistSQLUpdateProxyBackendHandlerTest {
assertThat(new DistSQLUpdateProxyBackendHandler(sqlStatement,
mockQueryContext(), mockConnectionSession(), contextManager).execute(),
isA(UpdateResponseHeader.class));
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("provideLockClusterStatements")
+ void assertUpdateWithLockCluster(final String name, final
LockClusterStatement sqlStatement,
+ final long expectedTimeoutMillis, final
ShardingSphereState expectedState) throws SQLException {
+
when(contextManager.getComputeNodeInstanceContext().getModeConfiguration().isCluster()).thenReturn(true);
+
when(contextManager.getStateContext().getState()).thenReturn(ShardingSphereState.OK);
+ ExclusiveOperatorEngine exclusiveOperatorEngine =
contextManager.getExclusiveOperatorEngine();
+ doAnswer(invocation -> {
+ ExclusiveOperationVoidCallback callback =
invocation.getArgument(2);
+ callback.execute();
+ return null;
+ }).when(exclusiveOperatorEngine).operate(any(), anyLong(),
any(ExclusiveOperationVoidCallback.class));
+ assertThat(new DistSQLUpdateProxyBackendHandler(sqlStatement,
mockQueryContext(), mockConnectionSession(), contextManager).execute(),
isA(UpdateResponseHeader.class));
+ verify(exclusiveOperatorEngine).operate(any(),
eq(expectedTimeoutMillis), any(ExclusiveOperationVoidCallback.class));
+
verify(contextManager.getPersistServiceFacade().getStateService()).update(expectedState);
+ }
+
+ private static Stream<Arguments> provideLockClusterStatements() {
+ return Stream.of(
+ Arguments.of("write lock with explicit timeout", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties()), 2000L),
2000L, ShardingSphereState.READ_ONLY),
+ Arguments.of("write lock with default timeout", new
LockClusterStatement(new AlgorithmSegment("WRITE", new Properties())), 3000L,
ShardingSphereState.READ_ONLY),
+ Arguments.of("read-write lock with explicit timeout", new
LockClusterStatement(new AlgorithmSegment("READ_WRITE", new Properties()),
2000L), 2000L, ShardingSphereState.UNAVAILABLE));
+ }
+
private QueryContext mockQueryContext() {
return mock(QueryContext.class, RETURNS_DEEP_STUBS);
}
