Remove event bus for lock and unlock cluster

mode/core/src/main/java/org/apache/shardingsphere/mode/state/StateContext.java

@@ -50,4 +50,13 @@ public ClusterState getCurrentClusterState() {
     public void switchCurrentClusterState(final ClusterState state) {
         currentClusterState.set(state);
     }
+
+    /**
+     * Update cluster state.
+     *
+     * @param state cluster state
+     */
+    public void updateClusterState(final ClusterState state) {
+        stateService.persist(state);
+    }
 }

mode/core/src/main/java/org/apache/shardingsphere/mode/state/StateService.java

@@ -39,9 +39,7 @@ public final class StateService {
      * @param state cluster state
      */
     public void persist(final ClusterState state) {
-        if (Strings.isNullOrEmpty(repository.getDirectly(ComputeNode.getClusterStateNodePath()))) {
-            repository.persist(ComputeNode.getClusterStateNodePath(), state.name());
-        }
+        repository.persist(ComputeNode.getClusterStateNodePath(), state.name());
     }
 
     /**

mode/core/src/test/java/org/apache/shardingsphere/mode/state/StateContextTest.java

@@ -23,6 +23,7 @@
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 
 class StateContextTest {
 
@@ -39,4 +40,10 @@ void assertSwitchCurrentClusterState() {
         stateContext.switchCurrentClusterState(ClusterState.UNAVAILABLE);
         assertThat(stateContext.getCurrentClusterState(), is(ClusterState.UNAVAILABLE));
     }
+
+    @Test
+    void assertUpdateClusterState() {
+        stateContext.updateClusterState(ClusterState.OK);
+        verify(stateContext.getStateService()).persist(ClusterState.OK);
+    }
 }

mode/core/src/test/java/org/apache/shardingsphere/mode/state/StateServiceTest.java

@@ -25,9 +25,7 @@
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.times;
 
 @ExtendWith(MockitoExtension.class)
 class StateServiceTest {
@@ -42,14 +40,6 @@ void assertPersistClusterStateWithoutPath() {
         verify(repository).persist(ComputeNode.getClusterStateNodePath(), ClusterState.OK.name());
     }
 
-    @Test
-    void assertPersistClusterStateWithPath() {
-        StateService stateService = new StateService(repository);
-        when(repository.getDirectly("/nodes/compute_nodes/status")).thenReturn(ClusterState.OK.name());
-        stateService.persist(ClusterState.OK);
-        verify(repository, times(0)).persist(ComputeNode.getClusterStateNodePath(), ClusterState.OK.name());
-    }
-
     @Test
     void assertLoadClusterState() {
         new StateService(repository).load();

mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java

@@ -37,7 +37,6 @@
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.GovernanceWatcherFactory;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.metadata.subscriber.ShardingSphereSchemaDataRegistrySubscriber;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.process.subscriber.ClusterProcessSubscriber;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.cluster.subscriber.ClusterStateSubscriber;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service.ComputeNodeStatusService;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.subscriber.ComputeNodeStatusSubscriber;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.storage.subscriber.QualifiedDataSourceStatusSubscriber;
@@ -106,7 +105,6 @@ private void setContextManagerAware(final ContextManager contextManager) {
     // TODO remove the method, only keep ZooKeeper's events, remove all decouple events
     private void createSubscribers(final EventBusContext eventBusContext, final ClusterPersistRepository repository) {
         eventBusContext.register(new ComputeNodeStatusSubscriber(repository));
-        eventBusContext.register(new ClusterStateSubscriber(repository));
         eventBusContext.register(new QualifiedDataSourceStatusSubscriber(repository));
         eventBusContext.register(new ClusterProcessSubscriber(repository, eventBusContext));
         eventBusContext.register(new ShardingSphereSchemaDataRegistrySubscriber(repository));

proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/UnlockClusterExecutor.java

@@ -27,7 +27,6 @@
 import org.apache.shardingsphere.mode.exception.NotLockedClusterException;
 import org.apache.shardingsphere.mode.lock.GlobalLockDefinition;
 import org.apache.shardingsphere.mode.manager.ContextManager;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.cluster.event.ClusterStateChangedEvent;
 
 /**
  * Unlock cluster executor.
@@ -44,7 +43,7 @@ public void executeUpdate(final UnlockClusterStatement sqlStatement, final Conte
         if (lockContext.tryLock(lockDefinition, 3000L)) {
             try {
                 checkState(contextManager);
-                contextManager.getComputeNodeInstanceContext().getEventBusContext().post(new ClusterStateChangedEvent(ClusterState.OK));
+                contextManager.getStateContext().updateClusterState(ClusterState.OK);
                 // TODO unlock snapshot info if locked
             } finally {
                 lockContext.unlock(lockDefinition);

proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterReadWriteLockStrategy.java

@@ -18,7 +18,6 @@
 package org.apache.shardingsphere.proxy.backend.lock.impl;
 
 import org.apache.shardingsphere.infra.state.cluster.ClusterState;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.cluster.event.ClusterStateChangedEvent;
 import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
 import org.apache.shardingsphere.proxy.backend.lock.spi.ClusterLockStrategy;
 
@@ -29,7 +28,7 @@ public class ClusterReadWriteLockStrategy implements ClusterLockStrategy {
 
     @Override
     public void lock() {
-        ProxyContext.getInstance().getContextManager().getComputeNodeInstanceContext().getEventBusContext().post(new ClusterStateChangedEvent(ClusterState.UNAVAILABLE));
+        ProxyContext.getInstance().getContextManager().getStateContext().updateClusterState(ClusterState.UNAVAILABLE);
     }
 
     @Override

proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/lock/impl/ClusterWriteLockStrategy.java

@@ -18,7 +18,6 @@
 package org.apache.shardingsphere.proxy.backend.lock.impl;
 
 import org.apache.shardingsphere.infra.state.cluster.ClusterState;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.cluster.event.ClusterStateChangedEvent;
 import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
 import org.apache.shardingsphere.proxy.backend.lock.spi.ClusterLockStrategy;
 
@@ -29,7 +28,7 @@ public class ClusterWriteLockStrategy implements ClusterLockStrategy {
 
     @Override
     public void lock() {
-        ProxyContext.getInstance().getContextManager().getComputeNodeInstanceContext().getEventBusContext().post(new ClusterStateChangedEvent(ClusterState.READ_ONLY));
+        ProxyContext.getInstance().getContextManager().getStateContext().updateClusterState(ClusterState.READ_ONLY);
     }
 
     @Override