You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2022/03/01 06:19:28 UTC
[shardingsphere] branch master updated: Refactor curator cache for curator zookeeper repository (#15672)
This is an automated email from the ASF dual-hosted git repository.
panjuan 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 8f012fe Refactor curator cache for curator zookeeper repository (#15672)
8f012fe is described below
commit 8f012fe4adff996fd796feec038d3837d4a704cc
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Tue Mar 1 14:18:35 2022 +0800
Refactor curator cache for curator zookeeper repository (#15672)
---
.../registry/GovernanceWatcherFactory.java | 22 +++---------
.../watcher/PrivilegeNodeChangedWatcher.java | 3 +-
.../config/watcher/GlobalRuleChangedWatcher.java | 2 +-
.../config/watcher/PropertiesChangedWatcher.java | 2 +-
.../watcher/PrivilegeNodeChangedWatcherTest.java | 2 +-
.../watcher/GlobalRuleChangedWatcherTest.java | 2 +-
.../watcher/PropertiesChangedWatcherTest.java | 4 +--
.../zookeeper/CuratorZookeeperRepository.java | 39 ++++++----------------
.../zookeeper/CuratorZookeeperRepositoryTest.java | 9 ++---
9 files changed, 26 insertions(+), 59 deletions(-)
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/GovernanceWatcherFactory.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/GovernanceWatcherFactory.java
index 24b148a..2128c63 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/GovernanceWatcherFactory.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/GovernanceWatcherFactory.java
@@ -22,6 +22,8 @@ import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import org.apache.shardingsphere.spi.ShardingSphereServiceLoader;
+import java.util.Collection;
+
/**
* Governance watcher factory.
*/
@@ -38,22 +40,8 @@ public final class GovernanceWatcherFactory {
* Watch listeners.
*/
public void watchListeners() {
- for (GovernanceWatcher<?> each : ShardingSphereServiceLoader.getSingletonServiceInstances(GovernanceWatcher.class)) {
- watch(each);
- }
- }
-
- private void watch(final GovernanceWatcher<?> listener) {
- for (String each : listener.getWatchingKeys()) {
- watch(each, listener);
- }
- }
-
- private void watch(final String watchingKey, final GovernanceWatcher<?> listener) {
- repository.watch(watchingKey, dataChangedEventListener -> {
- if (listener.getWatchingTypes().contains(dataChangedEventListener.getType())) {
- listener.createGovernanceEvent(dataChangedEventListener).ifPresent(ShardingSphereEventBus.getInstance()::post);
- }
- });
+ Collection<GovernanceWatcher> governanceWatchers = ShardingSphereServiceLoader.getSingletonServiceInstances(GovernanceWatcher.class);
+ repository.watch("/", dataChangedEventListener -> governanceWatchers.stream().filter(each -> each.getWatchingTypes().contains(dataChangedEventListener.getType()))
+ .forEach(each -> each.createGovernanceEvent(dataChangedEventListener).ifPresent(ShardingSphereEventBus.getInstance()::post)));
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcher.java
index ee184d6..87785d2 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcher.java
@@ -46,6 +46,7 @@ public final class PrivilegeNodeChangedWatcher implements GovernanceWatcher<Auth
@Override
public Optional<AuthorityChangedEvent> createGovernanceEvent(final DataChangedEvent event) {
- return Optional.of(new AuthorityChangedEvent(YamlUsersConfigurationConverter.convertShardingSphereUser(YamlEngine.unmarshal(event.getValue(), Collection.class))));
+ return getWatchingKeys().contains(event.getKey()) ? Optional.of(new AuthorityChangedEvent(YamlUsersConfigurationConverter
+ .convertShardingSphereUser(YamlEngine.unmarshal(event.getValue(), Collection.class)))) : Optional.empty();
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcher.java
index 46b067c..c81649b 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcher.java
@@ -49,7 +49,7 @@ public final class GlobalRuleChangedWatcher implements GovernanceWatcher<GlobalR
@Override
public Optional<GlobalRuleConfigurationsChangedEvent> createGovernanceEvent(final DataChangedEvent event) {
- return Optional.of(new GlobalRuleConfigurationsChangedEvent(getGlobalRuleConfigurations(event)));
+ return getWatchingKeys().contains(event.getKey()) ? Optional.of(new GlobalRuleConfigurationsChangedEvent(getGlobalRuleConfigurations(event))) : Optional.empty();
}
@SuppressWarnings("unchecked")
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcher.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcher.java
index 2a41e73..831707f 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcher.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcher.java
@@ -47,6 +47,6 @@ public final class PropertiesChangedWatcher implements GovernanceWatcher<Propert
@Override
public Optional<PropertiesChangedEvent> createGovernanceEvent(final DataChangedEvent event) {
- return Optional.of(new PropertiesChangedEvent(YamlEngine.unmarshal(event.getValue(), Properties.class)));
+ return getWatchingKeys().contains(event.getKey()) ? Optional.of(new PropertiesChangedEvent(YamlEngine.unmarshal(event.getValue(), Properties.class))) : Optional.empty();
}
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcherTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcherTest.java
index 5e83e2e..9ff9a8e 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcherTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/authority/watcher/PrivilegeNodeChangedWatcherTest.java
@@ -36,7 +36,7 @@ public final class PrivilegeNodeChangedWatcherTest {
@Test
public void assertCreateEvent() {
- Optional<AuthorityChangedEvent> actual = new PrivilegeNodeChangedWatcher().createGovernanceEvent(new DataChangedEvent("test", AUTHENTICATION_YAML, Type.UPDATED));
+ Optional<AuthorityChangedEvent> actual = new PrivilegeNodeChangedWatcher().createGovernanceEvent(new DataChangedEvent("/nodes/privilege_nodes", AUTHENTICATION_YAML, Type.UPDATED));
assertTrue(actual.isPresent());
Optional<ShardingSphereUser> user = actual.get().getUsers().stream().filter(each -> each.getGrantee().equals(new Grantee("root1", ""))).findFirst();
assertTrue(user.isPresent());
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcherTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcherTest.java
index 1fbd4e0..0a5bc18 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcherTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/GlobalRuleChangedWatcherTest.java
@@ -38,7 +38,7 @@ public final class GlobalRuleChangedWatcherTest {
@Test
public void assertCreateEvent() {
- Optional<GlobalRuleConfigurationsChangedEvent> event = new GlobalRuleChangedWatcher().createGovernanceEvent(new DataChangedEvent("rule", readYAML("yaml/authority-rule.yaml"), Type.UPDATED));
+ Optional<GlobalRuleConfigurationsChangedEvent> event = new GlobalRuleChangedWatcher().createGovernanceEvent(new DataChangedEvent("/rules", readYAML("yaml/authority-rule.yaml"), Type.UPDATED));
assertTrue(event.isPresent());
assertThat(event.get(), instanceOf(GlobalRuleConfigurationsChangedEvent.class));
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcherTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcherTest.java
index 04a1252..dbdf768 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcherTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/config/watcher/PropertiesChangedWatcherTest.java
@@ -33,14 +33,14 @@ public final class PropertiesChangedWatcherTest {
@Test
public void assertCreateEvent() {
- Optional<PropertiesChangedEvent> actual = new PropertiesChangedWatcher().createGovernanceEvent(new DataChangedEvent("test", PROPERTIES_YAML, Type.UPDATED));
+ Optional<PropertiesChangedEvent> actual = new PropertiesChangedWatcher().createGovernanceEvent(new DataChangedEvent("/props", PROPERTIES_YAML, Type.UPDATED));
assertTrue(actual.isPresent());
assertTrue((boolean) actual.get().getProps().get(ConfigurationPropertyKey.SQL_SHOW.getKey()));
}
@Test
public void assertCreateAddEvent() {
- Optional<PropertiesChangedEvent> actual = new PropertiesChangedWatcher().createGovernanceEvent(new DataChangedEvent("test", PROPERTIES_YAML, Type.ADDED));
+ Optional<PropertiesChangedEvent> actual = new PropertiesChangedWatcher().createGovernanceEvent(new DataChangedEvent("/props", PROPERTIES_YAML, Type.ADDED));
assertTrue(actual.isPresent());
assertTrue((boolean) actual.get().getProps().get(ConfigurationPropertyKey.SQL_SHOW.getKey()));
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-prov [...]
index 6a948ba..69d0b7e 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/main/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepository.java
@@ -48,10 +48,8 @@ import org.apache.zookeeper.data.ACL;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Comparator;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
@@ -63,7 +61,7 @@ import java.util.concurrent.TimeUnit;
*/
public final class CuratorZookeeperRepository implements ClusterPersistRepository {
- private final Map<String, CuratorCache> caches = new HashMap<>();
+ private CuratorCache curatorCache;
private CuratorFramework client;
@@ -131,21 +129,13 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
@Override
public String get(final String key) {
- CuratorCache cache = findTreeCache(key);
- if (null == cache) {
- return getDirectly(key);
- }
- Optional<ChildData> resultInCache = cache.get(key);
- if (resultInCache.isPresent()) {
+ if (Optional.ofNullable(curatorCache).isPresent() && curatorCache.get(key).isPresent()) {
+ Optional<ChildData> resultInCache = curatorCache.get(key);
return null == resultInCache.get().getData() ? null : new String(resultInCache.get().getData(), StandardCharsets.UTF_8);
}
return getDirectly(key);
}
- private CuratorCache findTreeCache(final String key) {
- return caches.entrySet().stream().filter(entry -> key.startsWith(entry.getKey())).findFirst().map(Entry::getValue).orElse(null);
- }
-
@Override
public List<String> getChildrenKeys(final String key) {
try {
@@ -249,21 +239,19 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
@Override
public void watch(final String key, final DataChangedEventListener listener) {
- String path = key + PATH_SEPARATOR;
- if (!caches.containsKey(path)) {
- addCacheData(key);
- CuratorCache cache = caches.get(path);
- CuratorCacheListener curatorCacheListener = CuratorCacheListener.builder()
+ if (!Optional.ofNullable(curatorCache).isPresent()) {
+ curatorCache = CuratorCache.build(client, "/");
+ }
+ CuratorCacheListener curatorCacheListener = CuratorCacheListener.builder()
.forTreeCache(client, (framework, treeCacheListener) -> {
Type changedType = getChangedType(treeCacheListener.getType());
if (Type.IGNORED != changedType) {
- listener.onChange(new DataChangedEvent(treeCacheListener.getData().getPath(),
+ listener.onChange(new DataChangedEvent(treeCacheListener.getData().getPath(),
new String(treeCacheListener.getData().getData(), StandardCharsets.UTF_8), changedType));
}
}).build();
- cache.listenable().addListener(curatorCacheListener);
- start(cache);
- }
+ curatorCache.listenable().addListener(curatorCacheListener);
+ start(curatorCache);
}
private void start(final CuratorCache cache) {
@@ -276,11 +264,6 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
}
}
- private void addCacheData(final String cachePath) {
- CuratorCache cache = CuratorCache.build(client, cachePath);
- caches.put(cachePath + PATH_SEPARATOR, cache);
- }
-
private Type getChangedType(final TreeCacheEvent.Type type) {
switch (type) {
case NODE_ADDED:
@@ -334,7 +317,7 @@ public final class CuratorZookeeperRepository implements ClusterPersistRepositor
@Override
public void close() {
- caches.values().forEach(CuratorCache::close);
+ curatorCache.close();
waitForCacheClose();
CloseableUtils.closeQuietly(client);
}
diff --git a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository- [...]
index 864daba..3dcd998 100644
--- a/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java
+++ b/shardingsphere-mode/shardingsphere-mode-type/shardingsphere-cluster-mode/shardingsphere-cluster-mode-repository/shardingsphere-cluster-mode-repository-provider/shardingsphere-cluster-mode-repository-zookeeper-curator/src/test/java/org/apache/shardingsphere/mode/repository/cluster/zookeeper/CuratorZookeeperRepositoryTest.java
@@ -54,7 +54,6 @@ import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
@@ -80,9 +79,6 @@ public final class CuratorZookeeperRepositoryTest {
private static final String SERVER_LISTS = "127.0.0.1:2181";
@Mock
- private Map<String, CuratorCache> caches;
-
- @Mock
private CuratorCache curatorCache;
@Mock
@@ -251,10 +247,9 @@ public final class CuratorZookeeperRepositoryTest {
}
private void mockCache() throws Exception {
- Field cachesFiled = CuratorZookeeperRepository.class.getDeclaredField("caches");
+ Field cachesFiled = CuratorZookeeperRepository.class.getDeclaredField("curatorCache");
cachesFiled.setAccessible(true);
- cachesFiled.set(REPOSITORY, caches);
- when(caches.get(anyString())).thenReturn(curatorCache);
+ cachesFiled.set(REPOSITORY, curatorCache);
when(curatorCache.listenable()).thenReturn(listenable);
}