You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by su...@apache.org on 2023/06/20 18:13:25 UTC
[shardingsphere] branch master updated: Refactor readwrite-splitting event build (#26472)
This is an automated email from the ASF dual-hosted git repository.
sunnianjun 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 0cecf62b4bd Refactor readwrite-splitting event build (#26472)
0cecf62b4bd is described below
commit 0cecf62b4bddbed012d985f42d0bdc1618640a56
Author: zhaojinchao <zh...@apache.org>
AuthorDate: Wed Jun 21 02:13:17 2023 +0800
Refactor readwrite-splitting event build (#26472)
* Refactor readwrite-splitting event build
* Fix checkstyle
---
...riteSplittingRuleConfigurationEventBuilder.java | 53 ++++++----------------
.../AddReadwriteSplittingConfigurationEvent.java | 3 +-
.../AlterReadwriteSplittingConfigurationEvent.java | 3 +-
...DeleteReadwriteSplittingConfigurationEvent.java | 4 --
.../event/loadbalance/AlterLoadBalanceEvent.java | 3 --
.../converter/ReadwriteSplittingNodeConverter.java | 30 ++++++------
.../ReadwriteSplittingConfigurationSubscriber.java | 26 +++++++++--
.../ReadwriteSplittingLoadBalanceSubscriber.java | 11 ++++-
.../ReadwriteSplittingNodeConverterTest.java | 12 ++---
.../infra/instance/mode/ModeContextManager.java | 13 ++++++
.../cluster/NewClusterModeContextManager.java | 6 +++
11 files changed, 86 insertions(+), 78 deletions(-)
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/ReadwriteSplittingRuleConfigurationEventBuilder.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/ReadwriteSplittingRuleConfigurationEventBuilder.java
index c3435ee8e0e..ce11b68c31e 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/ReadwriteSplittingRuleConfigurationEventBuilder.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/ReadwriteSplittingRuleConfigurationEventBuilder.java
@@ -18,23 +18,16 @@
package org.apache.shardingsphere.readwritesplitting.event;
import com.google.common.base.Strings;
-import org.apache.shardingsphere.infra.config.algorithm.AlgorithmConfiguration;
import org.apache.shardingsphere.infra.rule.event.GovernanceEvent;
-import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
-import org.apache.shardingsphere.infra.yaml.config.pojo.algorithm.YamlAlgorithmConfiguration;
-import org.apache.shardingsphere.infra.yaml.config.swapper.algorithm.YamlAlgorithmConfigurationSwapper;
import org.apache.shardingsphere.mode.event.DataChangedEvent;
import org.apache.shardingsphere.mode.event.DataChangedEvent.Type;
import org.apache.shardingsphere.mode.spi.RuleConfigurationEventBuilder;
-import org.apache.shardingsphere.readwritesplitting.api.rule.ReadwriteSplittingDataSourceRuleConfiguration;
-import org.apache.shardingsphere.readwritesplitting.api.transaction.TransactionalReadQueryStrategy;
import org.apache.shardingsphere.readwritesplitting.event.config.AddReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.event.config.AlterReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.event.config.DeleteReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.AlterLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.DeleteLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.metadata.converter.ReadwriteSplittingNodeConverter;
-import org.apache.shardingsphere.readwritesplitting.yaml.config.rule.YamlReadwriteSplittingDataSourceRuleConfiguration;
import java.util.Optional;
@@ -48,51 +41,31 @@ public final class ReadwriteSplittingRuleConfigurationEventBuilder implements Ru
if (!ReadwriteSplittingNodeConverter.isReadwriteSplittingPath(event.getKey()) || Strings.isNullOrEmpty(event.getValue())) {
return Optional.empty();
}
- if (ReadwriteSplittingNodeConverter.isDataSourcePath(event.getKey())) {
- Optional<String> groupNameVersion = ReadwriteSplittingNodeConverter.getGroupNameVersion(event.getKey());
- if (groupNameVersion.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
- return createReadwriteSplittingConfigEvent(databaseName, groupNameVersion.get(), event);
- }
+ Optional<String> groupName = ReadwriteSplittingNodeConverter.getGroupNameByActiveVersionPath(event.getKey());
+ if (groupName.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
+ return createReadwriteSplittingConfigEvent(databaseName, groupName.get(), event);
}
- Optional<String> loadBalancerNameVersion = ReadwriteSplittingNodeConverter.getLoadBalancerNameVersion(event.getKey());
- if (loadBalancerNameVersion.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
- return createLoadBalanceEvent(databaseName, loadBalancerNameVersion.get(), event);
+ Optional<String> loadBalancerName = ReadwriteSplittingNodeConverter.getLoadBalancerNameByActiveVersionPath(event.getKey());
+ if (loadBalancerName.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
+ return createLoadBalanceEvent(databaseName, loadBalancerName.get(), event);
}
return Optional.empty();
}
- private Optional<GovernanceEvent> createReadwriteSplittingConfigEvent(final String databaseName, final String version, final DataChangedEvent event) {
- String groupName = ReadwriteSplittingNodeConverter.getGroupName(event.getKey()).orElse("");
+ private Optional<GovernanceEvent> createReadwriteSplittingConfigEvent(final String databaseName, final String groupName, final DataChangedEvent event) {
if (Type.ADDED == event.getType()) {
- return Optional.of(new AddReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), event.getKey(), version));
+ return Optional.of(new AddReadwriteSplittingConfigurationEvent(databaseName, groupName, event.getKey(), event.getValue()));
}
if (Type.UPDATED == event.getType()) {
- return Optional.of(new AlterReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), event.getKey(), version));
+ return Optional.of(new AlterReadwriteSplittingConfigurationEvent(databaseName, groupName, event.getKey(), event.getValue()));
}
- return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName, event.getKey(), version));
+ return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName));
}
- private ReadwriteSplittingDataSourceRuleConfiguration swapDataSource(final String name, final String yamlContext) {
- YamlReadwriteSplittingDataSourceRuleConfiguration yamlDataSourceRuleConfig = YamlEngine.unmarshal(yamlContext, YamlReadwriteSplittingDataSourceRuleConfiguration.class);
- return new ReadwriteSplittingDataSourceRuleConfiguration(name, yamlDataSourceRuleConfig.getWriteDataSourceName(), yamlDataSourceRuleConfig.getReadDataSourceNames(),
- getTransactionalReadQueryStrategy(yamlDataSourceRuleConfig), yamlDataSourceRuleConfig.getLoadBalancerName());
- }
-
- private TransactionalReadQueryStrategy getTransactionalReadQueryStrategy(final YamlReadwriteSplittingDataSourceRuleConfiguration yamlDataSourceRuleConfig) {
- return Strings.isNullOrEmpty(yamlDataSourceRuleConfig.getTransactionalReadQueryStrategy())
- ? TransactionalReadQueryStrategy.DYNAMIC
- : TransactionalReadQueryStrategy.valueOf(yamlDataSourceRuleConfig.getTransactionalReadQueryStrategy());
- }
-
- private Optional<GovernanceEvent> createLoadBalanceEvent(final String databaseName, final String version, final DataChangedEvent event) {
- String loadBalanceName = ReadwriteSplittingNodeConverter.getLoadBalancerName(event.getKey()).orElse("");
+ private Optional<GovernanceEvent> createLoadBalanceEvent(final String databaseName, final String loadBalancerName, final DataChangedEvent event) {
if (Type.ADDED == event.getType() || Type.UPDATED == event.getType()) {
- return Optional.of(new AlterLoadBalanceEvent(databaseName, loadBalanceName, swapToAlgorithmConfig(event.getValue()), event.getKey(), version));
+ return Optional.of(new AlterLoadBalanceEvent(databaseName, loadBalancerName, event.getKey(), event.getValue()));
}
- return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalanceName, event.getKey(), version));
- }
-
- private AlgorithmConfiguration swapToAlgorithmConfig(final String yamlContext) {
- return new YamlAlgorithmConfigurationSwapper().swapToObject(YamlEngine.unmarshal(yamlContext, YamlAlgorithmConfiguration.class));
+ return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalancerName, event.getKey(), event.getValue()));
}
}
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AddReadwriteSplittingConfigurationEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AddReadwriteSplittingConfigurationEvent.java
index e5a717f0508..bd17e75f93e 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AddReadwriteSplittingConfigurationEvent.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AddReadwriteSplittingConfigurationEvent.java
@@ -20,7 +20,6 @@ package org.apache.shardingsphere.readwritesplitting.event.config;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.rule.event.GovernanceEvent;
-import org.apache.shardingsphere.readwritesplitting.api.rule.ReadwriteSplittingDataSourceRuleConfiguration;
/**
* Add readwrite-splitting configuration event.
@@ -31,7 +30,7 @@ public final class AddReadwriteSplittingConfigurationEvent implements Governance
private final String databaseName;
- private final ReadwriteSplittingDataSourceRuleConfiguration config;
+ private final String groupName;
private final String activeVersionKey;
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AlterReadwriteSplittingConfigurationEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AlterReadwriteSplittingConfigurationEvent.java
index 0bd0d423196..094d7209cc2 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AlterReadwriteSplittingConfigurationEvent.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/AlterReadwriteSplittingConfigurationEvent.java
@@ -20,7 +20,6 @@ package org.apache.shardingsphere.readwritesplitting.event.config;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.rule.event.GovernanceEvent;
-import org.apache.shardingsphere.readwritesplitting.api.rule.ReadwriteSplittingDataSourceRuleConfiguration;
/**
* Alter readwrite-splitting configuration event.
@@ -31,7 +30,7 @@ public final class AlterReadwriteSplittingConfigurationEvent implements Governan
private final String databaseName;
- private final ReadwriteSplittingDataSourceRuleConfiguration config;
+ private final String groupName;
private final String activeVersionKey;
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/DeleteReadwriteSplittingConfigurationEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/DeleteReadwriteSplittingConfigurationEvent.java
index 0b74264c768..02ef1fe1e19 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/DeleteReadwriteSplittingConfigurationEvent.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/config/DeleteReadwriteSplittingConfigurationEvent.java
@@ -31,8 +31,4 @@ public final class DeleteReadwriteSplittingConfigurationEvent implements Governa
private final String databaseName;
private final String groupName;
-
- private final String activeVersionKey;
-
- private final String activeVersion;
}
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AlterLoadBalanceEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AlterLoadBalanceEvent.java
index 719cd2a80eb..d6904ccf7d4 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AlterLoadBalanceEvent.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AlterLoadBalanceEvent.java
@@ -19,7 +19,6 @@ package org.apache.shardingsphere.readwritesplitting.event.loadbalance;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.config.algorithm.AlgorithmConfiguration;
import org.apache.shardingsphere.infra.rule.event.GovernanceEvent;
/**
@@ -33,8 +32,6 @@ public final class AlterLoadBalanceEvent implements GovernanceEvent {
private final String loadBalanceName;
- private final AlgorithmConfiguration config;
-
private final String activeVersionKey;
private final String activeVersion;
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverter.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverter.java
index 9a0d7fbb573..e842aa49b26 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverter.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverter.java
@@ -40,7 +40,7 @@ public final class ReadwriteSplittingNodeConverter {
private static final String RULE_NAME_PATTERN = "/([\\w\\-]+)/versions/?";
- private static final String RULE_VERSION = "/([\\w\\-]+)/versions/([\\w\\-]+)$";
+ private static final String RULE_ACTIVE_VERSION = "/([\\w\\-]+)/active_version$";
/**
* Get group name path.
@@ -111,27 +111,27 @@ public final class ReadwriteSplittingNodeConverter {
}
/**
- * Get group name version.
+ * Get group name by active version path.
*
- * @param rulePath rule path
- * @return group name version
+ * @param activeVersionPath active version path
+ * @return group name
*/
- public static Optional<String> getGroupNameVersion(final String rulePath) {
- Pattern pattern = Pattern.compile(RULES_NODE_PREFIX + ROOT_NODE + "/" + DATA_SOURCES_NODE + RULE_VERSION, Pattern.CASE_INSENSITIVE);
- Matcher matcher = pattern.matcher(rulePath);
- return matcher.find() ? Optional.of(matcher.group(4)) : Optional.empty();
+ public static Optional<String> getGroupNameByActiveVersionPath(final String activeVersionPath) {
+ Pattern pattern = Pattern.compile(RULES_NODE_PREFIX + ROOT_NODE + "/" + DATA_SOURCES_NODE + RULE_ACTIVE_VERSION, Pattern.CASE_INSENSITIVE);
+ Matcher matcher = pattern.matcher(activeVersionPath);
+ return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
/**
- * Get load balancer name version.
+ * Get load balancer name by active version path.
*
- * @param rulePath rule path
- * @return load balancer name version
+ * @param activeVersionPath active version path
+ * @return load balancer name
*/
- public static Optional<String> getLoadBalancerNameVersion(final String rulePath) {
- Pattern pattern = Pattern.compile(RULES_NODE_PREFIX + ROOT_NODE + "/" + LOAD_BALANCER_NODE + RULE_VERSION, Pattern.CASE_INSENSITIVE);
- Matcher matcher = pattern.matcher(rulePath);
- return matcher.find() ? Optional.of(matcher.group(4)) : Optional.empty();
+ public static Optional<String> getLoadBalancerNameByActiveVersionPath(final String activeVersionPath) {
+ Pattern pattern = Pattern.compile(RULES_NODE_PREFIX + ROOT_NODE + "/" + LOAD_BALANCER_NODE + RULE_ACTIVE_VERSION, Pattern.CASE_INSENSITIVE);
+ Matcher matcher = pattern.matcher(activeVersionPath);
+ return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
/**
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingConfigurationSubscriber.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingConfigurationSubscriber.java
index 1522de32f36..d00c6941cf3 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingConfigurationSubscriber.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingConfigurationSubscriber.java
@@ -17,18 +17,22 @@
package org.apache.shardingsphere.readwritesplitting.subscriber;
+import com.google.common.base.Strings;
import com.google.common.eventbus.Subscribe;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.instance.InstanceContext;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.rule.RuleConfigurationSubscribeCoordinator;
+import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.mode.event.config.DatabaseRuleConfigurationChangedEvent;
import org.apache.shardingsphere.readwritesplitting.api.ReadwriteSplittingRuleConfiguration;
import org.apache.shardingsphere.readwritesplitting.api.rule.ReadwriteSplittingDataSourceRuleConfiguration;
+import org.apache.shardingsphere.readwritesplitting.api.transaction.TransactionalReadQueryStrategy;
import org.apache.shardingsphere.readwritesplitting.event.config.AddReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.event.config.AlterReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.event.config.DeleteReadwriteSplittingConfigurationEvent;
import org.apache.shardingsphere.readwritesplitting.rule.ReadwriteSplittingRule;
+import org.apache.shardingsphere.readwritesplitting.yaml.config.rule.YamlReadwriteSplittingDataSourceRuleConfiguration;
import java.util.Collections;
import java.util.Map;
@@ -62,8 +66,10 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
if (!event.getActiveVersion().equals(instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey()))) {
return;
}
+
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
- ReadwriteSplittingDataSourceRuleConfiguration needToAddedConfig = event.getConfig();
+ ReadwriteSplittingDataSourceRuleConfiguration needToAddedConfig = swapDataSource(event.getGroupName(),
+ instanceContext.getModeContextManager().getVersionPathByActiveVersionKey(event.getActiveVersionKey(), event.getActiveVersion()));
Optional<ReadwriteSplittingRule> rule = database.getRuleMetaData().findSingleRule(ReadwriteSplittingRule.class);
ReadwriteSplittingRuleConfiguration config;
if (rule.isPresent()) {
@@ -86,7 +92,8 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
- ReadwriteSplittingDataSourceRuleConfiguration needToAlteredConfig = event.getConfig();
+ ReadwriteSplittingDataSourceRuleConfiguration needToAlteredConfig = swapDataSource(event.getGroupName(),
+ instanceContext.getModeContextManager().getVersionPathByActiveVersionKey(event.getActiveVersionKey(), event.getActiveVersion()));
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
config.getDataSources().removeIf(each -> each.getName().equals(needToAlteredConfig.getName()));
config.getDataSources().add(needToAlteredConfig);
@@ -100,12 +107,21 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
*/
@Subscribe
public synchronized void renew(final DeleteReadwriteSplittingConfigurationEvent event) {
- if (!event.getActiveVersion().equals(instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey()))) {
- return;
- }
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
config.getDataSources().removeIf(each -> each.getName().equals(event.getGroupName()));
instanceContext.getEventBusContext().post(new DatabaseRuleConfigurationChangedEvent(event.getDatabaseName(), config));
}
+
+ private ReadwriteSplittingDataSourceRuleConfiguration swapDataSource(final String name, final String yamlContext) {
+ YamlReadwriteSplittingDataSourceRuleConfiguration yamlDataSourceRuleConfig = YamlEngine.unmarshal(yamlContext, YamlReadwriteSplittingDataSourceRuleConfiguration.class);
+ return new ReadwriteSplittingDataSourceRuleConfiguration(name, yamlDataSourceRuleConfig.getWriteDataSourceName(), yamlDataSourceRuleConfig.getReadDataSourceNames(),
+ getTransactionalReadQueryStrategy(yamlDataSourceRuleConfig), yamlDataSourceRuleConfig.getLoadBalancerName());
+ }
+
+ private TransactionalReadQueryStrategy getTransactionalReadQueryStrategy(final YamlReadwriteSplittingDataSourceRuleConfiguration yamlDataSourceRuleConfig) {
+ return Strings.isNullOrEmpty(yamlDataSourceRuleConfig.getTransactionalReadQueryStrategy())
+ ? TransactionalReadQueryStrategy.DYNAMIC
+ : TransactionalReadQueryStrategy.valueOf(yamlDataSourceRuleConfig.getTransactionalReadQueryStrategy());
+ }
}
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingLoadBalanceSubscriber.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingLoadBalanceSubscriber.java
index 83547e663f2..e665d34ee26 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingLoadBalanceSubscriber.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/subscriber/ReadwriteSplittingLoadBalanceSubscriber.java
@@ -19,9 +19,13 @@ package org.apache.shardingsphere.readwritesplitting.subscriber;
import com.google.common.eventbus.Subscribe;
import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.config.algorithm.AlgorithmConfiguration;
import org.apache.shardingsphere.infra.instance.InstanceContext;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.rule.RuleConfigurationSubscribeCoordinator;
+import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
+import org.apache.shardingsphere.infra.yaml.config.pojo.algorithm.YamlAlgorithmConfiguration;
+import org.apache.shardingsphere.infra.yaml.config.swapper.algorithm.YamlAlgorithmConfigurationSwapper;
import org.apache.shardingsphere.mode.event.config.DatabaseRuleConfigurationChangedEvent;
import org.apache.shardingsphere.readwritesplitting.api.ReadwriteSplittingRuleConfiguration;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.AlterLoadBalanceEvent;
@@ -60,7 +64,8 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
- config.getLoadBalancers().put(event.getLoadBalanceName(), event.getConfig());
+ config.getLoadBalancers().put(event.getLoadBalanceName(),
+ swapToAlgorithmConfig(instanceContext.getModeContextManager().getVersionPathByActiveVersionKey(event.getActiveVersionKey(), event.getActiveVersion())));
}
/**
@@ -78,4 +83,8 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
config.getLoadBalancers().remove(event.getLoadBalanceName());
instanceContext.getEventBusContext().post(new DatabaseRuleConfigurationChangedEvent(event.getDatabaseName(), config));
}
+
+ private AlgorithmConfiguration swapToAlgorithmConfig(final String yamlContext) {
+ return new YamlAlgorithmConfigurationSwapper().swapToObject(YamlEngine.unmarshal(yamlContext, YamlAlgorithmConfiguration.class));
+ }
}
diff --git a/features/readwrite-splitting/core/src/test/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverterTest.java b/features/readwrite-splitting/core/src/test/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverterTest.java
index c2e2f5846cd..053921ab964 100644
--- a/features/readwrite-splitting/core/src/test/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverterTest.java
+++ b/features/readwrite-splitting/core/src/test/java/org/apache/shardingsphere/readwritesplitting/metadata/converter/ReadwriteSplittingNodeConverterTest.java
@@ -49,10 +49,10 @@ class ReadwriteSplittingNodeConverterTest {
}
@Test
- void assertGetGroupNameVersion() {
- Optional<String> actual = ReadwriteSplittingNodeConverter.getGroupNameVersion("/metadata/foo_db/rules/readwrite_splitting/data_sources/group_0/versions/0");
+ void assertGetGroupNameByActiveVersionPath() {
+ Optional<String> actual = ReadwriteSplittingNodeConverter.getGroupNameByActiveVersionPath("/metadata/foo_db/rules/readwrite_splitting/data_sources/group_0/active_version");
assertTrue(actual.isPresent());
- assertThat(actual.get(), is("0"));
+ assertThat(actual.get(), is("group_0"));
}
@Test
@@ -63,10 +63,10 @@ class ReadwriteSplittingNodeConverterTest {
}
@Test
- void assertGetLoadBalancerNameVersion() {
- Optional<String> actual = ReadwriteSplittingNodeConverter.getLoadBalancerNameVersion("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random/versions/1");
+ void assertGetLoadBalancerNameByActiveVersionPath() {
+ Optional<String> actual = ReadwriteSplittingNodeConverter.getLoadBalancerNameByActiveVersionPath("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random/active_version");
assertTrue(actual.isPresent());
- assertThat(actual.get(), is("1"));
+ assertThat(actual.get(), is("random"));
}
@Test
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/instance/mode/ModeContextManager.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/instance/mode/ModeContextManager.java
index 6ca92f08fd3..14205c35cc8 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/instance/mode/ModeContextManager.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/instance/mode/ModeContextManager.java
@@ -167,6 +167,7 @@ public interface ModeContextManager {
Collection<MetaDataVersion> newAlterProperties(Properties props);
/**
+ * TODO Remove to MetaDataPersistService
* Get active version by key.
*
* @param key key
@@ -175,4 +176,16 @@ public interface ModeContextManager {
default String getActiveVersionByKey(String key) {
return "";
}
+
+ /**
+ * TODO Remove to MetaDataPersistService
+ * Get version path by active version key.
+ *
+ * @param key key
+ * @param activeVersion active version
+ * @return version path
+ */
+ default String getVersionPathByActiveVersionKey(String key, String activeVersion) {
+ return "";
+ }
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/NewClusterModeContextManager.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/NewClusterModeContextManager.java
index b55f1e057a9..9bfe5b6f5f3 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/NewClusterModeContextManager.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/NewClusterModeContextManager.java
@@ -26,6 +26,7 @@ import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSp
import org.apache.shardingsphere.infra.metadata.database.schema.pojo.AlterSchemaMetaDataPOJO;
import org.apache.shardingsphere.infra.metadata.database.schema.pojo.AlterSchemaPOJO;
import org.apache.shardingsphere.infra.metadata.version.MetaDataVersion;
+import org.apache.shardingsphere.metadata.persist.node.NewDatabaseMetaDataNode;
import org.apache.shardingsphere.metadata.persist.service.database.DatabaseMetaDataBasedPersistService;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.mode.manager.ContextManagerAware;
@@ -154,6 +155,11 @@ public final class NewClusterModeContextManager implements ModeContextManager, C
return contextManager.getMetaDataContexts().getPersistService().getRepository().getDirectly(key);
}
+ @Override
+ public String getVersionPathByActiveVersionKey(final String key, final String activeVersion) {
+ return contextManager.getMetaDataContexts().getPersistService().getRepository().getDirectly(NewDatabaseMetaDataNode.getVersionNodeByActiveVersionPath(key, activeVersion));
+ }
+
@Override
public void setContextManagerAware(final ContextManager contextManager) {
this.contextManager = contextManager;