You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by ji...@apache.org on 2023/06/17 06:28:12 UTC
[shardingsphere] branch master updated: Refactor Readwirte-Splitting build event and refresh metadata (#26397)
This is an automated email from the ASF dual-hosted git repository.
jianglongtao 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 eb89ea824be Refactor Readwirte-Splitting build event and refresh metadata (#26397)
eb89ea824be is described below
commit eb89ea824bec617ab44b1e8e8468e173e5318205
Author: zhaojinchao <zh...@apache.org>
AuthorDate: Sat Jun 17 14:28:05 2023 +0800
Refactor Readwirte-Splitting build event and refresh metadata (#26397)
* Refactor Readwirte-Splitting build event and refresh metadata
* Fix checkstyle
* Fix ci
---
...riteSplittingRuleConfigurationEventBuilder.java | 38 ++++++++++----------
.../AddReadwriteSplittingConfigurationEvent.java | 11 +++---
.../AlterReadwriteSplittingConfigurationEvent.java | 11 +++---
...DeleteReadwriteSplittingConfigurationEvent.java | 4 +++
.../event/loadbalance/AddLoadBalanceEvent.java | 38 --------------------
.../event/loadbalance/AlterLoadBalanceEvent.java | 11 +++---
.../event/loadbalance/DeleteLoadBalanceEvent.java | 4 +++
.../converter/ReadwriteSplittingNodeConverter.java | 40 ++++++++++++++++++++++
.../ReadwriteSplittingConfigurationSubscriber.java | 16 ++++++---
.../ReadwriteSplittingLoadBalanceSubscriber.java | 28 +++++----------
.../ReadwriteSplittingNodeConverterTest.java | 26 ++++++++++++--
.../infra/instance/mode/ModeContextManager.java | 10 ++++++
.../cluster/NewClusterModeContextManager.java | 7 ++++
13 files changed, 149 insertions(+), 95 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 0ccabc1afc1..bff5ed40184 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
@@ -31,7 +31,6 @@ import org.apache.shardingsphere.readwritesplitting.api.transaction.Transactiona
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.AddLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.AlterLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.DeleteLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.metadata.converter.ReadwriteSplittingNodeConverter;
@@ -49,25 +48,29 @@ public final class ReadwriteSplittingRuleConfigurationEventBuilder implements Ru
if (!ReadwriteSplittingNodeConverter.isReadwriteSplittingPath(event.getKey()) || Strings.isNullOrEmpty(event.getValue())) {
return Optional.empty();
}
- Optional<String> groupName = ReadwriteSplittingNodeConverter.getGroupName(event.getKey());
- if (groupName.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
- return createReadwriteSplittingConfigEvent(databaseName, groupName.get(), event);
+ if (ReadwriteSplittingNodeConverter.isDataSourcePath(event.getKey())) {
+ Optional<String> groupNameVersion = ReadwriteSplittingNodeConverter.getGroupNameVersion(event.getKey());
+ if (groupNameVersion.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
+ return createReadwriteSplittingConfigEvent(databaseName, Integer.parseInt(groupNameVersion.get()), event);
+ }
}
- Optional<String> loadBalancerName = ReadwriteSplittingNodeConverter.getLoadBalancerName(event.getKey());
- if (loadBalancerName.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
- return createLoadBalanceEvent(databaseName, loadBalancerName.get(), event);
+ Optional<String> loadBalancerNameVersion = ReadwriteSplittingNodeConverter.getLoadBalancerNameVersion(event.getKey());
+ if (loadBalancerNameVersion.isPresent() && !Strings.isNullOrEmpty(event.getValue())) {
+ return createLoadBalanceEvent(databaseName, Integer.parseInt(loadBalancerNameVersion.get()), event);
}
return Optional.empty();
}
- private Optional<GovernanceEvent> createReadwriteSplittingConfigEvent(final String databaseName, final String groupName, final DataChangedEvent event) {
+ private Optional<GovernanceEvent> createReadwriteSplittingConfigEvent(final String databaseName, final int version, final DataChangedEvent event) {
+ String groupName = ReadwriteSplittingNodeConverter.getGroupName(event.getKey()).orElse("");
+ String activeVersionPath = ReadwriteSplittingNodeConverter.appendActiveVersion(event.getKey());
if (Type.ADDED == event.getType()) {
- return Optional.of(new AddReadwriteSplittingConfigurationEvent<>(databaseName, swapDataSource(groupName, event.getValue())));
+ return Optional.of(new AddReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), activeVersionPath, version));
}
if (Type.UPDATED == event.getType()) {
- return Optional.of(new AlterReadwriteSplittingConfigurationEvent<>(databaseName, groupName, swapDataSource(groupName, event.getValue())));
+ return Optional.of(new AlterReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), activeVersionPath, version));
}
- return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName));
+ return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName, activeVersionPath, version));
}
private ReadwriteSplittingDataSourceRuleConfiguration swapDataSource(final String name, final String yamlContext) {
@@ -82,14 +85,13 @@ public final class ReadwriteSplittingRuleConfigurationEventBuilder implements Ru
: TransactionalReadQueryStrategy.valueOf(yamlDataSourceRuleConfig.getTransactionalReadQueryStrategy());
}
- private Optional<GovernanceEvent> createLoadBalanceEvent(final String databaseName, final String loadBalanceName, final DataChangedEvent event) {
- if (Type.ADDED == event.getType()) {
- return Optional.of(new AddLoadBalanceEvent<>(databaseName, loadBalanceName, swapToAlgorithmConfig(event.getValue())));
- }
- if (Type.UPDATED == event.getType()) {
- return Optional.of(new AlterLoadBalanceEvent<>(databaseName, loadBalanceName, swapToAlgorithmConfig(event.getValue())));
+ private Optional<GovernanceEvent> createLoadBalanceEvent(final String databaseName, final int version, final DataChangedEvent event) {
+ String loadBalanceName = ReadwriteSplittingNodeConverter.getLoadBalancerName(event.getKey()).orElse("");
+ String activeVersionPath = ReadwriteSplittingNodeConverter.appendActiveVersion(event.getKey());
+ if (Type.ADDED == event.getType() || Type.UPDATED == event.getType()) {
+ return Optional.of(new AlterLoadBalanceEvent(databaseName, loadBalanceName, swapToAlgorithmConfig(event.getValue()), activeVersionPath, version));
}
- return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalanceName));
+ return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalanceName, activeVersionPath, version));
}
private AlgorithmConfiguration swapToAlgorithmConfig(final String yamlContext) {
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 92dfa6010a6..79a3aff7b4e 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,17 +20,20 @@ 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.
- *
- * @param <T> readwrite-splitting configuration
*/
@RequiredArgsConstructor
@Getter
-public final class AddReadwriteSplittingConfigurationEvent<T> implements GovernanceEvent {
+public final class AddReadwriteSplittingConfigurationEvent implements GovernanceEvent {
private final String databaseName;
- private final T config;
+ private final ReadwriteSplittingDataSourceRuleConfiguration config;
+
+ private final String activeVersionKey;
+
+ private final int version;
}
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 71f4eae8242..66a9bdfdb98 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,19 +20,20 @@ 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.
- *
- * @param <T> readwrite-splitting configuration
*/
@RequiredArgsConstructor
@Getter
-public final class AlterReadwriteSplittingConfigurationEvent<T> implements GovernanceEvent {
+public final class AlterReadwriteSplittingConfigurationEvent implements GovernanceEvent {
private final String databaseName;
- private final String groupName;
+ private final ReadwriteSplittingDataSourceRuleConfiguration config;
+
+ private final String activeVersionKey;
- private final T config;
+ private final int version;
}
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 02ef1fe1e19..6c66911f008 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,4 +31,8 @@ public final class DeleteReadwriteSplittingConfigurationEvent implements Governa
private final String databaseName;
private final String groupName;
+
+ private final String activeVersionKey;
+
+ private final int version;
}
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AddLoadBalanceEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AddLoadBalanceEvent.java
deleted file mode 100644
index 3f7e7dc2977..00000000000
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/AddLoadBalanceEvent.java
+++ /dev/null
@@ -1,38 +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.readwritesplitting.event.loadbalance;
-
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.rule.event.GovernanceEvent;
-
-/**
- * Add load-balance event.
- *
- * @param <T> algorithm configuration
- */
-@RequiredArgsConstructor
-@Getter
-public final class AddLoadBalanceEvent<T> implements GovernanceEvent {
-
- private final String databaseName;
-
- private final String loadBalanceName;
-
- private final T config;
-}
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 16520928fc5..70c699321e4 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,20 +19,23 @@ 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;
/**
* Alter load-balance event.
- *
- * @param <T> algorithm configuration
*/
@RequiredArgsConstructor
@Getter
-public final class AlterLoadBalanceEvent<T> implements GovernanceEvent {
+public final class AlterLoadBalanceEvent implements GovernanceEvent {
private final String databaseName;
private final String loadBalanceName;
- private final T config;
+ private final AlgorithmConfiguration config;
+
+ private final String activeVersionKey;
+
+ private final int version;
}
diff --git a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/DeleteLoadBalanceEvent.java b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/DeleteLoadBalanceEvent.java
index 99397c8b4f0..7b30623b323 100644
--- a/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/DeleteLoadBalanceEvent.java
+++ b/features/readwrite-splitting/core/src/main/java/org/apache/shardingsphere/readwritesplitting/event/loadbalance/DeleteLoadBalanceEvent.java
@@ -31,4 +31,8 @@ public final class DeleteLoadBalanceEvent implements GovernanceEvent {
private final String databaseName;
private final String loadBalanceName;
+
+ private final String activeVersionKey;
+
+ private final int version;
}
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 01676f84688..5fd4bc5cbaf 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
@@ -36,10 +36,16 @@ public final class ReadwriteSplittingNodeConverter {
private static final String LOAD_BALANCER_NODE = "load_balancers";
+ private static final String VERSIONS = "versions";
+
+ private static final String ACTIVE_VERSION = "active_version";
+
private static final String RULES_NODE_PREFIX = "/([\\w\\-]+)/([\\w\\-]+)/rules/";
private static final String RULE_NAME_PATTERN = "/([\\w\\-]+)?";
+ private static final String RULE_VERSION = "/([\\w\\-]+)/versions/([\\w\\-]+)$";
+
/**
* Get group name path.
*
@@ -108,6 +114,30 @@ public final class ReadwriteSplittingNodeConverter {
return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
+ /**
+ * Get group name version.
+ *
+ * @param rulePath rule path
+ * @return group name version
+ */
+ 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();
+ }
+
+ /**
+ * Get load balancer name version.
+ *
+ * @param rulePath rule path
+ * @return load balancer name version
+ */
+ 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();
+ }
+
/**
* Get load balancer name.
*
@@ -119,4 +149,14 @@ public final class ReadwriteSplittingNodeConverter {
Matcher matcher = pattern.matcher(rulePath);
return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
+
+ /**
+ * Append active version.
+ *
+ * @param rulePath rule path
+ * @return group name
+ */
+ public static String appendActiveVersion(final String rulePath) {
+ return rulePath.substring(0, rulePath.indexOf(VERSIONS)) + ACTIVE_VERSION;
+ }
}
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 a033c3bf4ec..76a4184dcee 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
@@ -58,14 +58,16 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
* @param event add readwrite-splitting configuration event
*/
@Subscribe
- public synchronized void renew(final AddReadwriteSplittingConfigurationEvent<ReadwriteSplittingDataSourceRuleConfiguration> event) {
+ public synchronized void renew(final AddReadwriteSplittingConfigurationEvent event) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ return;
+ }
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingDataSourceRuleConfiguration needToAddedConfig = event.getConfig();
Optional<ReadwriteSplittingRule> rule = database.getRuleMetaData().findSingleRule(ReadwriteSplittingRule.class);
ReadwriteSplittingRuleConfiguration config;
if (rule.isPresent()) {
config = (ReadwriteSplittingRuleConfiguration) rule.get().getConfiguration();
- config.getDataSources().removeIf(each -> each.getName().equals(needToAddedConfig.getName()));
config.getDataSources().add(needToAddedConfig);
} else {
config = new ReadwriteSplittingRuleConfiguration(Collections.singletonList(needToAddedConfig), Collections.emptyMap());
@@ -79,11 +81,14 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
* @param event alter readwrite-splitting configuration event
*/
@Subscribe
- public synchronized void renew(final AlterReadwriteSplittingConfigurationEvent<ReadwriteSplittingDataSourceRuleConfiguration> event) {
+ public synchronized void renew(final AlterReadwriteSplittingConfigurationEvent event) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ return;
+ }
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingDataSourceRuleConfiguration needToAlteredConfig = event.getConfig();
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
- config.getDataSources().removeIf(each -> each.getName().equals(event.getGroupName()));
+ config.getDataSources().removeIf(each -> each.getName().equals(needToAlteredConfig.getName()));
config.getDataSources().add(needToAlteredConfig);
instanceContext.getEventBusContext().post(new DatabaseRuleConfigurationChangedEvent(event.getDatabaseName(), config));
}
@@ -95,6 +100,9 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
*/
@Subscribe
public synchronized void renew(final DeleteReadwriteSplittingConfigurationEvent event) {
+ if (event.getVersion() < 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()));
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 131d520ae24..2b27a8f19ca 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,13 +19,11 @@ 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.mode.event.config.DatabaseRuleConfigurationChangedEvent;
import org.apache.shardingsphere.readwritesplitting.api.ReadwriteSplittingRuleConfiguration;
-import org.apache.shardingsphere.readwritesplitting.event.loadbalance.AddLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.AlterLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.event.loadbalance.DeleteLoadBalanceEvent;
import org.apache.shardingsphere.readwritesplitting.rule.ReadwriteSplittingRule;
@@ -50,30 +48,19 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
instanceContext.getEventBusContext().register(this);
}
- /**
- * Renew with add load-balance.
- *
- * @param event add load-balance event
- */
- @Subscribe
- public synchronized void renew(final AddLoadBalanceEvent<AlgorithmConfiguration> event) {
- renew(event.getDatabaseName(), event.getLoadBalanceName(), event.getConfig());
- }
-
/**
* Renew with alter load-balance.
*
* @param event alter load-balance event
*/
@Subscribe
- public synchronized void renew(final AlterLoadBalanceEvent<AlgorithmConfiguration> event) {
- renew(event.getDatabaseName(), event.getLoadBalanceName(), event.getConfig());
- }
-
- private void renew(final String databaseName, final String loadBalanceName, final AlgorithmConfiguration algorithmConfig) {
- ShardingSphereDatabase database = databases.get(databaseName);
+ public synchronized void renew(final AlterLoadBalanceEvent event) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ return;
+ }
+ ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
- config.getLoadBalancers().put(loadBalanceName, algorithmConfig);
+ config.getLoadBalancers().put(event.getLoadBalanceName(), event.getConfig());
}
/**
@@ -83,6 +70,9 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
*/
@Subscribe
public synchronized void renew(final DeleteLoadBalanceEvent event) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ return;
+ }
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
ReadwriteSplittingRuleConfiguration config = (ReadwriteSplittingRuleConfiguration) database.getRuleMetaData().getSingleRule(ReadwriteSplittingRule.class).getConfiguration();
config.getLoadBalancers().remove(event.getLoadBalanceName());
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 6de042f87e7..60c423f1251 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,16 +49,36 @@ class ReadwriteSplittingNodeConverterTest {
}
@Test
- void assertGetGroupNameByRulePath() {
- Optional<String> actual = ReadwriteSplittingNodeConverter.getGroupName("/metadata/foo_db/rules/readwrite_splitting/data_sources/group_0/active_version");
+ void assertGetGroupNameVersion() {
+ Optional<String> actual = ReadwriteSplittingNodeConverter.getGroupNameVersion("/metadata/foo_db/rules/readwrite_splitting/data_sources/group_0/versions/0");
+ assertTrue(actual.isPresent());
+ assertThat(actual.get(), is("0"));
+ }
+
+ @Test
+ void assertGetGroupName() {
+ Optional<String> actual = ReadwriteSplittingNodeConverter.getGroupName("/metadata/foo_db/rules/readwrite_splitting/data_sources/group_0/versions/0");
assertTrue(actual.isPresent());
assertThat(actual.get(), is("group_0"));
}
@Test
- void assertGetLoadBalancerNameByRulePath() {
+ void assertGetLoadBalancerNameVersion() {
+ Optional<String> actual = ReadwriteSplittingNodeConverter.getLoadBalancerNameVersion("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random/versions/1");
+ assertTrue(actual.isPresent());
+ assertThat(actual.get(), is("1"));
+ }
+
+ @Test
+ void assertGetLoadBalancerName() {
Optional<String> actual = ReadwriteSplittingNodeConverter.getLoadBalancerName("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random");
assertTrue(actual.isPresent());
assertThat(actual.get(), is("random"));
}
+
+ @Test
+ void assertAppendActiveVersion() {
+ assertThat(ReadwriteSplittingNodeConverter.appendActiveVersion("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random/versions/0"),
+ is("/metadata/foo_db/rules/readwrite_splitting/load_balancers/random/active_version"));
+ }
}
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 5dce6870e21..b8a74f69929 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
@@ -150,4 +150,14 @@ public interface ModeContextManager {
* @param props pros
*/
void alterProperties(Properties props);
+
+ /**
+ * Get active version by key.
+ *
+ * @param key key
+ * @return active version
+ */
+ default int getActiveVersionByKey(String key) {
+ return 0;
+ }
}
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 44655a0bfb2..2246184f0db 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
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.mode.manager.cluster;
+import com.google.common.base.Strings;
import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import org.apache.shardingsphere.infra.datasource.props.DataSourceProperties;
import org.apache.shardingsphere.infra.instance.mode.ModeContextManager;
@@ -144,6 +145,12 @@ public final class NewClusterModeContextManager implements ModeContextManager, C
contextManager.getMetaDataContexts().getPersistService().getPropsService().persist(props);
}
+ @Override
+ public int getActiveVersionByKey(final String key) {
+ String activeVersion = contextManager.getMetaDataContexts().getPersistService().getRepository().getDirectly(key);
+ return Strings.isNullOrEmpty(activeVersion) ? 0 : Integer.parseInt(activeVersion);
+ }
+
@Override
public void setContextManagerAware(final ContextManager contextManager) {
this.contextManager = contextManager;