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/18 15:24:21 UTC
[shardingsphere] branch master updated: Add NewResourceMetaDataChangedSubscriber to refresh schema meta data (#26421)
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 885127f121e Add NewResourceMetaDataChangedSubscriber to refresh schema meta data (#26421)
885127f121e is described below
commit 885127f121e70f693acc1234806696790af16b12
Author: zhaojinchao <zh...@apache.org>
AuthorDate: Sun Jun 18 23:24:13 2023 +0800
Add NewResourceMetaDataChangedSubscriber to refresh schema meta data (#26421)
* Add NewResourceMetaDataChangedSubscriber to refresh schema meta data
* Remove unless unit test
---
...riteSplittingRuleConfigurationEventBuilder.java | 12 +-
.../AddReadwriteSplittingConfigurationEvent.java | 2 +-
.../AlterReadwriteSplittingConfigurationEvent.java | 2 +-
...DeleteReadwriteSplittingConfigurationEvent.java | 2 +-
.../event/loadbalance/AlterLoadBalanceEvent.java | 2 +-
.../event/loadbalance/DeleteLoadBalanceEvent.java | 2 +-
.../converter/ReadwriteSplittingNodeConverter.java | 14 ---
.../ReadwriteSplittingConfigurationSubscriber.java | 6 +-
.../ReadwriteSplittingLoadBalanceSubscriber.java | 4 +-
.../ReadwriteSplittingNodeConverterTest.java | 6 -
.../persist/node/NewDatabaseMetaDataNode.java | 38 +++++-
.../persist/node/NewDatabaseMetaDataNodeTest.java | 22 +++-
.../event/datasource/AlterStorageUnitEvent.java | 2 +-
.../event/datasource/RegisterStorageUnitEvent.java | 2 +-
.../datasource/UnregisterStorageUnitEvent.java | 2 +-
.../mode/event/schema/table/AlterTableEvent.java | 6 +-
.../mode/event/schema/table/DropTableEvent.java | 4 +
.../mode/event/schema/view/AlterViewEvent.java | 6 +-
.../mode/event/schema/view/DropViewEvent.java | 4 +
.../cluster/NewClusterModeContextManager.java | 3 +-
.../watcher/NewMetaDataChangedWatcher.java | 32 ++---
.../NewContextManagerSubscriberFacade.java | 2 +-
.../subscriber/NewDataSourceChangedSubscriber.java | 6 +-
.../NewResourceMetaDataChangedSubscriber.java | 137 +++++++++++++++++++++
24 files changed, 252 insertions(+), 66 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 bff5ed40184..bd728bbb2aa 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
@@ -63,14 +63,13 @@ public final class ReadwriteSplittingRuleConfigurationEventBuilder implements Ru
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()), activeVersionPath, version));
+ return Optional.of(new AddReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), event.getKey(), version));
}
if (Type.UPDATED == event.getType()) {
- return Optional.of(new AlterReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), activeVersionPath, version));
+ return Optional.of(new AlterReadwriteSplittingConfigurationEvent(databaseName, swapDataSource(groupName, event.getValue()), event.getKey(), version));
}
- return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName, activeVersionPath, version));
+ return Optional.of(new DeleteReadwriteSplittingConfigurationEvent(databaseName, groupName, event.getKey(), version));
}
private ReadwriteSplittingDataSourceRuleConfiguration swapDataSource(final String name, final String yamlContext) {
@@ -87,11 +86,10 @@ public final class ReadwriteSplittingRuleConfigurationEventBuilder implements Ru
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 AlterLoadBalanceEvent(databaseName, loadBalanceName, swapToAlgorithmConfig(event.getValue()), event.getKey(), version));
}
- return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalanceName, activeVersionPath, version));
+ return Optional.of(new DeleteLoadBalanceEvent(databaseName, loadBalanceName, event.getKey(), 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 79a3aff7b4e..6affb7bf400 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
@@ -33,7 +33,7 @@ public final class AddReadwriteSplittingConfigurationEvent implements Governance
private final ReadwriteSplittingDataSourceRuleConfiguration config;
- private final String activeVersionKey;
+ private final String versionKey;
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 66a9bdfdb98..e17987c4320 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
@@ -33,7 +33,7 @@ public final class AlterReadwriteSplittingConfigurationEvent implements Governan
private final ReadwriteSplittingDataSourceRuleConfiguration config;
- private final String activeVersionKey;
+ private final String versionKey;
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 6c66911f008..2c5ec048a05 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
@@ -32,7 +32,7 @@ public final class DeleteReadwriteSplittingConfigurationEvent implements Governa
private final String groupName;
- private final String activeVersionKey;
+ private final String versionKey;
private final int version;
}
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 70c699321e4..32546b28d31 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
@@ -35,7 +35,7 @@ public final class AlterLoadBalanceEvent implements GovernanceEvent {
private final AlgorithmConfiguration config;
- private final String activeVersionKey;
+ private final String versionKey;
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 7b30623b323..782844e0938 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
@@ -32,7 +32,7 @@ public final class DeleteLoadBalanceEvent implements GovernanceEvent {
private final String loadBalanceName;
- private final String activeVersionKey;
+ private final String versionKey;
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 b6a4a8e3165..9a0d7fbb573 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,6 @@ 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\\-]+)/versions/?";
@@ -149,14 +145,4 @@ 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 76a4184dcee..7332bd77806 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
@@ -59,7 +59,7 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
*/
@Subscribe
public synchronized void renew(final AddReadwriteSplittingConfigurationEvent event) {
- if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
@@ -82,7 +82,7 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
*/
@Subscribe
public synchronized void renew(final AlterReadwriteSplittingConfigurationEvent event) {
- if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
@@ -100,7 +100,7 @@ public final class ReadwriteSplittingConfigurationSubscriber implements RuleConf
*/
@Subscribe
public synchronized void renew(final DeleteReadwriteSplittingConfigurationEvent event) {
- if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
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 2b27a8f19ca..274d470068c 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
@@ -55,7 +55,7 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
*/
@Subscribe
public synchronized void renew(final AlterLoadBalanceEvent event) {
- if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
@@ -70,7 +70,7 @@ public final class ReadwriteSplittingLoadBalanceSubscriber implements RuleConfig
*/
@Subscribe
public synchronized void renew(final DeleteLoadBalanceEvent event) {
- if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < instanceContext.getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
ShardingSphereDatabase database = databases.get(event.getDatabaseName());
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 1890d04d928..c2e2f5846cd 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
@@ -75,10 +75,4 @@ class ReadwriteSplittingNodeConverterTest {
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/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNode.java b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNode.java
index 0bc06538d89..a0ccc916be3 100644
--- a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNode.java
+++ b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNode.java
@@ -343,11 +343,23 @@ public final class NewDatabaseMetaDataNode {
* @return table name
*/
public static Optional<String> getTableName(final String path) {
- Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/tables" + "/([\\w\\-]+)$", Pattern.CASE_INSENSITIVE);
+ Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/tables" + "/([\\w\\-]+)?", Pattern.CASE_INSENSITIVE);
Matcher matcher = pattern.matcher(path);
return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
+ /**
+ * Get table name version.
+ *
+ * @param path path
+ * @return table name version
+ */
+ public static Optional<String> getTableNameVersion(final String path) {
+ Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/tables" + "/([\\w\\-]+)/versions/([\\w\\-]+)$", Pattern.CASE_INSENSITIVE);
+ Matcher matcher = pattern.matcher(path);
+ return matcher.find() ? Optional.of(matcher.group(4)) : Optional.empty();
+ }
+
/**
* Get view name.
*
@@ -355,11 +367,33 @@ public final class NewDatabaseMetaDataNode {
* @return view name
*/
public static Optional<String> getViewName(final String path) {
- Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/views" + "/([\\w\\-]+)$", Pattern.CASE_INSENSITIVE);
+ Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/views" + "/([\\w\\-]+)?", Pattern.CASE_INSENSITIVE);
Matcher matcher = pattern.matcher(path);
return matcher.find() ? Optional.of(matcher.group(3)) : Optional.empty();
}
+ /**
+ * Get view name version.
+ *
+ * @param path path
+ * @return view name version
+ */
+ public static Optional<String> getViewNameVersion(final String path) {
+ Pattern pattern = Pattern.compile(getMetaDataNodeNode() + "/([\\w\\-]+)/schemas/([\\w\\-]+)/views" + "/([\\w\\-]+)/versions/([\\w\\-]+)$", Pattern.CASE_INSENSITIVE);
+ Matcher matcher = pattern.matcher(path);
+ return matcher.find() ? Optional.of(matcher.group(4)) : Optional.empty();
+ }
+
+ /**
+ * Decorate active version.
+ *
+ * @param rulePath rule path
+ * @return group name
+ */
+ public static String decorateActiveVersion(final String rulePath) {
+ return rulePath.substring(0, rulePath.indexOf(VERSIONS)) + ACTIVE_VERSION;
+ }
+
private static String getMetaDataNodeNode() {
return String.join("/", "", ROOT_NODE);
}
diff --git a/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNodeTest.java b/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNodeTest.java
index 0c9fcebf041..d86f504a6d8 100644
--- a/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNodeTest.java
+++ b/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/NewDatabaseMetaDataNodeTest.java
@@ -77,11 +77,18 @@ class NewDatabaseMetaDataNodeTest {
@Test
void assertGetTableName() {
- Optional<String> actual = NewDatabaseMetaDataNode.getTableName("/metadata/foo_db/schemas/foo_schema/tables/foo_table");
+ Optional<String> actual = NewDatabaseMetaDataNode.getTableName("/metadata/foo_db/schemas/foo_schema/tables/foo_table/versions/0");
assertTrue(actual.isPresent());
assertThat(actual.get(), is("foo_table"));
}
+ @Test
+ void assertGetTableVersion() {
+ Optional<String> actual = NewDatabaseMetaDataNode.getTableNameVersion("/metadata/foo_db/schemas/foo_schema/tables/foo_table/versions/0");
+ assertTrue(actual.isPresent());
+ assertThat(actual.get(), is("0"));
+ }
+
@Test
void assertGetViewName() {
Optional<String> actual = NewDatabaseMetaDataNode.getViewName("/metadata/foo_db/schemas/foo_schema/views/foo_view");
@@ -89,6 +96,19 @@ class NewDatabaseMetaDataNodeTest {
assertThat(actual.get(), is("foo_view"));
}
+ @Test
+ void assertGetViewNameVersion() {
+ Optional<String> actual = NewDatabaseMetaDataNode.getViewNameVersion("/metadata/foo_db/schemas/foo_schema/views/foo_view/versions/0");
+ assertTrue(actual.isPresent());
+ assertThat(actual.get(), is("0"));
+ }
+
+ @Test
+ void assertDecorateActiveVersion() {
+ assertThat(NewDatabaseMetaDataNode.decorateActiveVersion("/metadata/foo_db/schemas/foo_schema/views/foo_view/versions/0"),
+ is("/metadata/foo_db/schemas/foo_schema/views/foo_view/active_version"));
+ }
+
@Test
void assertGetMetaDataDataSourcesNode() {
assertThat(NewDatabaseMetaDataNode.getDataSourcesNode("foo_db"), is("/metadata/foo_db/data_sources"));
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/AlterStorageUnitEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/AlterStorageUnitEvent.java
index a2ae0660d82..14071820767 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/AlterStorageUnitEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/AlterStorageUnitEvent.java
@@ -35,7 +35,7 @@ public final class AlterStorageUnitEvent implements GovernanceEvent {
private final DataSourceProperties props;
- private final String activeVersionKey;
+ private final String versionKey;
private final int version;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/RegisterStorageUnitEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/RegisterStorageUnitEvent.java
index 7b55f07c1b2..18b5e01cbb4 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/RegisterStorageUnitEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/RegisterStorageUnitEvent.java
@@ -35,7 +35,7 @@ public final class RegisterStorageUnitEvent implements GovernanceEvent {
private final DataSourceProperties props;
- private final String activeVersionKey;
+ private final String versionKey;
private final int version;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/UnregisterStorageUnitEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/UnregisterStorageUnitEvent.java
index 4198e7647c4..7a036bf56d3 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/UnregisterStorageUnitEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/datasource/UnregisterStorageUnitEvent.java
@@ -32,7 +32,7 @@ public final class UnregisterStorageUnitEvent implements GovernanceEvent {
private final String storageUnitName;
- private final String activeVersionKey;
+ private final String versionKey;
private final int version;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/AlterTableEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/AlterTableEvent.java
index d96c7269f6a..278bb4fef71 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/AlterTableEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/AlterTableEvent.java
@@ -33,7 +33,9 @@ public final class AlterTableEvent implements GovernanceEvent {
private final String schemaName;
- private final String version;
-
private final ShardingSphereTable table;
+
+ private final int version;
+
+ private final String versionKey;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/DropTableEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/DropTableEvent.java
index 39378fa29da..27146d7c2bb 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/DropTableEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/table/DropTableEvent.java
@@ -33,4 +33,8 @@ public final class DropTableEvent implements GovernanceEvent {
private final String schemaName;
private final String tableName;
+
+ private final int version;
+
+ private final String versionKey;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/AlterViewEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/AlterViewEvent.java
index 94b7490a38c..77914291c40 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/AlterViewEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/AlterViewEvent.java
@@ -33,7 +33,9 @@ public final class AlterViewEvent implements GovernanceEvent {
private final String schemaName;
- private final String version;
-
private final ShardingSphereView view;
+
+ private final int version;
+
+ private final String versionKey;
}
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/DropViewEvent.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/DropViewEvent.java
index 80053c4ba6f..1054db29149 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/DropViewEvent.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/event/schema/view/DropViewEvent.java
@@ -33,4 +33,8 @@ public final class DropViewEvent implements GovernanceEvent {
private final String schemaName;
private final String viewName;
+
+ private final int version;
+
+ private final String versionKey;
}
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 2246184f0db..dd3c97a2188 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.model.ShardingSphereView;
import org.apache.shardingsphere.infra.metadata.database.schema.pojo.AlterSchemaMetaDataPOJO;
import org.apache.shardingsphere.infra.metadata.database.schema.pojo.AlterSchemaPOJO;
+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;
@@ -147,7 +148,7 @@ public final class NewClusterModeContextManager implements ModeContextManager, C
@Override
public int getActiveVersionByKey(final String key) {
- String activeVersion = contextManager.getMetaDataContexts().getPersistService().getRepository().getDirectly(key);
+ String activeVersion = contextManager.getMetaDataContexts().getPersistService().getRepository().getDirectly(NewDatabaseMetaDataNode.decorateActiveVersion(key));
return Strings.isNullOrEmpty(activeVersion) ? 0 : Integer.parseInt(activeVersion);
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/metadata/watcher/NewMetaDataChangedWatcher.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/metadata/watcher/NewMetaDataChangedWatcher.java
index ff5551da820..19e78207628 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/metadata/watcher/NewMetaDataChangedWatcher.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/metadata/watcher/NewMetaDataChangedWatcher.java
@@ -81,13 +81,13 @@ public final class NewMetaDataChangedWatcher implements NewGovernanceWatcher<Gov
return createSchemaChangedEvent(databaseName.get(), schemaName.get(), event);
}
schemaName = NewDatabaseMetaDataNode.getSchemaNameByTableNode(key);
- Optional<String> tableName = NewDatabaseMetaDataNode.getTableName(key);
- if (databaseName.isPresent() && schemaName.isPresent() && tableName.isPresent()) {
- return createTableChangedEvent(databaseName.get(), schemaName.get(), tableName.get(), event);
+ Optional<String> tableNameVersion = NewDatabaseMetaDataNode.getTableNameVersion(key);
+ if (databaseName.isPresent() && schemaName.isPresent() && tableNameVersion.isPresent()) {
+ return createTableChangedEvent(databaseName.get(), schemaName.get(), tableNameVersion.get(), event);
}
- Optional<String> viewName = NewDatabaseMetaDataNode.getViewName(key);
- if (databaseName.isPresent() && schemaName.isPresent() && viewName.isPresent()) {
- return createViewChangedEvent(databaseName.get(), schemaName.get(), viewName.get(), event);
+ Optional<String> viewNameVersion = NewDatabaseMetaDataNode.getViewNameVersion(key);
+ if (databaseName.isPresent() && schemaName.isPresent() && viewNameVersion.isPresent()) {
+ return createViewChangedEvent(databaseName.get(), schemaName.get(), viewNameVersion.get(), event);
}
if (!databaseName.isPresent()) {
return Optional.empty();
@@ -118,16 +118,20 @@ public final class NewMetaDataChangedWatcher implements NewGovernanceWatcher<Gov
return Optional.empty();
}
- private Optional<GovernanceEvent> createTableChangedEvent(final String databaseName, final String schemaName, final String tableName, final DataChangedEvent event) {
- return Type.DELETED == event.getType()
- ? Optional.of(new DropTableEvent(databaseName, schemaName, tableName))
- : Optional.of(new AlterTableEvent(databaseName, schemaName, null, new YamlTableSwapper().swapToObject(YamlEngine.unmarshal(event.getValue(), YamlShardingSphereTable.class))));
+ private Optional<GovernanceEvent> createTableChangedEvent(final String databaseName, final String schemaName, final String tableNameVersion, final DataChangedEvent event) {
+ if (Type.DELETED == event.getType()) {
+ return Optional.of(new DropTableEvent(databaseName, schemaName, NewDatabaseMetaDataNode.getTableName(event.getKey()).orElse(""), Integer.parseInt(tableNameVersion), event.getKey()));
+ }
+ return Optional.of(new AlterTableEvent(databaseName, schemaName,
+ new YamlTableSwapper().swapToObject(YamlEngine.unmarshal(event.getValue(), YamlShardingSphereTable.class)), Integer.parseInt(tableNameVersion), event.getKey()));
}
- private Optional<GovernanceEvent> createViewChangedEvent(final String databaseName, final String schemaName, final String viewName, final DataChangedEvent event) {
- return Type.DELETED == event.getType()
- ? Optional.of(new DropViewEvent(databaseName, schemaName, viewName))
- : Optional.of(new AlterViewEvent(databaseName, schemaName, null, new YamlViewSwapper().swapToObject(YamlEngine.unmarshal(event.getValue(), YamlShardingSphereView.class))));
+ private Optional<GovernanceEvent> createViewChangedEvent(final String databaseName, final String schemaName, final String viewNameVersion, final DataChangedEvent event) {
+ if (Type.DELETED == event.getType()) {
+ return Optional.of(new DropViewEvent(databaseName, schemaName, NewDatabaseMetaDataNode.getViewName(event.getKey()).orElse(""), Integer.parseInt(viewNameVersion), event.getKey()));
+ }
+ return Optional.of(new AlterViewEvent(databaseName, schemaName,
+ new YamlViewSwapper().swapToObject(YamlEngine.unmarshal(event.getValue(), YamlShardingSphereView.class)), Integer.parseInt(viewNameVersion), event.getKey()));
}
@SuppressWarnings("unchecked")
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewContextManagerSubscriberFacade.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewContextManagerSubscriberFacade.java
index 1672cf8390c..03f52e00856 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewContextManagerSubscriberFacade.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewContextManagerSubscriberFacade.java
@@ -30,7 +30,7 @@ public final class NewContextManagerSubscriberFacade {
public NewContextManagerSubscriberFacade(final NewRegistryCenter registryCenter, final ContextManager contextManager) {
new NewConfigurationChangedSubscriber(contextManager);
new NewDataSourceChangedSubscriber(contextManager);
- new ResourceMetaDataChangedSubscriber(contextManager);
+ new NewResourceMetaDataChangedSubscriber(contextManager);
new DatabaseChangedSubscriber(contextManager);
new NewStateChangedSubscriber(registryCenter, contextManager);
new NewProcessListChangedSubscriber(registryCenter, contextManager);
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewDataSourceChangedSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewDataSourceChangedSubscriber.java
index 71100f34799..e90cb07e78d 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewDataSourceChangedSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewDataSourceChangedSubscriber.java
@@ -44,7 +44,7 @@ public final class NewDataSourceChangedSubscriber {
*/
@Subscribe
public void renew(final RegisterStorageUnitEvent event) {
- if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
contextManager.registerStorageUnit(event.getDatabaseName(), event.getStorageUnitName(), event.getProps());
@@ -57,7 +57,7 @@ public final class NewDataSourceChangedSubscriber {
*/
@Subscribe
public void renew(final AlterStorageUnitEvent event) {
- if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
contextManager.alterStorageUnit(event.getDatabaseName(), event.getStorageUnitName(), event.getProps());
@@ -70,7 +70,7 @@ public final class NewDataSourceChangedSubscriber {
*/
@Subscribe
public void renew(final UnregisterStorageUnitEvent event) {
- if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getActiveVersionKey())) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
return;
}
contextManager.unregisterStorageUnit(event.getDatabaseName(), event.getStorageUnitName());
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewResourceMetaDataChangedSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewResourceMetaDataChangedSubscriber.java
new file mode 100644
index 00000000000..6122ff27d98
--- /dev/null
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/NewResourceMetaDataChangedSubscriber.java
@@ -0,0 +1,137 @@
+/*
+ * 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.subscriber;
+
+import com.google.common.eventbus.Subscribe;
+import org.apache.shardingsphere.mode.event.schema.table.AlterTableEvent;
+import org.apache.shardingsphere.mode.event.schema.table.DropTableEvent;
+import org.apache.shardingsphere.mode.event.schema.view.AlterViewEvent;
+import org.apache.shardingsphere.mode.event.schema.view.DropViewEvent;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.metadata.event.DatabaseAddedEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.metadata.event.DatabaseDeletedEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.metadata.event.SchemaAddedEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.metadata.event.SchemaDeletedEvent;
+
+/**
+ * TODO Rename ResourceMetaDataChangedSubscriber when metadata structure adjustment completed. #25485
+ * Resource meta data changed subscriber.
+ */
+@SuppressWarnings("UnstableApiUsage")
+public final class NewResourceMetaDataChangedSubscriber {
+
+ private final ContextManager contextManager;
+
+ public NewResourceMetaDataChangedSubscriber(final ContextManager contextManager) {
+ this.contextManager = contextManager;
+ contextManager.getInstanceContext().getEventBusContext().register(this);
+ }
+
+ /**
+ * Renew to persist meta data.
+ *
+ * @param event database added event
+ */
+ @Subscribe
+ public synchronized void renew(final DatabaseAddedEvent event) {
+ contextManager.addDatabase(event.getDatabaseName());
+ }
+
+ /**
+ * Renew to delete database.
+ *
+ * @param event database delete event
+ */
+ @Subscribe
+ public synchronized void renew(final DatabaseDeletedEvent event) {
+ contextManager.dropDatabase(event.getDatabaseName());
+ }
+
+ /**
+ * Renew to added schema.
+ *
+ * @param event schema added event
+ */
+ @Subscribe
+ public synchronized void renew(final SchemaAddedEvent event) {
+ contextManager.addSchema(event.getDatabaseName(), event.getSchemaName());
+ }
+
+ /**
+ * Renew to delete schema.
+ *
+ * @param event schema delete event
+ */
+ @Subscribe
+ public synchronized void renew(final SchemaDeletedEvent event) {
+ contextManager.dropSchema(event.getDatabaseName(), event.getSchemaName());
+ }
+
+ /**
+ * Renew table.
+ *
+ * @param event alter table event
+ */
+ @Subscribe
+ public synchronized void renew(final AlterTableEvent event) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
+ return;
+ }
+ contextManager.alterSchema(event.getDatabaseName(), event.getSchemaName(), event.getTable(), null);
+ }
+
+ /**
+ * Renew table.
+ *
+ * @param event drop table event
+ */
+ @Subscribe
+ public synchronized void renew(final DropTableEvent event) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
+ return;
+ }
+ contextManager.alterSchema(event.getDatabaseName(), event.getSchemaName(), event.getTableName(), null);
+
+ }
+
+ /**
+ * Renew view.
+ *
+ * @param event alter view event
+ */
+ @Subscribe
+ public synchronized void renew(final AlterViewEvent event) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
+ return;
+ }
+ contextManager.alterSchema(event.getDatabaseName(), event.getSchemaName(), null, event.getView());
+ }
+
+ /**
+ * Renew view.
+ *
+ * @param event drop view event
+ */
+ @Subscribe
+ public synchronized void renew(final DropViewEvent event) {
+ if (event.getVersion() < contextManager.getInstanceContext().getModeContextManager().getActiveVersionByKey(event.getVersionKey())) {
+ return;
+ }
+ contextManager.alterSchema(event.getDatabaseName(), event.getSchemaName(), null, event.getViewName());
+ }
+}