You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by du...@apache.org on 2024/05/22 04:21:25 UTC
(shardingsphere) branch master updated: Use ContextServiceFacade to decouple EventBus for set instance state and label instance (#31338)
This is an automated email from the ASF dual-hosted git repository.
duanzhengqiang 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 fce413bff84 Use ContextServiceFacade to decouple EventBus for set instance state and label instance (#31338)
fce413bff84 is described below
commit fce413bff84290f24fa915048f4893e165f49b50
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Wed May 22 12:21:18 2024 +0800
Use ContextServiceFacade to decouple EventBus for set instance state and label instance (#31338)
* Use ContextServiceFacade to decouple EventBus for set instance state and label instance
* Use ContextServiceFacade to decouple EventBus for set instance state and label instance
* Use ContextServiceFacade to decouple EventBus for set instance state and label instance
---
.../shardingsphere/mode/spi/PersistRepository.java | 9 ++++
.../event/node/ComputeNodeStatusChangedEvent.java | 34 ------------
.../mode/manager/ContextManager.java | 4 ++
.../mode/service/ComputeNodeService.java} | 18 +++++--
.../mode/service/ContextServiceFacade.java} | 17 +++---
.../mode/service/ComputeNodeServiceTest.java} | 36 +++++++------
.../cluster/ClusterContextManagerBuilder.java | 8 +--
.../subscriber/ComputeNodeStatusSubscriber.java | 61 ----------------------
.../generator/ClusterWorkerIdGenerator.java | 12 ++---
.../subscriber/ClusterEventSubscriberRegistry.java | 2 +-
.../subscriber/StateChangedSubscriber.java | 9 +---
.../subscriber/StateChangedSubscriberTest.java | 3 +-
.../cluster/ClusterPersistRepository.java | 8 ---
.../SessionConnectionReconnectListener.java | 8 +--
.../ral/updatable/LabelComputeNodeExecutor.java | 3 +-
.../ral/updatable/SetInstanceStatusExecutor.java | 5 +-
.../ral/updatable/UnlabelComputeNodeExecutor.java | 5 +-
17 files changed, 77 insertions(+), 165 deletions(-)
diff --git a/mode/api/src/main/java/org/apache/shardingsphere/mode/spi/PersistRepository.java b/mode/api/src/main/java/org/apache/shardingsphere/mode/spi/PersistRepository.java
index d77b3f6f118..b6d1c0cb1e0 100644
--- a/mode/api/src/main/java/org/apache/shardingsphere/mode/spi/PersistRepository.java
+++ b/mode/api/src/main/java/org/apache/shardingsphere/mode/spi/PersistRepository.java
@@ -63,6 +63,15 @@ public interface PersistRepository extends TypedSPI {
*/
void persist(String key, String value);
+ /**
+ * Persist ephemeral data.
+ *
+ * @param key key of data
+ * @param value value of data
+ */
+ default void persistEphemeral(String key, String value) {
+ }
+
/**
* Update data.
*
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/node/ComputeNodeStatusChangedEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/node/ComputeNodeStatusChangedEvent.java
deleted file mode 100644
index 158812bdc7c..00000000000
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/node/ComputeNodeStatusChangedEvent.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.mode.event.node;
-
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.state.instance.InstanceState;
-
-/**
- * Compute node status changed event.
- */
-@RequiredArgsConstructor
-@Getter
-public final class ComputeNodeStatusChangedEvent {
-
- private final String instanceId;
-
- private final InstanceState state;
-}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/manager/ContextManager.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/manager/ContextManager.java
index 26629fd2cb0..ee37c2a8568 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/manager/ContextManager.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/manager/ContextManager.java
@@ -44,6 +44,7 @@ import org.apache.shardingsphere.mode.manager.context.ShardingSphereDatabaseCont
import org.apache.shardingsphere.mode.manager.switcher.ResourceSwitchManager;
import org.apache.shardingsphere.mode.manager.switcher.SwitchingResource;
import org.apache.shardingsphere.mode.metadata.MetaDataContexts;
+import org.apache.shardingsphere.mode.service.ContextServiceFacade;
import org.apache.shardingsphere.mode.state.StateContext;
import org.apache.shardingsphere.mode.state.StateService;
@@ -74,6 +75,8 @@ public final class ContextManager implements AutoCloseable {
private final StateContext stateContext;
+ private final ContextServiceFacade contextServiceFacade;
+
public ContextManager(final MetaDataContexts metaDataContexts, final ComputeNodeInstanceContext computeNodeInstanceContext) {
this.metaDataContexts = new AtomicReference<>(metaDataContexts);
this.computeNodeInstanceContext = computeNodeInstanceContext;
@@ -82,6 +85,7 @@ public final class ContextManager implements AutoCloseable {
resourceMetaDataContextManager = new ResourceMetaDataContextManager(this.metaDataContexts);
executorEngine = ExecutorEngine.createExecutorEngineWithSize(metaDataContexts.getMetaData().getProps().<Integer>getValue(ConfigurationPropertyKey.KERNEL_EXECUTOR_SIZE));
stateContext = new StateContext(new StateService(metaDataContexts.getPersistService().getRepository()));
+ contextServiceFacade = new ContextServiceFacade(metaDataContexts.getPersistService().getRepository());
}
/**
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusService.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/service/ComputeNodeService.java
similarity index 92%
rename from mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusService.java
rename to mode/core/src/main/java/org/apache/shardingsphere/mode/service/ComputeNodeService.java
index 103a51126e8..cb71e190cc8 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusService.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/service/ComputeNodeService.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service;
+package org.apache.shardingsphere.mode.service;
import com.google.common.base.Strings;
import lombok.RequiredArgsConstructor;
@@ -30,7 +30,7 @@ import org.apache.shardingsphere.infra.instance.yaml.YamlComputeNodeDataSwapper;
import org.apache.shardingsphere.infra.state.instance.InstanceState;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.metadata.persist.node.ComputeNode;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.mode.spi.PersistRepository;
import java.util.Collection;
import java.util.LinkedHashSet;
@@ -42,9 +42,9 @@ import java.util.Optional;
*/
@RequiredArgsConstructor
@Slf4j
-public final class ComputeNodeStatusService {
+public final class ComputeNodeService {
- private final ClusterPersistRepository repository;
+ private final PersistRepository repository;
/**
* Register compute node online.
@@ -173,4 +173,14 @@ public final class ComputeNodeStatusService {
}
return result;
}
+
+ /**
+ * Update compute node state.
+ *
+ * @param instanceId instance id
+ * @param instanceState instance state
+ */
+ public void updateComputeNodeState(final String instanceId, final InstanceState instanceState) {
+ repository.persistEphemeral(ComputeNode.getInstanceStatusNodePath(instanceId), instanceState.name());
+ }
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/event/LabelsChangedEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/service/ContextServiceFacade.java
similarity index 69%
rename from mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/event/LabelsChangedEvent.java
rename to mode/core/src/main/java/org/apache/shardingsphere/mode/service/ContextServiceFacade.java
index 43293f9aa8c..f178744d028 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/event/LabelsChangedEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/service/ContextServiceFacade.java
@@ -15,21 +15,20 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event;
+package org.apache.shardingsphere.mode.service;
import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
-import java.util.Collection;
+import org.apache.shardingsphere.mode.spi.PersistRepository;
/**
- * Labels changed event.
+ * Context service facade.
*/
-@RequiredArgsConstructor
@Getter
-public final class LabelsChangedEvent {
+public final class ContextServiceFacade {
- private final String instanceId;
+ private final ComputeNodeService computeNodeService;
- private final Collection<String> labels;
+ public ContextServiceFacade(final PersistRepository repository) {
+ computeNodeService = new ComputeNodeService(repository);
+ }
}
diff --git a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusServiceTest.java b/mode/core/src/test/java/org/apache/shardingsphere/mode/service/ComputeNodeServiceTest.java
similarity index 82%
rename from mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusServiceTest.java
rename to mode/core/src/test/java/org/apache/shardingsphere/mode/service/ComputeNodeServiceTest.java
index 187338dac01..40a16f324fb 100644
--- a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/service/ComputeNodeStatusServiceTest.java
+++ b/mode/core/src/test/java/org/apache/shardingsphere/mode/service/ComputeNodeServiceTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service;
+package org.apache.shardingsphere.mode.service;
import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.infra.instance.metadata.InstanceMetaData;
@@ -26,7 +26,7 @@ import org.apache.shardingsphere.infra.instance.yaml.YamlComputeNodeData;
import org.apache.shardingsphere.infra.state.instance.InstanceState;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.metadata.persist.node.ComputeNode;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.mode.spi.PersistRepository;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
@@ -47,16 +47,16 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
-class ComputeNodeStatusServiceTest {
+class ComputeNodeServiceTest {
@Mock
- private ClusterPersistRepository repository;
+ private PersistRepository repository;
@Test
void assertRegisterOnline() {
ComputeNodeInstance computeNodeInstance = new ComputeNodeInstance(new ProxyInstanceMetaData("foo_instance_id", 3307));
computeNodeInstance.getLabels().add("test");
- new ComputeNodeStatusService(repository).registerOnline(computeNodeInstance);
+ new ComputeNodeService(repository).registerOnline(computeNodeInstance);
verify(repository).persistEphemeral(eq("/nodes/compute_nodes/online/proxy/" + computeNodeInstance.getMetaData().getId()), anyString());
verify(repository).persistEphemeral(ComputeNode.getInstanceStatusNodePath(computeNodeInstance.getMetaData().getId()), InstanceState.OK.name());
verify(repository).persistEphemeral(ComputeNode.getInstanceLabelsNodePath(computeNodeInstance.getMetaData().getId()), YamlEngine.marshal(Collections.singletonList("test")));
@@ -64,12 +64,12 @@ class ComputeNodeStatusServiceTest {
@Test
void assertPersistInstanceLabels() {
- ComputeNodeStatusService computeNodeStatusService = new ComputeNodeStatusService(repository);
+ ComputeNodeService computeNodeService = new ComputeNodeService(repository);
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
final String instanceId = instanceMetaData.getId();
- computeNodeStatusService.persistInstanceLabels(instanceId, Collections.singletonList("test"));
+ computeNodeService.persistInstanceLabels(instanceId, Collections.singletonList("test"));
verify(repository).persistEphemeral(ComputeNode.getInstanceLabelsNodePath(instanceId), YamlEngine.marshal(Collections.singletonList("test")));
- computeNodeStatusService.persistInstanceLabels(instanceId, Collections.emptyList());
+ computeNodeService.persistInstanceLabels(instanceId, Collections.emptyList());
verify(repository).persistEphemeral(ComputeNode.getInstanceLabelsNodePath(instanceId), YamlEngine.marshal(Collections.emptyList()));
}
@@ -77,7 +77,7 @@ class ComputeNodeStatusServiceTest {
void assertPersistInstanceWorkerId() {
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
final String instanceId = instanceMetaData.getId();
- new ComputeNodeStatusService(repository).persistInstanceWorkerId(instanceId, 100);
+ new ComputeNodeService(repository).persistInstanceWorkerId(instanceId, 100);
verify(repository).persistEphemeral(ComputeNode.getInstanceWorkerIdNodePath(instanceId), String.valueOf(100));
}
@@ -85,7 +85,7 @@ class ComputeNodeStatusServiceTest {
void assertLoadInstanceLabels() {
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
final String instanceId = instanceMetaData.getId();
- new ComputeNodeStatusService(repository).loadInstanceLabels(instanceId);
+ new ComputeNodeService(repository).loadInstanceLabels(instanceId);
verify(repository).getDirectly(ComputeNode.getInstanceLabelsNodePath(instanceId));
}
@@ -93,7 +93,7 @@ class ComputeNodeStatusServiceTest {
void assertLoadInstanceStatus() {
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
final String instanceId = instanceMetaData.getId();
- new ComputeNodeStatusService(repository).loadInstanceStatus(instanceId);
+ new ComputeNodeService(repository).loadInstanceStatus(instanceId);
verify(repository).getDirectly(ComputeNode.getInstanceStatusNodePath(instanceId));
}
@@ -101,7 +101,7 @@ class ComputeNodeStatusServiceTest {
void assertLoadInstanceWorkerId() {
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
final String instanceId = instanceMetaData.getId();
- new ComputeNodeStatusService(repository).loadInstanceWorkerId(instanceId);
+ new ComputeNodeService(repository).loadInstanceWorkerId(instanceId);
verify(repository).getDirectly(ComputeNode.getInstanceWorkerIdNodePath(instanceId));
}
@@ -117,7 +117,7 @@ class ComputeNodeStatusServiceTest {
yamlComputeNodeData1.setAttribute("127.0.0.1@3308");
yamlComputeNodeData1.setVersion("foo_version");
when(repository.getDirectly("/nodes/compute_nodes/online/proxy/foo_instance_3308")).thenReturn(YamlEngine.marshal(yamlComputeNodeData1));
- List<ComputeNodeInstance> actual = new ArrayList<>(new ComputeNodeStatusService(repository).loadAllComputeNodeInstances());
+ List<ComputeNodeInstance> actual = new ArrayList<>(new ComputeNodeService(repository).loadAllComputeNodeInstances());
assertThat(actual.size(), is(2));
assertThat(actual.get(0).getMetaData().getId(), is("foo_instance_3307"));
assertThat(actual.get(0).getMetaData().getIp(), is(IpUtils.getIp()));
@@ -130,13 +130,19 @@ class ComputeNodeStatusServiceTest {
@Test
void assertLoadComputeNodeInstance() {
InstanceMetaData instanceMetaData = new ProxyInstanceMetaData("foo_instance_id", 3307);
- ComputeNodeInstance actual = new ComputeNodeStatusService(repository).loadComputeNodeInstance(instanceMetaData);
+ ComputeNodeInstance actual = new ComputeNodeService(repository).loadComputeNodeInstance(instanceMetaData);
assertThat(actual.getMetaData(), is(instanceMetaData));
}
@Test
void assertGetUsedWorkerIds() {
- new ComputeNodeStatusService(repository).getAssignedWorkerIds();
+ new ComputeNodeService(repository).getAssignedWorkerIds();
verify(repository).getChildrenKeys(ComputeNode.getInstanceWorkerIdRootNodePath());
}
+
+ @Test
+ void assertUpdateComputeNodeState() {
+ new ComputeNodeService(repository).updateComputeNodeState("foo_instance_id", InstanceState.OK);
+ verify(repository).persistEphemeral(ComputeNode.getInstanceStatusNodePath("foo_instance_id"), InstanceState.OK.name());
+ }
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
index 89e189e6309..afe0ec0d9b9 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
@@ -37,8 +37,6 @@ import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.GlobalLoc
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.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;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.workerid.generator.ClusterWorkerIdGenerator;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.subscriber.ClusterEventSubscriberRegistry;
@@ -104,7 +102,6 @@ public final class ClusterContextManagerBuilder implements ContextManagerBuilder
// 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 QualifiedDataSourceStatusSubscriber(repository));
eventBusContext.register(new ClusterProcessSubscriber(repository, eventBusContext));
eventBusContext.register(new ShardingSphereSchemaDataRegistrySubscriber(repository));
@@ -112,14 +109,13 @@ public final class ClusterContextManagerBuilder implements ContextManagerBuilder
private void registerOnline(final EventBusContext eventBusContext, final ComputeNodeInstanceContext computeNodeInstanceContext,
final ClusterPersistRepository repository, final ContextManagerBuilderParameter param, final ContextManager contextManager) {
- ComputeNodeStatusService computeNodeStatusService = new ComputeNodeStatusService(repository);
- computeNodeStatusService.registerOnline(computeNodeInstanceContext.getInstance());
+ contextManager.getContextServiceFacade().getComputeNodeService().registerOnline(computeNodeInstanceContext.getInstance());
new GovernanceWatcherFactory(repository,
eventBusContext, param.getInstanceMetaData() instanceof JDBCInstanceMetaData ? param.getDatabaseConfigs().keySet() : Collections.emptyList()).watchListeners();
if (null != param.getLabels()) {
contextManager.getComputeNodeInstanceContext().getInstance().getLabels().addAll(param.getLabels());
}
- contextManager.getComputeNodeInstanceContext().getAllClusterInstances().addAll(computeNodeStatusService.loadAllComputeNodeInstances());
+ contextManager.getComputeNodeInstanceContext().getAllClusterInstances().addAll(contextManager.getContextServiceFacade().getComputeNodeService().loadAllComputeNodeInstances());
new ClusterEventSubscriberRegistry(contextManager, repository).register();
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/subscriber/ComputeNodeStatusSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/subscriber/ComputeNodeStatusSubscriber.java
deleted file mode 100644
index 8be1bd53afd..00000000000
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/subscriber/ComputeNodeStatusSubscriber.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.subscriber;
-
-import com.google.common.eventbus.Subscribe;
-import org.apache.shardingsphere.infra.util.eventbus.EventSubscriber;
-import org.apache.shardingsphere.metadata.persist.node.ComputeNode;
-import org.apache.shardingsphere.mode.event.node.ComputeNodeStatusChangedEvent;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.LabelsChangedEvent;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service.ComputeNodeStatusService;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
-
-/**
- * Compute node status subscriber.
- */
-public final class ComputeNodeStatusSubscriber implements EventSubscriber {
-
- private final ClusterPersistRepository repository;
-
- private final ComputeNodeStatusService computeNodeStatusService;
-
- public ComputeNodeStatusSubscriber(final ClusterPersistRepository repository) {
- this.repository = repository;
- computeNodeStatusService = new ComputeNodeStatusService(repository);
- }
-
- /**
- * Update compute node status.
- *
- * @param event compute node status changed event
- */
- @Subscribe
- public void update(final ComputeNodeStatusChangedEvent event) {
- repository.persistEphemeral(ComputeNode.getInstanceStatusNodePath(event.getInstanceId()), event.getState().name());
- }
-
- /**
- * Update compute node labels.
- *
- * @param event labels changed event
- */
- @Subscribe
- public void update(final LabelsChangedEvent event) {
- computeNodeStatusService.persistInstanceLabels(event.getInstanceId(), event.getLabels());
- }
-}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/workerid/generator/ClusterWorkerIdGenerator.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/workerid/generator/ClusterWorkerIdGenerator.java
index eb3e9ba8f3d..814b44e0505 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/workerid/generator/ClusterWorkerIdGenerator.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/workerid/generator/ClusterWorkerIdGenerator.java
@@ -22,10 +22,10 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.instance.workerid.WorkerIdAssignedException;
import org.apache.shardingsphere.infra.instance.workerid.WorkerIdGenerator;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service.ComputeNodeStatusService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.workerid.node.WorkerIdReservationNode;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import org.apache.shardingsphere.mode.repository.cluster.exception.ClusterPersistRepositoryException;
+import org.apache.shardingsphere.mode.service.ComputeNodeService;
import java.util.Collection;
import java.util.Optional;
@@ -45,14 +45,14 @@ public final class ClusterWorkerIdGenerator implements WorkerIdGenerator {
private final String instanceId;
- private final ComputeNodeStatusService computeNodeStatusService;
+ private final ComputeNodeService computeNodeService;
private final AtomicBoolean isWarned = new AtomicBoolean(false);
public ClusterWorkerIdGenerator(final ClusterPersistRepository repository, final String instanceId) {
this.repository = repository;
this.instanceId = instanceId;
- computeNodeStatusService = new ComputeNodeStatusService(repository);
+ computeNodeService = new ComputeNodeService(repository);
}
@Override
@@ -63,7 +63,7 @@ public final class ClusterWorkerIdGenerator implements WorkerIdGenerator {
}
private Optional<Integer> loadExistedWorkerId() {
- return computeNodeStatusService.loadInstanceWorkerId(instanceId);
+ return computeNodeService.loadInstanceWorkerId(instanceId);
}
private int generateNewWorkerId() {
@@ -72,12 +72,12 @@ public final class ClusterWorkerIdGenerator implements WorkerIdGenerator {
generatedWorkId = generateAvailableWorkerId();
} while (!generatedWorkId.isPresent());
int result = generatedWorkId.get();
- computeNodeStatusService.persistInstanceWorkerId(instanceId, result);
+ computeNodeService.persistInstanceWorkerId(instanceId, result);
return result;
}
private Optional<Integer> generateAvailableWorkerId() {
- Collection<Integer> assignedWorkerIds = computeNodeStatusService.getAssignedWorkerIds();
+ Collection<Integer> assignedWorkerIds = computeNodeService.getAssignedWorkerIds();
ShardingSpherePreconditions.checkState(assignedWorkerIds.size() <= MAX_WORKER_ID + 1, WorkerIdAssignedException::new);
PriorityQueue<Integer> availableWorkerIds = IntStream.range(0, 1024).boxed().filter(each -> !assignedWorkerIds.contains(each)).collect(Collectors.toCollection(PriorityQueue::new));
Integer preselectedWorkerId = availableWorkerIds.poll();
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ClusterEventSubscriberRegistry.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ClusterEventSubscriberRegistry.java
index 7c421b5de1c..5326fe989c4 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ClusterEventSubscriberRegistry.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ClusterEventSubscriberRegistry.java
@@ -34,7 +34,7 @@ public final class ClusterEventSubscriberRegistry extends EventSubscriberRegistr
new ConfigurationChangedSubscriber(contextManager),
new ConfigurationChangedSubscriber(contextManager),
new ResourceMetaDataChangedSubscriber(contextManager),
- new StateChangedSubscriber(contextManager, repository),
+ new StateChangedSubscriber(contextManager),
new DatabaseChangedSubscriber(contextManager),
new ProcessListChangedSubscriber(contextManager, repository),
new CacheEvictedSubscriber());
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
index 349461fc284..43759586e20 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
@@ -32,9 +32,7 @@ import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.statu
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.LabelsEvent;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ComputeNodeInstanceStateChangedEvent;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.WorkerIdEvent;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service.ComputeNodeStatusService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.storage.event.StorageNodeChangedEvent;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
/**
* State changed subscriber.
@@ -44,11 +42,8 @@ public final class StateChangedSubscriber implements EventSubscriber {
private final ContextManager contextManager;
- private final ComputeNodeStatusService computeNodeStatusService;
-
- public StateChangedSubscriber(final ContextManager contextManager, final ClusterPersistRepository repository) {
+ public StateChangedSubscriber(final ContextManager contextManager) {
this.contextManager = contextManager;
- computeNodeStatusService = new ComputeNodeStatusService(repository);
}
/**
@@ -117,7 +112,7 @@ public final class StateChangedSubscriber implements EventSubscriber {
*/
@Subscribe
public synchronized void renew(final InstanceOnlineEvent event) {
- contextManager.getComputeNodeInstanceContext().addComputeNodeInstance(computeNodeStatusService.loadComputeNodeInstance(event.getInstanceMetaData()));
+ contextManager.getComputeNodeInstanceContext().addComputeNodeInstance(contextManager.getContextServiceFacade().getComputeNodeService().loadComputeNodeInstance(event.getInstanceMetaData()));
}
/**
diff --git a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriberTest.java b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriberTest.java
index 1147704fb19..c344e148831 100644
--- a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriberTest.java
+++ b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriberTest.java
@@ -44,7 +44,6 @@ import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.statu
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.WorkerIdEvent;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.storage.event.StorageNodeChangedEvent;
import org.apache.shardingsphere.mode.metadata.MetaDataContexts;
-import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepositoryConfiguration;
import org.apache.shardingsphere.mode.storage.QualifiedDataSourceStatus;
import org.junit.jupiter.api.BeforeEach;
@@ -90,7 +89,7 @@ class StateChangedSubscriberTest {
contextManager.renewMetaDataContexts(new MetaDataContexts(contextManager.getMetaDataContexts().getPersistService(), new ShardingSphereMetaData(createDatabases(),
contextManager.getMetaDataContexts().getMetaData().getGlobalResourceMetaData(), contextManager.getMetaDataContexts().getMetaData().getGlobalRuleMetaData(),
new ConfigurationProperties(new Properties()))));
- subscriber = new StateChangedSubscriber(contextManager, mock(ClusterPersistRepository.class));
+ subscriber = new StateChangedSubscriber(contextManager);
}
private ContextManagerBuilderParameter createContextManagerBuilderParameter() {
diff --git a/mode/type/cluster/repository/api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java b/mode/type/cluster/repository/api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java
index 6554332ad04..b240ecea27b 100644
--- a/mode/type/cluster/repository/api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java
+++ b/mode/type/cluster/repository/api/src/main/java/org/apache/shardingsphere/mode/repository/cluster/ClusterPersistRepository.java
@@ -33,14 +33,6 @@ public interface ClusterPersistRepository extends PersistRepository {
*/
void init(ClusterPersistRepositoryConfiguration config);
- /**
- * Persist ephemeral data.
- *
- * @param key key of data
- * @param value value of data
- */
- void persistEphemeral(String key, String value);
-
/**
* Persist exclusive ephemeral data.
*
diff --git a/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/listener/SessionConnectionReconnectListener.java b/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/listener/SessionConnectionReconnectListener.java
index 1d1742be85b..00ace637d52 100644
--- a/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/listener/SessionConnectionReconnectListener.java
+++ b/mode/type/cluster/repository/provider/zookeeper/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/listener/SessionConnectionReconnectListener.java
@@ -23,8 +23,8 @@ import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.shardingsphere.infra.instance.ComputeNodeInstanceContext;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.service.ComputeNodeStatusService;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+import org.apache.shardingsphere.mode.service.ComputeNodeService;
import java.util.Properties;
@@ -38,11 +38,11 @@ public final class SessionConnectionReconnectListener implements ConnectionState
private final ComputeNodeInstanceContext computeNodeInstanceContext;
- private final ComputeNodeStatusService computeNodeStatusService;
+ private final ComputeNodeService computeNodeService;
public SessionConnectionReconnectListener(final ComputeNodeInstanceContext computeNodeInstanceContext, final ClusterPersistRepository repository) {
this.computeNodeInstanceContext = computeNodeInstanceContext;
- computeNodeStatusService = new ComputeNodeStatusService(repository);
+ this.computeNodeService = new ComputeNodeService(repository);
}
@Override
@@ -63,7 +63,7 @@ public final class SessionConnectionReconnectListener implements ConnectionState
if (isNeedGenerateWorkerId()) {
computeNodeInstanceContext.generateWorkerId(new Properties());
}
- computeNodeStatusService.registerOnline(computeNodeInstanceContext.getInstance());
+ computeNodeService.registerOnline(computeNodeInstanceContext.getInstance());
return true;
}
sleepInterval();
diff --git a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/LabelComputeNodeExecutor.java b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/LabelComputeNodeExecutor.java
index 86f60b7800f..6538f1a4d10 100644
--- a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/LabelComputeNodeExecutor.java
+++ b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/LabelComputeNodeExecutor.java
@@ -22,7 +22,6 @@ import org.apache.shardingsphere.distsql.handler.engine.update.DistSQLUpdateExec
import org.apache.shardingsphere.distsql.statement.ral.updatable.LabelComputeNodeStatement;
import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.mode.manager.ContextManager;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.LabelsChangedEvent;
import java.sql.SQLException;
import java.util.Collection;
@@ -45,7 +44,7 @@ public final class LabelComputeNodeExecutor implements DistSQLUpdateExecutor<Lab
if (!sqlStatement.isOverwrite()) {
labels.addAll(computeNodeInstance.get().getLabels());
}
- contextManager.getComputeNodeInstanceContext().getEventBusContext().post(new LabelsChangedEvent(instanceId, new LinkedList<>(labels)));
+ contextManager.getContextServiceFacade().getComputeNodeService().persistInstanceLabels(instanceId, new LinkedList<>(labels));
}
}
diff --git a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/SetInstanceStatusExecutor.java b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/SetInstanceStatusExecutor.java
index 6c44c7db823..272b30b5ea8 100644
--- a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/SetInstanceStatusExecutor.java
+++ b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/SetInstanceStatusExecutor.java
@@ -23,7 +23,6 @@ import org.apache.shardingsphere.distsql.statement.ral.updatable.SetInstanceStat
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import org.apache.shardingsphere.infra.state.instance.InstanceState;
-import org.apache.shardingsphere.mode.event.node.ComputeNodeStatusChangedEvent;
import org.apache.shardingsphere.mode.manager.ContextManager;
/**
@@ -39,8 +38,8 @@ public final class SetInstanceStatusExecutor implements DistSQLUpdateExecutor<Se
} else {
checkEnablingIsValid(contextManager, sqlStatement.getInstanceId());
}
- contextManager.getComputeNodeInstanceContext().getEventBusContext().post(
- new ComputeNodeStatusChangedEvent(sqlStatement.getInstanceId(), "DISABLE".equals(sqlStatement.getStatus()) ? InstanceState.CIRCUIT_BREAK : InstanceState.OK));
+ contextManager.getContextServiceFacade().getComputeNodeService().updateComputeNodeState(sqlStatement.getInstanceId(),
+ "DISABLE".equals(sqlStatement.getStatus()) ? InstanceState.CIRCUIT_BREAK : InstanceState.OK);
}
private void checkEnablingIsValid(final ContextManager contextManager, final String instanceId) {
diff --git a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/UnlabelComputeNodeExecutor.java b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/UnlabelComputeNodeExecutor.java
index 7bb1dd1400a..4948c0b9ac5 100644
--- a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/UnlabelComputeNodeExecutor.java
+++ b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/UnlabelComputeNodeExecutor.java
@@ -22,7 +22,6 @@ import org.apache.shardingsphere.distsql.handler.engine.update.DistSQLUpdateExec
import org.apache.shardingsphere.distsql.statement.ral.updatable.UnlabelComputeNodeStatement;
import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.mode.manager.ContextManager;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.LabelsChangedEvent;
import java.util.ArrayList;
import java.util.Collection;
@@ -43,10 +42,10 @@ public final class UnlabelComputeNodeExecutor implements DistSQLUpdateExecutor<U
if (computeNodeInstance.isPresent()) {
Collection<String> labels = new LinkedHashSet<>(computeNodeInstance.get().getLabels());
if (sqlStatement.getLabels().isEmpty()) {
- contextManager.getComputeNodeInstanceContext().getEventBusContext().post(new LabelsChangedEvent(instanceId, Collections.emptyList()));
+ contextManager.getContextServiceFacade().getComputeNodeService().persistInstanceLabels(instanceId, Collections.emptyList());
} else {
labels.removeAll(sqlStatement.getLabels());
- contextManager.getComputeNodeInstanceContext().getEventBusContext().post(new LabelsChangedEvent(instanceId, new ArrayList<>(labels)));
+ contextManager.getContextServiceFacade().getComputeNodeService().persistInstanceLabels(instanceId, new ArrayList<>(labels));
}
}
}