You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by du...@apache.org on 2024/06/06 03:46:07 UTC
(shardingsphere) branch master updated: Refactor DataSourceStateManager (#31596)
This is an automated email from the ASF dual-hosted git repository.
duanzhengqiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new c47ef3567a8 Refactor DataSourceStateManager (#31596)
c47ef3567a8 is described below
commit c47ef3567a8375454a9432eb1d41f87a86456a60
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Thu Jun 6 11:46:01 2024 +0800
Refactor DataSourceStateManager (#31596)
---
.../infra/database/DatabaseTypeEngine.java | 28 ++--
.../metadata/database/ShardingSphereDatabase.java | 3 +-
.../state/datasource/DataSourceStateManager.java | 162 ---------------------
.../infra/database/DatabaseTypeEngineTest.java | 8 +-
.../metadata/factory/ExternalMetaDataFactory.java | 6 +-
.../metadata/factory/InternalMetaDataFactory.java | 2 +-
.../SingleRuleConfigurationDecorator.java | 6 +-
.../shardingsphere/single/rule/SingleRule.java | 6 +-
.../mode/metadata/MetaDataContextsFactory.java | 11 --
.../subscriber/StateChangedSubscriber.java | 3 -
.../proxy/version/ShardingSphereProxyVersion.java | 7 +-
11 files changed, 31 insertions(+), 211 deletions(-)
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngine.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngine.java
index 241d59987f4..a973e47d5f7 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngine.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngine.java
@@ -27,7 +27,6 @@ import org.apache.shardingsphere.infra.database.core.type.DatabaseTypeFactory;
import org.apache.shardingsphere.infra.datasource.pool.CatalogSwitchableDataSource;
import org.apache.shardingsphere.infra.exception.core.external.sql.type.wrapper.SQLWrapperException;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import javax.sql.DataSource;
import java.sql.Connection;
@@ -38,6 +37,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.stream.Collectors;
/**
* Database type engine.
@@ -50,18 +50,18 @@ public final class DatabaseTypeEngine {
/**
* Get protocol type.
*
- * @param databaseName database name
* @param databaseConfig database configuration
* @param props configuration properties
* @return protocol type
*/
- public static DatabaseType getProtocolType(final String databaseName, final DatabaseConfiguration databaseConfig, final ConfigurationProperties props) {
+ public static DatabaseType getProtocolType(final DatabaseConfiguration databaseConfig, final ConfigurationProperties props) {
Optional<DatabaseType> configuredDatabaseType = findConfiguredDatabaseType(props);
if (configuredDatabaseType.isPresent()) {
return configuredDatabaseType.get();
}
- Collection<DataSource> enabledDataSources = DataSourceStateManager.getInstance().getEnabledDataSources(databaseName, databaseConfig).values();
- return enabledDataSources.isEmpty() ? getDefaultStorageType() : getStorageType(enabledDataSources.iterator().next());
+ Collection<DataSource> dataSources = databaseConfig.getStorageUnits().entrySet().stream()
+ .collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getDataSource(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new)).values();
+ return dataSources.isEmpty() ? getDefaultStorageType() : getStorageType(dataSources.iterator().next());
}
/**
@@ -76,8 +76,8 @@ public final class DatabaseTypeEngine {
if (configuredDatabaseType.isPresent()) {
return configuredDatabaseType.get();
}
- Map<String, DataSource> enabledDataSources = getEnabledDataSources(databaseConfigs);
- return enabledDataSources.isEmpty() ? getDefaultStorageType() : getStorageType(enabledDataSources.values().iterator().next());
+ Map<String, DataSource> dataSources = getDataSources(databaseConfigs);
+ return dataSources.isEmpty() ? getDefaultStorageType() : getStorageType(dataSources.values().iterator().next());
}
private static Optional<DatabaseType> findConfiguredDatabaseType(final ConfigurationProperties props) {
@@ -85,10 +85,12 @@ public final class DatabaseTypeEngine {
return null == configuredDatabaseType ? Optional.empty() : Optional.of(configuredDatabaseType.getTrunkDatabaseType().orElse(configuredDatabaseType));
}
- private static Map<String, DataSource> getEnabledDataSources(final Map<String, ? extends DatabaseConfiguration> databaseConfigs) {
+ private static Map<String, DataSource> getDataSources(final Map<String, ? extends DatabaseConfiguration> databaseConfigs) {
Map<String, DataSource> result = new LinkedHashMap<>();
for (Entry<String, ? extends DatabaseConfiguration> entry : databaseConfigs.entrySet()) {
- result.putAll(DataSourceStateManager.getInstance().getEnabledDataSources(entry.getKey(), entry.getValue()));
+ Map<String, DataSource> dataSources = entry.getValue().getStorageUnits().entrySet().stream()
+ .collect(Collectors.toMap(Entry::getKey, storageUnit -> storageUnit.getValue().getDataSource(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
+ result.putAll(dataSources);
}
return result;
}
@@ -96,14 +98,14 @@ public final class DatabaseTypeEngine {
/**
* Get storage types.
*
- * @param databaseName database name
* @param databaseConfig database configuration
* @return storage types
*/
- public static Map<String, DatabaseType> getStorageTypes(final String databaseName, final DatabaseConfiguration databaseConfig) {
+ public static Map<String, DatabaseType> getStorageTypes(final DatabaseConfiguration databaseConfig) {
Map<String, DatabaseType> result = new LinkedHashMap<>(databaseConfig.getStorageUnits().size(), 1F);
- Map<String, DataSource> enabledDataSources = DataSourceStateManager.getInstance().getEnabledDataSources(databaseName, databaseConfig);
- for (Entry<String, DataSource> entry : enabledDataSources.entrySet()) {
+ Map<String, DataSource> dataSources = databaseConfig.getStorageUnits().entrySet().stream()
+ .collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getDataSource(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
+ for (Entry<String, DataSource> entry : dataSources.entrySet()) {
result.put(entry.getKey(), getStorageType(entry.getValue()));
}
return result;
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/ShardingSphereDatabase.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/ShardingSphereDatabase.java
index a4216384d7f..72215aae24b 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/ShardingSphereDatabase.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/database/ShardingSphereDatabase.java
@@ -36,7 +36,6 @@ import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSp
import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
import org.apache.shardingsphere.infra.rule.attribute.datanode.MutableDataNodeRuleAttribute;
import org.apache.shardingsphere.infra.rule.builder.database.DatabaseRulesBuilder;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import javax.sql.DataSource;
import java.sql.SQLException;
@@ -92,7 +91,7 @@ public final class ShardingSphereDatabase {
ResourceMetaData resourceMetaData = createResourceMetaData(databaseConfig.getDataSources(), databaseConfig.getStorageUnits());
Collection<ShardingSphereRule> databaseRules = DatabaseRulesBuilder.build(name, protocolType, databaseConfig, computeNodeInstanceContext, resourceMetaData);
Map<String, ShardingSphereSchema> schemas = new ConcurrentHashMap<>(GenericSchemaBuilder
- .build(new GenericSchemaBuilderMaterial(protocolType, storageTypes, DataSourceStateManager.getInstance().getEnabledDataSources(name, databaseConfig), databaseRules,
+ .build(new GenericSchemaBuilderMaterial(protocolType, storageTypes, resourceMetaData.getDataSourceMap(), databaseRules,
props, new DatabaseTypeRegistry(protocolType).getDefaultSchemaName(name))));
SystemSchemaBuilder.build(name, protocolType, props).forEach(schemas::putIfAbsent);
return create(name, protocolType, databaseRules, schemas, resourceMetaData);
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/state/datasource/DataSourceStateManager.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/state/datasource/DataSourceStateManager.java
deleted file mode 100644
index 0a52864bb66..00000000000
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/state/datasource/DataSourceStateManager.java
+++ /dev/null
@@ -1,162 +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.infra.state.datasource;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.infra.config.database.DatabaseConfiguration;
-import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
-import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
-import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedDataSource;
-import org.apache.shardingsphere.infra.state.datasource.exception.UnavailableDataSourceException;
-
-import javax.sql.DataSource;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
-
-/**
- * Data source state manager.
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-@Slf4j
-public final class DataSourceStateManager {
-
- private static final DataSourceStateManager INSTANCE = new DataSourceStateManager();
-
- private final Map<String, DataSourceState> dataSourceStates = new ConcurrentHashMap<>();
-
- private volatile boolean forceStart;
-
- private final AtomicBoolean initialized = new AtomicBoolean(false);
-
- /**
- * Get data source state manager.
- *
- * @return data source state manager
- */
- public static DataSourceStateManager getInstance() {
- return INSTANCE;
- }
-
- /**
- * Set data source states when bootstrap.
- *
- * @param databaseName database name
- * @param storageUnits storage units
- * @param storageDataSourceStates storage node data source state
- * @param forceStart whether to force start
- */
- public void initStates(final String databaseName, final Map<String, StorageUnit> storageUnits, final Map<String, DataSourceState> storageDataSourceStates, final boolean forceStart) {
- this.forceStart = forceStart;
- if (initialized.compareAndSet(false, true)) {
- storageUnits.forEach((key, value) -> initState(databaseName, storageDataSourceStates, key, value.getDataSource()));
- }
- }
-
- private void initState(final String databaseName, final Map<String, DataSourceState> storageDataSourceStates, final String actualDataSourceName, final DataSource dataSource) {
- DataSourceState storageState = storageDataSourceStates.get(getCacheKey(databaseName, actualDataSourceName));
- if (DataSourceState.DISABLED == storageState) {
- dataSourceStates.put(getCacheKey(databaseName, actualDataSourceName), storageState);
- } else {
- checkState(databaseName, actualDataSourceName, dataSource);
- }
- }
-
- private void checkState(final String databaseName, final String actualDataSourceName, final DataSource dataSource) {
- try (Connection ignored = dataSource.getConnection()) {
- dataSourceStates.put(getCacheKey(databaseName, actualDataSourceName), DataSourceState.ENABLED);
- } catch (final SQLException ex) {
- ShardingSpherePreconditions.checkState(forceStart, () -> new UnavailableDataSourceException(actualDataSourceName, ex));
- log.error("Data source unavailable, ignored with the -f parameter.", ex);
- }
- }
-
- /**
- * Get enabled data sources.
- *
- * @param databaseName database name
- * @param databaseConfig database configuration
- * @return enabled data sources
- */
- public Map<String, DataSource> getEnabledDataSources(final String databaseName, final DatabaseConfiguration databaseConfig) {
- Map<String, DataSource> dataSources = databaseConfig.getStorageUnits().entrySet().stream()
- .collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getDataSource(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
- return getEnabledDataSources(databaseName, dataSources);
- }
-
- /**
- * Get enabled data sources.
- *
- * @param databaseName database name
- * @param dataSources data sources
- * @return enabled data sources
- */
- public Map<String, DataSource> getEnabledDataSources(final String databaseName, final Map<String, DataSource> dataSources) {
- if (dataSources.isEmpty() || !initialized.get()) {
- return dataSources;
- }
- Map<String, DataSource> result = filterDisabledDataSources(databaseName, dataSources);
- checkForceConnection(result);
- return result;
- }
-
- private Map<String, DataSource> filterDisabledDataSources(final String databaseName, final Map<String, DataSource> dataSources) {
- Map<String, DataSource> result = new LinkedHashMap<>(dataSources.size(), 1F);
- dataSources.forEach((key, value) -> {
- DataSourceState dataSourceState = dataSourceStates.get(getCacheKey(databaseName, key));
- if (DataSourceState.DISABLED != dataSourceState) {
- result.put(key, value);
- }
- });
- return result;
- }
-
- private void checkForceConnection(final Map<String, DataSource> dataSources) {
- if (forceStart) {
- dataSources.entrySet().removeIf(entry -> {
- try (Connection ignored = entry.getValue().getConnection()) {
- return false;
- } catch (final SQLException ex) {
- log.error("Data source state unavailable, ignored with the -f parameter.", ex);
- return true;
- }
- });
- }
- }
-
- /**
- * Update data source state.
- *
- * @param qualifiedDataSource qualified data source
- * @param dataSourceState data source state
- */
- public void updateState(final QualifiedDataSource qualifiedDataSource, final DataSourceState dataSourceState) {
- dataSourceStates.put(getCacheKey(qualifiedDataSource.getDatabaseName(), qualifiedDataSource.getDataSourceName()), dataSourceState);
- }
-
- private String getCacheKey(final String databaseName, final String dataSourceName) {
- return databaseName + "." + dataSourceName;
- }
-}
diff --git a/infra/common/src/test/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngineTest.java b/infra/common/src/test/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngineTest.java
index 6c62d5db197..bc49a70a10e 100644
--- a/infra/common/src/test/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngineTest.java
+++ b/infra/common/src/test/java/org/apache/shardingsphere/infra/database/DatabaseTypeEngineTest.java
@@ -53,7 +53,7 @@ class DatabaseTypeEngineTest {
void assertGetProtocolTypeFromConfiguredProperties() {
Properties props = PropertiesBuilder.build(new Property(ConfigurationPropertyKey.PROXY_FRONTEND_DATABASE_PROTOCOL_TYPE.getKey(), "MySQL"));
DatabaseConfiguration databaseConfig = new DataSourceProvidedDatabaseConfiguration(Collections.emptyMap(), Collections.singleton(new FixtureRuleConfiguration()));
- assertThat(DatabaseTypeEngine.getProtocolType("sharding_db", databaseConfig, new ConfigurationProperties(props)), instanceOf(MySQLDatabaseType.class));
+ assertThat(DatabaseTypeEngine.getProtocolType(databaseConfig, new ConfigurationProperties(props)), instanceOf(MySQLDatabaseType.class));
assertThat(DatabaseTypeEngine.getProtocolType(Collections.singletonMap("foo_db", databaseConfig), new ConfigurationProperties(props)), instanceOf(MySQLDatabaseType.class));
}
@@ -61,7 +61,7 @@ class DatabaseTypeEngineTest {
void assertGetProtocolTypeFromDataSource() throws SQLException {
DataSource datasource = mockDataSource(TypedSPILoader.getService(DatabaseType.class, "PostgreSQL"));
DatabaseConfiguration databaseConfig = new DataSourceProvidedDatabaseConfiguration(Collections.singletonMap("foo_ds", datasource), Collections.singleton(new FixtureRuleConfiguration()));
- assertThat(DatabaseTypeEngine.getProtocolType("sharding_db", databaseConfig, new ConfigurationProperties(new Properties())), instanceOf(PostgreSQLDatabaseType.class));
+ assertThat(DatabaseTypeEngine.getProtocolType(databaseConfig, new ConfigurationProperties(new Properties())), instanceOf(PostgreSQLDatabaseType.class));
assertThat(DatabaseTypeEngine.getProtocolType(Collections.singletonMap("foo_db", databaseConfig), new ConfigurationProperties(new Properties())), instanceOf(PostgreSQLDatabaseType.class));
}
@@ -69,8 +69,8 @@ class DatabaseTypeEngineTest {
void assertGetStorageTypes() throws SQLException {
DataSource datasource = mockDataSource(TypedSPILoader.getService(DatabaseType.class, "MySQL"));
DatabaseConfiguration databaseConfig = new DataSourceProvidedDatabaseConfiguration(Collections.singletonMap("foo_db", datasource), Collections.singletonList(new FixtureRuleConfiguration()));
- assertTrue(DatabaseTypeEngine.getStorageTypes("foo_db", databaseConfig).containsKey("foo_db"));
- assertThat(DatabaseTypeEngine.getStorageTypes("foo_db", databaseConfig).get("foo_db"), instanceOf(MySQLDatabaseType.class));
+ assertTrue(DatabaseTypeEngine.getStorageTypes(databaseConfig).containsKey("foo_db"));
+ assertThat(DatabaseTypeEngine.getStorageTypes(databaseConfig).get("foo_db"), instanceOf(MySQLDatabaseType.class));
}
@Test
diff --git a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/ExternalMetaDataFactory.java b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/ExternalMetaDataFactory.java
index 70f26940427..3746fd2d408 100644
--- a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/ExternalMetaDataFactory.java
+++ b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/ExternalMetaDataFactory.java
@@ -51,8 +51,8 @@ public final class ExternalMetaDataFactory {
*/
public static ShardingSphereDatabase create(final String databaseName, final DatabaseConfiguration databaseConfig,
final ConfigurationProperties props, final ComputeNodeInstanceContext computeNodeInstanceContext) throws SQLException {
- return ShardingSphereDatabase.create(databaseName, DatabaseTypeEngine.getProtocolType(databaseName, databaseConfig, props),
- DatabaseTypeEngine.getStorageTypes(databaseName, databaseConfig), databaseConfig, props, computeNodeInstanceContext);
+ return ShardingSphereDatabase.create(databaseName, DatabaseTypeEngine.getProtocolType(databaseConfig, props),
+ DatabaseTypeEngine.getStorageTypes(databaseConfig), databaseConfig, props, computeNodeInstanceContext);
}
/**
@@ -81,7 +81,7 @@ public final class ExternalMetaDataFactory {
for (Entry<String, DatabaseConfiguration> entry : databaseConfigMap.entrySet()) {
String databaseName = entry.getKey();
if (!entry.getValue().getStorageUnits().isEmpty() || !systemDatabase.getSystemSchemas().contains(databaseName)) {
- Map<String, DatabaseType> storageTypes = DatabaseTypeEngine.getStorageTypes(entry.getKey(), entry.getValue());
+ Map<String, DatabaseType> storageTypes = DatabaseTypeEngine.getStorageTypes(entry.getValue());
result.put(databaseName.toLowerCase(), ShardingSphereDatabase.create(databaseName, protocolType, storageTypes, entry.getValue(), props, computeNodeInstanceContext));
}
}
diff --git a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/InternalMetaDataFactory.java b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/InternalMetaDataFactory.java
index b12e59cca40..62a85f17021 100644
--- a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/InternalMetaDataFactory.java
+++ b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/factory/InternalMetaDataFactory.java
@@ -49,7 +49,7 @@ public final class InternalMetaDataFactory {
*/
public static ShardingSphereDatabase create(final String databaseName, final MetaDataPersistService persistService, final DatabaseConfiguration databaseConfig,
final ConfigurationProperties props, final ComputeNodeInstanceContext computeNodeInstanceContext) {
- DatabaseType protocolType = DatabaseTypeEngine.getProtocolType(databaseName, databaseConfig, props);
+ DatabaseType protocolType = DatabaseTypeEngine.getProtocolType(databaseConfig, props);
return ShardingSphereDatabase.create(databaseName, protocolType, databaseConfig, computeNodeInstanceContext, persistService.getDatabaseMetaDataService().loadSchemas(databaseName));
}
diff --git a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decorator/SingleRuleConfigurationDecorator.java b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decorator/SingleRuleConfigurationDecorator.java
index e9b2c81838f..b5886bfa591 100644
--- a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decorator/SingleRuleConfigurationDecorator.java
+++ b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/decorator/SingleRuleConfigurationDecorator.java
@@ -24,7 +24,6 @@ import org.apache.shardingsphere.infra.database.core.type.DatabaseTypeRegistry;
import org.apache.shardingsphere.infra.datanode.DataNode;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import org.apache.shardingsphere.single.api.config.SingleRuleConfiguration;
import org.apache.shardingsphere.single.api.constant.SingleTableConstants;
import org.apache.shardingsphere.single.datanode.SingleTableDataNodeLoader;
@@ -62,9 +61,8 @@ public final class SingleRuleConfigurationDecorator implements RuleConfiguration
if (!isExpandRequired(splitTables)) {
return splitTables;
}
- Map<String, DataSource> enabledDataSources = DataSourceStateManager.getInstance().getEnabledDataSources(databaseName, dataSources);
- Map<String, DataSource> aggregatedDataSources = SingleTableLoadUtils.getAggregatedDataSourceMap(enabledDataSources, builtRules);
- DatabaseType databaseType = enabledDataSources.isEmpty() ? DatabaseTypeEngine.getDefaultStorageType() : DatabaseTypeEngine.getStorageType(enabledDataSources.values().iterator().next());
+ Map<String, DataSource> aggregatedDataSources = SingleTableLoadUtils.getAggregatedDataSourceMap(dataSources, builtRules);
+ DatabaseType databaseType = dataSources.isEmpty() ? DatabaseTypeEngine.getDefaultStorageType() : DatabaseTypeEngine.getStorageType(dataSources.values().iterator().next());
Collection<String> excludedTables = SingleTableLoadUtils.getExcludedTables(builtRules);
Map<String, Collection<DataNode>> actualDataNodes = SingleTableDataNodeLoader.load(databaseName, aggregatedDataSources, excludedTables);
Collection<DataNode> configuredDataNodes = SingleTableLoadUtils.convertToDataNodes(databaseName, databaseType, splitTables);
diff --git a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
index 822fcbd2944..9866f43b904 100644
--- a/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
+++ b/kernel/single/core/src/main/java/org/apache/shardingsphere/single/rule/SingleRule.java
@@ -27,9 +27,8 @@ import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
import org.apache.shardingsphere.infra.metadata.database.schema.util.IndexMetaDataUtils;
import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
-import org.apache.shardingsphere.infra.rule.scope.DatabaseRule;
import org.apache.shardingsphere.infra.rule.attribute.RuleAttributes;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
+import org.apache.shardingsphere.infra.rule.scope.DatabaseRule;
import org.apache.shardingsphere.single.api.config.SingleRuleConfiguration;
import org.apache.shardingsphere.single.datanode.SingleTableDataNodeLoader;
import org.apache.shardingsphere.single.util.SingleTableLoadUtils;
@@ -69,8 +68,7 @@ public final class SingleRule implements DatabaseRule {
final DatabaseType protocolType, final Map<String, DataSource> dataSourceMap, final Collection<ShardingSphereRule> builtRules) {
configuration = ruleConfig;
defaultDataSource = ruleConfig.getDefaultDataSource().orElse(null);
- Map<String, DataSource> enabledDataSources = DataSourceStateManager.getInstance().getEnabledDataSources(databaseName, dataSourceMap);
- Map<String, DataSource> aggregateDataSourceMap = SingleTableLoadUtils.getAggregatedDataSourceMap(enabledDataSources, builtRules);
+ Map<String, DataSource> aggregateDataSourceMap = SingleTableLoadUtils.getAggregatedDataSourceMap(dataSourceMap, builtRules);
dataSourceNames = aggregateDataSourceMap.keySet();
this.protocolType = protocolType;
singleTableDataNodes = SingleTableDataNodeLoader.load(databaseName, protocolType, aggregateDataSourceMap, builtRules, configuration.getTables());
diff --git a/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/MetaDataContextsFactory.java b/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/MetaDataContextsFactory.java
index aa0431f6e72..fefaef1ede1 100644
--- a/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/MetaDataContextsFactory.java
+++ b/mode/core/src/main/java/org/apache/shardingsphere/mode/metadata/MetaDataContextsFactory.java
@@ -45,7 +45,6 @@ import org.apache.shardingsphere.infra.metadata.statistics.builder.ShardingSpher
import org.apache.shardingsphere.infra.rule.builder.global.GlobalRulesBuilder;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.state.datasource.DataSourceState;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import org.apache.shardingsphere.metadata.factory.ExternalMetaDataFactory;
import org.apache.shardingsphere.metadata.factory.InternalMetaDataFactory;
import org.apache.shardingsphere.metadata.persist.MetaDataPersistService;
@@ -100,7 +99,6 @@ public final class MetaDataContextsFactory {
Map<String, DatabaseConfiguration> effectiveDatabaseConfigs = isDatabaseMetaDataExisted
? createEffectiveDatabaseConfigurations(getDatabaseNames(computeNodeInstanceContext, param.getDatabaseConfigs(), persistService), param.getDatabaseConfigs(), persistService)
: param.getDatabaseConfigs();
- checkDataSourceStates(effectiveDatabaseConfigs, statusMap, param.isForce());
// TODO load global data sources from persist service
Map<String, DataSource> globalDataSources = param.getGlobalDataSources();
Collection<RuleConfiguration> globalRuleConfigs = isDatabaseMetaDataExisted ? persistService.getGlobalRuleService().load() : param.getGlobalRuleConfigs();
@@ -155,15 +153,6 @@ public final class MetaDataContextsFactory {
}
}
- private static void checkDataSourceStates(final Map<String, DatabaseConfiguration> databaseConfigs, final Map<String, QualifiedDataSourceState> statusMap, final boolean force) {
- Map<String, DataSourceState> storageDataSourceStates = getStorageDataSourceStates(statusMap);
- databaseConfigs.forEach((key, value) -> {
- if (!value.getStorageUnits().isEmpty()) {
- DataSourceStateManager.getInstance().initStates(key, value.getStorageUnits(), storageDataSourceStates, force);
- }
- });
- }
-
private static Map<String, DataSourceState> getStorageDataSourceStates(final Map<String, QualifiedDataSourceState> statusMap) {
Map<String, DataSourceState> result = new HashMap<>(statusMap.size(), 1F);
statusMap.forEach((key, value) -> {
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
index f122e1be841..353c83650b8 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/StateChangedSubscriber.java
@@ -22,8 +22,6 @@ import org.apache.shardingsphere.infra.instance.ComputeNodeInstance;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedDataSource;
import org.apache.shardingsphere.infra.rule.attribute.datasource.StaticDataSourceRuleAttribute;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceState;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import org.apache.shardingsphere.infra.util.eventbus.EventSubscriber;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.cluster.event.ClusterStateEvent;
@@ -57,7 +55,6 @@ public final class StateChangedSubscriber implements EventSubscriber {
if (!contextManager.getMetaDataContexts().getMetaData().containsDatabase(qualifiedDataSource.getDatabaseName())) {
return;
}
- DataSourceStateManager.getInstance().updateState(qualifiedDataSource, DataSourceState.valueOf(event.getStatus().getStatus().name()));
ShardingSphereDatabase database = contextManager.getMetaDataContexts().getMetaData().getDatabase(qualifiedDataSource.getDatabaseName());
for (StaticDataSourceRuleAttribute each : database.getRuleMetaData().getAttributes(StaticDataSourceRuleAttribute.class)) {
each.updateStatus(qualifiedDataSource, event.getStatus().getStatus());
diff --git a/proxy/bootstrap/src/main/java/org/apache/shardingsphere/proxy/version/ShardingSphereProxyVersion.java b/proxy/bootstrap/src/main/java/org/apache/shardingsphere/proxy/version/ShardingSphereProxyVersion.java
index 89e84985120..653e46946e4 100644
--- a/proxy/bootstrap/src/main/java/org/apache/shardingsphere/proxy/version/ShardingSphereProxyVersion.java
+++ b/proxy/bootstrap/src/main/java/org/apache/shardingsphere/proxy/version/ShardingSphereProxyVersion.java
@@ -27,7 +27,6 @@ import org.apache.shardingsphere.infra.autogen.version.ShardingSphereVersion;
import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.resource.ResourceMetaData;
-import org.apache.shardingsphere.infra.state.datasource.DataSourceStateManager;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.proxy.database.DatabaseServerInfo;
@@ -66,7 +65,7 @@ public final class ShardingSphereProxyVersion {
}
private static void setDatabaseVersion(final ShardingSphereDatabase database) {
- Optional<DataSource> dataSource = findDataSourceByProtocolType(database.getName(), database.getResourceMetaData(), database.getProtocolType());
+ Optional<DataSource> dataSource = findDataSourceByProtocolType(database.getResourceMetaData(), database.getProtocolType());
if (!dataSource.isPresent()) {
return;
}
@@ -75,11 +74,11 @@ public final class ShardingSphereProxyVersion {
DatabaseProtocolServerInfo.setProtocolVersion(database.getName(), databaseServerInfo.getDatabaseVersion());
}
- private static Optional<DataSource> findDataSourceByProtocolType(final String databaseName, final ResourceMetaData resourceMetaData, final DatabaseType protocolType) {
+ private static Optional<DataSource> findDataSourceByProtocolType(final ResourceMetaData resourceMetaData, final DatabaseType protocolType) {
Optional<String> dataSourceName = resourceMetaData.getStorageUnits().entrySet()
.stream().filter(entry -> entry.getValue().getStorageType().equals(protocolType)).map(Entry::getKey).findFirst();
Map<String, DataSource> dataSources = resourceMetaData.getStorageUnits().entrySet().stream()
.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getDataSource(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
- return dataSourceName.flatMap(optional -> Optional.ofNullable(DataSourceStateManager.getInstance().getEnabledDataSources(databaseName, dataSources).get(optional)));
+ return dataSourceName.flatMap(optional -> Optional.ofNullable(dataSources.get(optional)));
}
}