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 <[email protected]>
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)));
}
}
