Use ProcessPersistService instead of ProcessListEvent and remove Proc…

…essSubscriber (#31411)
apache · May 27, 2024 · d0abb75 · d0abb75
1 parent da31876
commit d0abb75
Show file tree

Hide file tree

Showing 13 changed files with 58 additions and 352 deletions.
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/process/ProcessSubscriber.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/process/ProcessSubscriber.java
diff --git a/...e/src/main/java/org/apache/shardingsphere/mode/process/event/KillProcessRequestEvent.java b/...e/src/main/java/org/apache/shardingsphere/mode/process/event/KillProcessRequestEvent.java
diff --git a/...c/main/java/org/apache/shardingsphere/mode/process/event/ShowProcessListRequestEvent.java b/...c/main/java/org/apache/shardingsphere/mode/process/event/ShowProcessListRequestEvent.java
diff --git a/.../main/java/org/apache/shardingsphere/mode/process/event/ShowProcessListResponseEvent.java b/.../main/java/org/apache/shardingsphere/mode/process/event/ShowProcessListResponseEvent.java
diff --git a/...ain/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java b/...ain/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
@@ -34,7 +34,6 @@
 import org.apache.shardingsphere.mode.manager.ContextManagerBuilderParameter;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.GlobalLockPersistService;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.GovernanceWatcherFactory;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.process.subscriber.ClusterProcessSubscriber;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.storage.subscriber.QualifiedDataSourceStatusSubscriber;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.workerid.generator.ClusterWorkerIdGenerator;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.subscriber.ClusterEventSubscriberRegistry;
@@ -97,7 +96,6 @@ private DistributedLockHolder initDistributedLockHolder(final ClusterPersistRepo
     // 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 QualifiedDataSourceStatusSubscriber(repository));
-        eventBusContext.register(new ClusterProcessSubscriber(repository, eventBusContext));
     }
 
     private void registerOnline(final EventBusContext eventBusContext, final ComputeNodeInstanceContext computeNodeInstanceContext,

diff --git a/...ode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessSubscriber.java b/...ode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessSubscriber.java
diff --git a/...scriber/ClusterProcessSubscriberTest.java → ...ice/ClusterProcessPersistServiceTest.java b/...scriber/ClusterProcessSubscriberTest.java → ...ice/ClusterProcessPersistServiceTest.java
@@ -15,12 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.process.subscriber;
+package org.apache.shardingsphere.mode.manager.cluster.service;
 
 import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
-import org.apache.shardingsphere.infra.util.eventbus.EventBusContext;
 import org.apache.shardingsphere.metadata.persist.node.ComputeNode;
-import org.apache.shardingsphere.mode.process.event.ShowProcessListRequestEvent;
 import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -35,26 +33,32 @@
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-class ClusterProcessSubscriberTest {
+class ClusterProcessPersistServiceTest {
 
     @Mock
     private ClusterPersistRepository repository;
 
-    private final EventBusContext eventBusContext = new EventBusContext();
-
-    private ClusterProcessSubscriber clusterProcessListSubscriber;
+    private ClusterProcessPersistService processPersistService;
 
     @BeforeEach
     void setUp() {
-        clusterProcessListSubscriber = new ClusterProcessSubscriber(repository, eventBusContext);
+        processPersistService = new ClusterProcessPersistService(repository);
     }
 
     @Test
-    void assertPostShowProcessListData() {
+    void getProcessList() {
         when(repository.getChildrenKeys(ComputeNode.getOnlineNodePath(InstanceType.JDBC))).thenReturn(Collections.emptyList());
         when(repository.getChildrenKeys(ComputeNode.getOnlineNodePath(InstanceType.PROXY))).thenReturn(Collections.singletonList("abc"));
         when(repository.query(any())).thenReturn(null);
-        clusterProcessListSubscriber.postShowProcessListData(new ShowProcessListRequestEvent());
+        processPersistService.getProcessList();
+        verify(repository).persist(any(), any());
+    }
+
+    @Test
+    void killProcess() {
+        when(repository.getChildrenKeys(ComputeNode.getOnlineNodePath(InstanceType.JDBC))).thenReturn(Collections.emptyList());
+        when(repository.getChildrenKeys(ComputeNode.getOnlineNodePath(InstanceType.PROXY))).thenReturn(Collections.singletonList("abc"));
+        processPersistService.killProcess("foo_process_id");
         verify(repository).persist(any(), any());
     }
 }
diff --git a/.../shardingsphere/mode/manager/standalone/subscriber/StandaloneEventSubscriberRegistry.java b/.../shardingsphere/mode/manager/standalone/subscriber/StandaloneEventSubscriberRegistry.java
@@ -27,8 +27,6 @@
 public final class StandaloneEventSubscriberRegistry extends EventSubscriberRegistry {
 
     public StandaloneEventSubscriberRegistry(final ContextManager contextManager) {
-        super(contextManager,
-                new StandaloneProcessSubscriber(contextManager.getComputeNodeInstanceContext().getEventBusContext()),
-                new RuleItemChangedSubscriber(contextManager));
+        super(contextManager, new RuleItemChangedSubscriber(contextManager));
     }
 }
diff --git a/...apache/shardingsphere/mode/manager/standalone/subscriber/StandaloneProcessSubscriber.java b/...apache/shardingsphere/mode/manager/standalone/subscriber/StandaloneProcessSubscriber.java