This is an automated email from the ASF dual-hosted git repository.
terrymanu 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 8cf85f9cc9f Refactor pipeline data source configuration boundary
(#39921)
8cf85f9cc9f is described below
commit 8cf85f9cc9f401bdae49c31324fb90e7b059d577
Author: Liang Zhang <[email protected]>
AuthorDate: Tue Sep 22 16:26:26 2026 +0800
Refactor pipeline data source configuration boundary (#39921)
---
.../api/PipelineDataSourceConfiguration.java | 3 +-
...rdingSpherePipelineDataSourceConfiguration.java | 120 +++++++++++++++++----
.../pipeline/spi/PipelineDataSourceCreator.java | 3 +-
...gSpherePipelineDataSourceConfigurationTest.java | 76 +++++++++++--
.../ShardingSpherePipelineDataSourceCreator.java | 55 ++++------
.../PipelineRuleConfigurationReviser.java} | 18 ++--
.../IncrementalTaskPositionManager.java | 3 +-
.../util/PipelineDataSourceConfigurationUtils.java | 6 +-
...hardingSpherePipelineDataSourceCreatorTest.java | 112 +++++++++++++++++++
.../StandardPipelineDataSourceCreatorTest.java | 46 ++++++++
.../PipelineDataSourceConfigurationUtilsTest.java | 15 +--
.../ShardingPipelineRuleConfigurationReviser.java | 66 ++++++++++++
...ardingPipelineYamlRuleConfigurationReviser.java | 64 -----------
...tasource.rule.PipelineRuleConfigurationReviser} | 6 +-
...ardingPipelineRuleConfigurationReviserTest.java | 96 +++++++++++++++++
...ngPipelineYamlRuleConfigurationReviserTest.java | 94 ----------------
.../shardingsphere/data/pipeline/cdc/CDCJob.java | 3 +-
.../data/pipeline/cdc/api/CDCJobAPI.java | 16 +--
.../data/pipeline/cdc/CDCJobTest.java | 22 +---
.../data/pipeline/cdc/api/CDCJobAPITest.java | 4 +
.../migration/MigrationJobExecutorCallback.java | 4 +-
.../MigrationJobExecutorCallbackTest.java | 24 ++---
.../pipeline/core/util/PipelineContextUtils.java | 15 ++-
23 files changed, 562 insertions(+), 309 deletions(-)
diff --git
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineDataSourceConfiguration.java
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineDataSourceConfiguration.java
index c99f69084b1..3760181ff88 100644
---
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineDataSourceConfiguration.java
+++
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/PipelineDataSourceConfiguration.java
@@ -39,7 +39,8 @@ public interface PipelineDataSourceConfiguration {
String getParameter();
/**
- * Get data source configuration, related to {@link #getParameter()}.
+ * Get the runtime data source configuration derived from {@link
#getParameter()}.
+ * The returned object must not be a YAML configuration object.
*
* @return data source configuration
*/
diff --git
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfiguration.java
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfiguration.java
index 4a6a8591e27..03ffb396c6d 100644
---
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfiguration.java
+++
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfiguration.java
@@ -28,14 +28,20 @@ import
org.apache.shardingsphere.database.connector.core.jdbcurl.parser.Standard
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeFactory;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
+import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Optional;
@@ -52,12 +58,16 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
private final String parameter;
- private final YamlRootConfiguration rootConfig;
+ private final String databaseName;
+
+ private volatile Map<String, DataSourcePoolProperties>
dataSourcePoolPropertiesMap;
+
+ private volatile Collection<RuleConfiguration> ruleConfigurations;
private final DatabaseType databaseType;
public ShardingSpherePipelineDataSourceConfiguration(final String param) {
- rootConfig = YamlEngine.unmarshal(param, YamlRootConfiguration.class,
true);
+ YamlRootConfiguration rootConfig = YamlEngine.unmarshal(param,
YamlRootConfiguration.class, true);
// Need remove dataSourceProperties, because if the parameter at
dataSourceProperties will override parameter at jdbcUrl
for (Map<String, Object> each : rootConfig.getDataSources().values()) {
each.remove("dataSourceProperties");
@@ -65,8 +75,7 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
parameter = YamlEngine.marshal(rootConfig);
Map<String, Object> props =
rootConfig.getDataSources().values().iterator().next();
databaseType = DatabaseTypeFactory.get(getJdbcUrl(props));
- appendJdbcQueryProperties();
- adjustDataSourcePoolProperties(rootConfig.getDataSources());
+ databaseName = rootConfig.getDatabaseName();
}
public ShardingSpherePipelineDataSourceConfiguration(final
YamlRootConfiguration rootConfig) {
@@ -88,32 +97,90 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
return result.toString();
}
- private void appendJdbcQueryProperties() {
- Optional<JdbcQueryPropertiesExtension> extension =
DatabaseTypedSPILoader.findService(JdbcQueryPropertiesExtension.class,
databaseType);
- if (!extension.isPresent()) {
- return;
+ private void appendJdbcQueryProperties(final Map<String, Object>
dataSourceProps, final JdbcQueryPropertiesExtension extension) {
+ String jdbcUrlKey = dataSourceProps.containsKey("url") ? "url" :
"jdbcUrl";
+ String jdbcUrl = dataSourceProps.get(jdbcUrlKey).toString();
+ Properties queryProps = new
StandardJdbcUrlParser().parseQueryProperties(jdbcUrl.contains("?") ?
jdbcUrl.substring(jdbcUrl.indexOf("?") + 1) : "");
+ extension.extendQueryProperties(queryProps);
+ dataSourceProps.put(jdbcUrlKey, new
JdbcUrlAppender().appendQueryProperties(jdbcUrl, queryProps));
+ }
+
+ private DataSourcePoolProperties adjustDataSourcePoolProperties(final
DataSourcePoolProperties dataSourcePoolProps, final
Optional<JdbcQueryPropertiesExtension> extension,
+ final
Map<String, Object> customPoolProps) {
+ Map<String, Object> localProps =
dataSourcePoolProps.getAllLocalProperties();
+ extension.ifPresent(optional -> appendJdbcQueryProperties(localProps,
optional));
+ for (String each : Arrays.asList("minPoolSize", "minimumIdle")) {
+ localProps.put(each, "1");
}
- StandardJdbcUrlParser standardJdbcUrlParser = new
StandardJdbcUrlParser();
- rootConfig.getDataSources().forEach((key, value) -> {
- String jdbcUrlKey = value.containsKey("url") ? "url" : "jdbcUrl";
- String jdbcUrl = value.get(jdbcUrlKey).toString();
- Properties queryProps =
standardJdbcUrlParser.parseQueryProperties(jdbcUrl.contains("?") ?
jdbcUrl.substring(jdbcUrl.indexOf("?") + 1) : "");
- extension.get().extendQueryProperties(queryProps);
- value.replace(jdbcUrlKey, new
JdbcUrlAppender().appendQueryProperties(jdbcUrl, queryProps));
+ localProps.putAll(customPoolProps);
+ return new
DataSourcePoolProperties(dataSourcePoolProps.getPoolClassName(), localProps);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, DataSourcePoolProperties>
createDataSourcePoolPropertiesMap(final Map<String, Map<String, Object>>
dataSources) {
+ Map<String, DataSourcePoolProperties> result = new
LinkedHashMap<>(dataSources.size(), 1F);
+ YamlDataSourceConfigurationSwapper swapper = new
YamlDataSourceConfigurationSwapper();
+ Optional<JdbcQueryPropertiesExtension> extension =
DatabaseTypedSPILoader.findService(JdbcQueryPropertiesExtension.class,
databaseType);
+ dataSources.forEach((key, value) -> {
+ Map<String, Object> customPoolProps = (Map<String, Object>)
value.get("customPoolProps");
+ result.put(key,
adjustDataSourcePoolProperties(swapper.swapToDataSourcePoolProperties(value),
extension,
+ null == customPoolProps ? Collections.emptyMap() :
customPoolProps));
});
+ return result;
}
- private void adjustDataSourcePoolProperties(final Map<String, Map<String,
Object>> dataSources) {
- for (Map<String, Object> queryProps : dataSources.values()) {
- for (String each : Arrays.asList("minPoolSize", "minimumIdle")) {
- queryProps.put(each, "1");
+ /**
+ * Get data source pool properties.
+ *
+ * @return data source pool properties
+ */
+ public Map<String, DataSourcePoolProperties>
getDataSourcePoolPropertiesMap() {
+ Map<String, DataSourcePoolProperties> result =
dataSourcePoolPropertiesMap;
+ if (null == result) {
+ synchronized (this) {
+ result = dataSourcePoolPropertiesMap;
+ if (null == result) {
+ YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(parameter, YamlRootConfiguration.class, true);
+ result =
createDataSourcePoolPropertiesMap(rootConfig.getDataSources());
+ dataSourcePoolPropertiesMap = result;
+ }
}
}
+ return result;
+ }
+
+ /**
+ * Get rule configurations.
+ *
+ * @return rule configurations
+ */
+ public Collection<RuleConfiguration> getRuleConfigurations() {
+ Collection<RuleConfiguration> result = ruleConfigurations;
+ if (null == result) {
+ synchronized (this) {
+ result = ruleConfigurations;
+ if (null == result) {
+ result = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(YamlEngine.unmarshal(parameter,
YamlRootConfiguration.class, true).getRules());
+ ruleConfigurations = result;
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Get rule configurations for data source creation.
+ *
+ * @return independent rule configurations
+ */
+ public Collection<RuleConfiguration> getCreationRuleConfigurations() {
+ return new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(YamlEngine.unmarshal(parameter,
YamlRootConfiguration.class, true).getRules());
}
@Override
public Object getDataSourceConfiguration() {
- return rootConfig;
+ getDataSourcePoolPropertiesMap();
+ return this;
}
@Override
@@ -128,9 +195,16 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
* @return actual data source configuration
*/
public StandardPipelineDataSourceConfiguration
getActualDataSourceConfiguration(final String actualDataSourceName) {
- Map<String, Object> yamlDataSourceConfig =
rootConfig.getDataSources().get(actualDataSourceName);
- Preconditions.checkNotNull(yamlDataSourceConfig, "actualDataSourceName
'{}' does not exist", actualDataSourceName);
- return new
StandardPipelineDataSourceConfiguration(yamlDataSourceConfig);
+ YamlRootConfiguration rootConfig = YamlEngine.unmarshal(parameter,
YamlRootConfiguration.class, true);
+ Map<String, Object> dataSourceConfig =
rootConfig.getDataSources().get(actualDataSourceName);
+ Preconditions.checkNotNull(dataSourceConfig, "actualDataSourceName
'{}' does not exist", actualDataSourceName);
+ StandardPipelineDataSourceConfiguration adjusted = new
StandardPipelineDataSourceConfiguration(dataSourceConfig);
+ String jdbcUrlKey = dataSourceConfig.containsKey("url") ? "url" :
"jdbcUrl";
+ dataSourceConfig.put(jdbcUrlKey, adjusted.getUrl());
+ for (String each : Arrays.asList("minPoolSize", "minimumIdle")) {
+ dataSourceConfig.put(each, "1");
+ }
+ return new StandardPipelineDataSourceConfiguration(dataSourceConfig);
}
/**
diff --git
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/PipelineDataSourceCreator.java
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/PipelineDataSourceCreator.java
index 56d2f528b92..a23d8b978b6 100644
---
a/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/PipelineDataSourceCreator.java
+++
b/kernel/data-pipeline/api/src/main/java/org/apache/shardingsphere/data/pipeline/spi/PipelineDataSourceCreator.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.data.pipeline.spi;
+import
org.apache.shardingsphere.data.pipeline.api.PipelineDataSourceConfiguration;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;
@@ -32,7 +33,7 @@ public interface PipelineDataSourceCreator extends TypedSPI {
/**
* Create pipeline data source.
*
- * @param dataSourceConfig data source configuration
+ * @param dataSourceConfig runtime configuration returned by {@link
PipelineDataSourceConfiguration#getDataSourceConfiguration()}
* @return pipeline data source
* @throws SQLException if create data source failed
*/
diff --git
a/kernel/data-pipeline/api/src/test/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfigurationTest.java
b/kernel/data-pipeline/api/src/test/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfigurationTest.java
index f02f2e6f9cf..af73f44e4b5 100644
---
a/kernel/data-pipeline/api/src/test/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfigurationTest.java
+++
b/kernel/data-pipeline/api/src/test/java/org/apache/shardingsphere/data/pipeline/api/type/ShardingSpherePipelineDataSourceConfigurationTest.java
@@ -18,8 +18,10 @@
package org.apache.shardingsphere.data.pipeline.api.type;
import
org.apache.shardingsphere.data.pipeline.spi.JdbcQueryPropertiesExtension;
+import
org.apache.shardingsphere.database.connector.core.jdbcurl.appender.JdbcUrlAppender;
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.file.SystemResourceFileUtils;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
@@ -28,19 +30,26 @@ import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExt
import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.MockedConstruction;
-import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.Properties;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.isA;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.when;
@ExtendWith(AutoMockExtension.class)
@@ -54,12 +63,16 @@ class ShardingSpherePipelineDataSourceConfigurationTest {
JdbcQueryPropertiesExtension queryPropsExtension =
mock(JdbcQueryPropertiesExtension.class);
when(DatabaseTypedSPILoader.findService(JdbcQueryPropertiesExtension.class,
databaseType)).thenReturn(Optional.of(queryPropsExtension));
YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"),
YamlRootConfiguration.class, true);
+ rootConfig.getDataSources().get("ds_0").put("dataSourceProperties",
Collections.singletonMap("foo", "bar"));
Map<String, Object> backupDataSource0 = new
HashMap<>(rootConfig.getDataSources().get("ds_0"));
Map<String, Object> backupDataSource1 = new
HashMap<>(rootConfig.getDataSources().get("ds_1"));
ShardingSpherePipelineDataSourceConfiguration actual = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig);
assertParameterUnchanged(backupDataSource0,
rootConfig.getDataSources().get("ds_0"));
assertParameterUnchanged(backupDataSource1,
rootConfig.getDataSources().get("ds_1"));
assertPipelineDataSourceConfiguration(actual);
+ YamlRootConfiguration persisted =
YamlEngine.unmarshal(actual.getParameter(), YamlRootConfiguration.class, true);
+
assertFalse(persisted.getDataSources().get("ds_0").containsKey("dataSourceProperties"));
+ assertThat(persisted.getDataSources().get("ds_0").get("minPoolSize"),
is(20));
}
private void assertParameterUnchanged(final Map<String, Object>
backupDataSource, final Map<String, Object> handledDataSource) {
@@ -74,24 +87,73 @@ class ShardingSpherePipelineDataSourceConfigurationTest {
private void assertPipelineDataSourceConfiguration(final
ShardingSpherePipelineDataSourceConfiguration actual) {
assertThat(actual.getDatabaseType().getType(), is("FIXTURE"));
assertThat(actual.getType(),
is(ShardingSpherePipelineDataSourceConfiguration.TYPE));
- assertThat(actual.getDataSourceConfiguration(),
isA(YamlRootConfiguration.class));
- Map<String, Map<String, Object>> dataSources =
actual.getRootConfig().getDataSources();
+ assertThat(actual.getDataSourceConfiguration(),
isA(ShardingSpherePipelineDataSourceConfiguration.class));
+ Map<String, DataSourcePoolProperties> dataSources =
actual.getDataSourcePoolPropertiesMap();
assertThat(dataSources.size(), is(2));
assertTrue(dataSources.containsKey("ds_0"));
assertTrue(dataSources.containsKey("ds_1"));
- for (Map<String, Object> queryProps : dataSources.values()) {
- for (String each : Arrays.asList("minPoolSize", "minimumIdle")) {
- assertThat(queryProps.get(each), is("1"));
- }
+ for (DataSourcePoolProperties dataSourcePoolProps :
dataSources.values()) {
+
assertThat(dataSourcePoolProps.getPoolPropertySynonyms().getStandardProperties().get("minPoolSize"),
is("1"));
+
assertThat(dataSourcePoolProps.getAllLocalProperties().get("minimumIdle"),
is("1"));
}
}
@Test
void assertGetActualDataSourceConfiguration() {
YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"),
YamlRootConfiguration.class, true);
+ rootConfig.getDataSources().get("ds_0").put("customPoolProps",
Collections.singletonMap("connectionTimeout", 30000));
ShardingSpherePipelineDataSourceConfiguration config = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig);
StandardPipelineDataSourceConfiguration actual =
config.getActualDataSourceConfiguration("ds_0");
assertThat(actual.getDatabaseType().getType(), is("FIXTURE"));
assertThat(actual.getUrl(), is("jdbc:mock://127.0.0.1/ds_0"));
+ Map<?, ?> parameter = YamlEngine.unmarshal(actual.getParameter(),
Map.class);
+ assertThat(((Map<?, ?>)
parameter.get("customPoolProps")).get("connectionTimeout"), is(30000));
+ }
+
+ @Test
+ void assertGetActualDataSourceConfigurationWithJdbcUrl() {
+ YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"),
YamlRootConfiguration.class, true);
+ Map<String, Object> dataSource =
rootConfig.getDataSources().get("ds_0");
+ dataSource.put("jdbcUrl", dataSource.remove("url"));
+ StandardPipelineDataSourceConfiguration actual = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig).getActualDataSourceConfiguration("ds_0");
+ assertThat(actual.getUrl(), is("jdbc:mock://127.0.0.1/ds_0"));
+ Map<?, ?> parameter = YamlEngine.unmarshal(actual.getParameter(),
Map.class);
+ assertThat(parameter.get("jdbcUrl"), is("jdbc:mock://127.0.0.1/ds_0"));
+ }
+
+ @Test
+ void assertCustomPoolPropertiesOverrideAdjustedPoolSize() {
+ YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"),
YamlRootConfiguration.class, true);
+ rootConfig.getDataSources().get("ds_0").put("customPoolProps",
Collections.singletonMap("minimumIdle", 7));
+ ShardingSpherePipelineDataSourceConfiguration config = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig);
+
assertThat(config.getDataSourcePoolPropertiesMap().get("ds_0").getAllLocalProperties().get("minimumIdle"),
is(7));
+ }
+
+ @Test
+ void assertAppendJdbcQueryPropertiesAfterPersistingParameter() {
+ JdbcQueryPropertiesExtension queryPropsExtension =
mock(JdbcQueryPropertiesExtension.class);
+ doAnswer(invocation -> {
+ Properties props = invocation.getArgument(0);
+ props.setProperty("foo", "bar");
+ return null;
+
}).when(queryPropsExtension).extendQueryProperties(any(Properties.class));
+
when(DatabaseTypedSPILoader.findService(JdbcQueryPropertiesExtension.class,
databaseType)).thenReturn(Optional.of(queryPropsExtension));
+ try (
+ MockedConstruction<JdbcUrlAppender> ignored =
mockConstruction(JdbcUrlAppender.class,
+ (mock, context) ->
when(mock.appendQueryProperties(anyString(),
any(Properties.class))).thenAnswer(invocation -> invocation.getArgument(0) +
"?foo=bar"))) {
+ ShardingSpherePipelineDataSourceConfiguration actual = new
ShardingSpherePipelineDataSourceConfiguration(
+
SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"));
+
assertThat(actual.getDataSourcePoolPropertiesMap().get("ds_0").getConnectionPropertySynonyms().getStandardProperties().get("url"),
is("jdbc:mock://127.0.0.1/ds_0?foo=bar"));
+ YamlRootConfiguration persisted =
YamlEngine.unmarshal(actual.getParameter(), YamlRootConfiguration.class, true);
+ assertThat(persisted.getDataSources().get("ds_0").get("url"),
is("jdbc:mock://127.0.0.1/ds_0"));
+
assertThat(actual.getActualDataSourceConfiguration("ds_0").getUrl(),
is("jdbc:mock://127.0.0.1/ds_0?foo=bar?foo=bar"));
+ YamlRootConfiguration jdbcUrlConfig =
YamlEngine.unmarshal(SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"),
+ YamlRootConfiguration.class, true);
+ jdbcUrlConfig.getDataSources().get("ds_0").put("jdbcUrl",
jdbcUrlConfig.getDataSources().get("ds_0").remove("url"));
+ ShardingSpherePipelineDataSourceConfiguration jdbcUrlActual = new
ShardingSpherePipelineDataSourceConfiguration(jdbcUrlConfig);
+
assertThat(jdbcUrlActual.getDataSourcePoolPropertiesMap().get("ds_0").getAllLocalProperties().get("jdbcUrl"),
+ is("jdbc:mock://127.0.0.1/ds_0?foo=bar"));
+
assertThat(jdbcUrlActual.getActualDataSourceConfiguration("ds_0").getUrl(),
is("jdbc:mock://127.0.0.1/ds_0?foo=bar?foo=bar"));
+ }
}
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
index 01e447bd8b9..f8a050f019b 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreator.java
@@ -17,26 +17,22 @@
package org.apache.shardingsphere.data.pipeline.core.datasource.creator;
-import
org.apache.shardingsphere.authority.yaml.config.YamlAuthorityRuleConfiguration;
+import org.apache.shardingsphere.authority.config.AuthorityRuleConfiguration;
import
org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDataSourceConfiguration;
import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
-import
org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser;
+import
org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser;
import org.apache.shardingsphere.data.pipeline.spi.PipelineDataSourceCreator;
import org.apache.shardingsphere.driver.api.ShardingSphereDataSourceFactory;
import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey;
import
org.apache.shardingsphere.infra.config.props.temporary.TemporaryConfigurationPropertyKey;
import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
+import
org.apache.shardingsphere.infra.datasource.pool.creator.DataSourcePoolCreator;
import
org.apache.shardingsphere.infra.datasource.pool.destroyer.DataSourcePoolDestroyer;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
-import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
-import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.mode.YamlModeConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.mode.YamlPersistRepositoryConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.mode.YamlModeConfigurationSwapper;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.mode.repository.standalone.StandalonePersistRepositoryConfiguration;
import javax.sql.DataSource;
import java.sql.SQLException;
@@ -53,16 +49,15 @@ public final class ShardingSpherePipelineDataSourceCreator
implements PipelineDa
@Override
public DataSource create(final Object dataSourceConfig) throws
SQLException {
- YamlRootConfiguration yamlRootConfig =
YamlEngine.unmarshal(YamlEngine.marshal(dataSourceConfig),
YamlRootConfiguration.class);
- removeAuthorityRuleConfiguration(yamlRootConfig);
- yamlRootConfig.setProps(createConfigurationProperties());
- reviseYamlRuleConfiguration(yamlRootConfig);
- yamlRootConfig.setMode(createStandaloneModeConfiguration());
- return createShardingSphereDataSource(yamlRootConfig);
+ ShardingSpherePipelineDataSourceConfiguration config =
(ShardingSpherePipelineDataSourceConfiguration) dataSourceConfig;
+ Collection<RuleConfiguration> ruleConfigs =
config.getCreationRuleConfigurations();
+ removeAuthorityRuleConfiguration(ruleConfigs);
+ reviseRuleConfiguration(ruleConfigs);
+ return createShardingSphereDataSource(config, ruleConfigs);
}
- private void removeAuthorityRuleConfiguration(final YamlRootConfiguration
yamlRootConfig) {
-
yamlRootConfig.getRules().removeIf(YamlAuthorityRuleConfiguration.class::isInstance);
+ private void removeAuthorityRuleConfiguration(final
Collection<RuleConfiguration> ruleConfigs) {
+ ruleConfigs.removeIf(AuthorityRuleConfiguration.class::isInstance);
}
private Properties createConfigurationProperties() {
@@ -96,23 +91,19 @@ public final class ShardingSpherePipelineDataSourceCreator
implements PipelineDa
}
@SuppressWarnings("unchecked")
- private void reviseYamlRuleConfiguration(final YamlRootConfiguration
yamlRootConfig) {
-
OrderedSPILoader.getServices(PipelineYamlRuleConfigurationReviser.class,
yamlRootConfig.getRules()).forEach((key, value) -> value.revise(key));
+ private void reviseRuleConfiguration(final Collection<RuleConfiguration>
ruleConfigs) {
+ OrderedSPILoader.getServices(PipelineRuleConfigurationReviser.class,
ruleConfigs).forEach((key, value) -> value.revise(key));
}
- private YamlModeConfiguration createStandaloneModeConfiguration() {
- YamlModeConfiguration result = new YamlModeConfiguration();
- result.setType("Standalone");
- YamlPersistRepositoryConfiguration yamlRepositoryConfig = new
YamlPersistRepositoryConfiguration();
- yamlRepositoryConfig.setType("Memory");
- result.setRepository(yamlRepositoryConfig);
- return result;
+ private ModeConfiguration createStandaloneModeConfiguration() {
+ return new ModeConfiguration("Standalone", new
StandalonePersistRepositoryConfiguration("Memory", new Properties()));
}
- private DataSource createShardingSphereDataSource(final
YamlRootConfiguration yamlRootConfig) throws SQLException {
- Map<String, DataSource> dataSourceMap = new
YamlDataSourceConfigurationSwapper().swapToDataSources(yamlRootConfig.getDataSources(),
false);
+ private DataSource createShardingSphereDataSource(final
ShardingSpherePipelineDataSourceConfiguration config, final
Collection<RuleConfiguration> ruleConfigs) throws SQLException {
+ Map<String, DataSourcePoolProperties> dataSourcePropsMap =
config.getDataSourcePoolPropertiesMap();
+ Map<String, DataSource> dataSourceMap =
DataSourcePoolCreator.create(dataSourcePropsMap, false);
try {
- return createShardingSphereDataSource(dataSourceMap,
yamlRootConfig);
+ return
ShardingSphereDataSourceFactory.createDataSource(config.getDatabaseName(),
createStandaloneModeConfiguration(), dataSourceMap, ruleConfigs,
createConfigurationProperties());
// CHECKSTYLE:OFF
} catch (final SQLException | RuntimeException ex) {
// CHECKSTYLE:ON
@@ -121,12 +112,6 @@ public final class ShardingSpherePipelineDataSourceCreator
implements PipelineDa
}
}
- private DataSource createShardingSphereDataSource(final Map<String,
DataSource> dataSourceMap, final YamlRootConfiguration yamlRootConfig) throws
SQLException {
- ModeConfiguration modeConfig = new
YamlModeConfigurationSwapper().swapToObject(yamlRootConfig.getMode());
- Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(yamlRootConfig.getRules());
- return
ShardingSphereDataSourceFactory.createDataSource(yamlRootConfig.getDatabaseName(),
modeConfig, dataSourceMap, ruleConfigs, yamlRootConfig.getProps());
- }
-
@Override
public String getType() {
return ShardingSpherePipelineDataSourceConfiguration.TYPE;
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/yaml/PipelineYamlRuleConfigurationReviser.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/rule/PipelineRuleConfigurationReviser.java
similarity index 70%
rename from
kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/yaml/PipelineYamlRuleConfigurationReviser.java
rename to
kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/rule/PipelineRuleConfigurationReviser.java
index 59789237bce..78ff4326d7a 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/yaml/PipelineYamlRuleConfigurationReviser.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/datasource/rule/PipelineRuleConfigurationReviser.java
@@ -15,24 +15,24 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.data.pipeline.core.datasource.yaml;
+package org.apache.shardingsphere.data.pipeline.core.datasource.rule;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPI;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
/**
- * Pipeline YAML rule configuration reviser.
- *
- * @param <T> type of YAML rule configuration
+ * Pipeline rule configuration reviser.
+ *
+ * @param <T> type of rule configuration
*/
@SingletonSPI
-public interface PipelineYamlRuleConfigurationReviser<T extends
YamlRuleConfiguration> extends OrderedSPI<T> {
+public interface PipelineRuleConfigurationReviser<T extends RuleConfiguration>
extends OrderedSPI<T> {
/**
- * Revise YAML rule configuration.
+ * Revise rule configuration.
*
- * @param yamlRuleConfig to be revised YAML rule configuration
+ * @param ruleConfig rule configuration to revise
*/
- void revise(T yamlRuleConfig);
+ void revise(T ruleConfig);
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/incremental/IncrementalTaskPositionManager.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/incremental/IncrementalTaskPositionManager.java
index 1c83b05a6fd..a3e5deff2fa 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/incremental/IncrementalTaskPositionManager.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/preparer/incremental/IncrementalTaskPositionManager.java
@@ -31,7 +31,6 @@ import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoa
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.infra.datasource.pool.creator.DataSourcePoolCreator;
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
import java.sql.SQLException;
import java.util.Optional;
@@ -90,7 +89,7 @@ public final class IncrementalTaskPositionManager {
}
private void destroyPosition(final String jobId, final
ShardingSpherePipelineDataSourceConfiguration pipelineDataSourceConfig) throws
SQLException {
- for (DataSourcePoolProperties each : new
YamlDataSourceConfigurationSwapper().getDataSourcePoolPropertiesMap(pipelineDataSourceConfig.getRootConfig()).values())
{
+ for (DataSourcePoolProperties each :
pipelineDataSourceConfig.getDataSourcePoolPropertiesMap().values()) {
try (PipelineDataSource dataSource = new
PipelineDataSource(DataSourcePoolCreator.create(each), databaseType)) {
dialectPositionManager.destroy(dataSource, jobId);
}
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtils.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtils.java
index bcb4ccc070a..16f0a4e0fa5 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtils.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtils.java
@@ -111,15 +111,17 @@ public final class PipelineDataSourceConfigurationUtils {
if (null == storageUnits || storageUnits.isEmpty()) {
return pipelineDataSourceConfig;
}
- for (Entry<String, Map<String, Object>> entry :
pipelineDataSourceConfig.getRootConfig().getDataSources().entrySet()) {
+ for (Entry<String, DataSourcePoolProperties> entry :
pipelineDataSourceConfig.getDataSourcePoolPropertiesMap().entrySet()) {
StorageUnit storageUnit = storageUnits.get(entry.getKey());
if (null == storageUnit) {
continue;
}
- Map<String, Object> jobDataSourceProps = entry.getValue();
+ DataSourcePoolProperties dataSourcePoolProps = entry.getValue();
+ Map<String, Object> jobDataSourceProps =
dataSourcePoolProps.getAllLocalProperties();
Map<String, Object> storageUnitStandardProps =
storageUnit.getDataSourcePoolProperties().getPoolPropertySynonyms().getStandardProperties();
logTransformPoolSize(jobId, entry.getKey(), jobDataSourceProps,
storageUnitStandardProps);
transformPoolSize(jobDataSourceProps, storageUnitStandardProps);
+ entry.setValue(new
DataSourcePoolProperties(dataSourcePoolProps.getPoolClassName(),
jobDataSourceProps));
}
return pipelineDataSourceConfig;
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreatorTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreatorTest.java
new file mode 100644
index 00000000000..3eccdec1587
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/ShardingSpherePipelineDataSourceCreatorTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.data.pipeline.core.datasource.creator;
+
+import org.apache.shardingsphere.authority.config.AuthorityRuleConfiguration;
+import
org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDataSourceConfiguration;
+import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManager;
+import org.apache.shardingsphere.driver.api.ShardingSphereDataSourceFactory;
+import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
+import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
+import
org.apache.shardingsphere.infra.datasource.pool.creator.DataSourcePoolCreator;
+import
org.apache.shardingsphere.infra.datasource.pool.destroyer.DataSourcePoolDestroyer;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
+import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.MockedConstruction;
+
+import javax.sql.DataSource;
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(AutoMockExtension.class)
+@StaticMockSettings({DataSourcePoolCreator.class,
ShardingSphereDataSourceFactory.class, PipelineContextManager.class})
+class ShardingSpherePipelineDataSourceCreatorTest {
+
+ @Test
+ void assertCreateWithProjectConfigurations() throws SQLException {
+ ShardingSpherePipelineDataSourceConfiguration config =
mockConfiguration();
+ DataSource expected = new MockedDataSource();
+ Map<String, DataSource> dataSources = Collections.singletonMap("ds_0",
new MockedDataSource());
+
when(DataSourcePoolCreator.create(config.getDataSourcePoolPropertiesMap(),
false)).thenReturn(dataSources);
+ AtomicReference<ModeConfiguration> actualMode = new
AtomicReference<>();
+ AtomicReference<Collection<RuleConfiguration>> actualRules = new
AtomicReference<>();
+ AtomicReference<Properties> actualProps = new AtomicReference<>();
+ when(ShardingSphereDataSourceFactory.createDataSource(eq("foo_db"),
any(ModeConfiguration.class), eq(dataSources), anyCollection(),
any(Properties.class)))
+ .thenAnswer(invocation -> {
+ actualMode.set(invocation.getArgument(1));
+ actualRules.set(invocation.getArgument(3));
+ actualProps.set(invocation.getArgument(4));
+ return expected;
+ });
+ DataSource actual = new
ShardingSpherePipelineDataSourceCreator().create(config);
+ assertThat(actual, sameInstance(expected));
+ assertThat(actualMode.get().getType(), is("Standalone"));
+ assertThat(actualMode.get().getRepository().getType(), is("Memory"));
+ assertThat(actualRules.get().size(), is(1));
+
assertTrue(actualRules.get().stream().noneMatch(AuthorityRuleConfiguration.class::isInstance));
+ assertThat(actualProps.get().get("max-connections-size-per-query"),
is(100000));
+ }
+
+ @Test
+ void assertCreateCleansUpDataSourcesOnFailure() throws SQLException {
+ ShardingSpherePipelineDataSourceConfiguration config =
mockConfiguration();
+ Map<String, DataSource> dataSources = Collections.singletonMap("ds_0",
new MockedDataSource());
+
when(DataSourcePoolCreator.create(config.getDataSourcePoolPropertiesMap(),
false)).thenReturn(dataSources);
+ SQLException expected = new SQLException("fixture_failure");
+ when(ShardingSphereDataSourceFactory.createDataSource(eq("foo_db"),
any(ModeConfiguration.class), eq(dataSources), anyCollection(),
any(Properties.class)))
+ .thenThrow(expected);
+ try (MockedConstruction<DataSourcePoolDestroyer> destroyers =
mockConstruction(DataSourcePoolDestroyer.class)) {
+ assertThat(assertThrows(SQLException.class, () -> new
ShardingSpherePipelineDataSourceCreator().create(config)),
sameInstance(expected));
+ assertThat(destroyers.constructed().size(), is(1));
+ verify(destroyers.constructed().get(0)).asyncDestroy();
+ }
+ }
+
+ private ShardingSpherePipelineDataSourceConfiguration mockConfiguration() {
+ ShardingSpherePipelineDataSourceConfiguration result =
mock(ShardingSpherePipelineDataSourceConfiguration.class);
+ when(result.getDatabaseName()).thenReturn("foo_db");
+
when(result.getDataSourcePoolPropertiesMap()).thenReturn(Collections.singletonMap("ds_0",
mock(DataSourcePoolProperties.class)));
+ Collection<RuleConfiguration> rules = new LinkedList<>();
+ rules.add(mock(AuthorityRuleConfiguration.class));
+ rules.add(mock(RuleConfiguration.class));
+ when(result.getCreationRuleConfigurations()).thenReturn(rules);
+ return result;
+ }
+}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/StandardPipelineDataSourceCreatorTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/StandardPipelineDataSourceCreatorTest.java
new file mode 100644
index 00000000000..41c64e2cbbd
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/creator/StandardPipelineDataSourceCreatorTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.data.pipeline.core.datasource.creator;
+
+import
org.apache.shardingsphere.infra.datasource.pool.creator.DataSourcePoolCreator;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
+import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
+import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import javax.sql.DataSource;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(AutoMockExtension.class)
+@StaticMockSettings(DataSourcePoolCreator.class)
+class StandardPipelineDataSourceCreatorTest {
+
+ @Test
+ void assertCreate() {
+ DataSourcePoolProperties dataSourceProps =
mock(DataSourcePoolProperties.class);
+ DataSource expected = new MockedDataSource();
+
when(DataSourcePoolCreator.create(dataSourceProps)).thenReturn(expected);
+ assertThat(new
StandardPipelineDataSourceCreator().create(dataSourceProps),
sameInstance(expected));
+ }
+}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtilsTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtilsTest.java
index 66341ae4964..9bd1afdfd43 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtilsTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/util/PipelineDataSourceConfigurationUtilsTest.java
@@ -27,7 +27,6 @@ import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePo
import
org.apache.shardingsphere.infra.datasource.pool.props.domain.synonym.PoolPropertySynonyms;
import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
-import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
@@ -62,10 +61,11 @@ class PipelineDataSourceConfigurationUtilsTest {
dataSourceProps.put("maxPoolSize", 2);
dataSourceProps.put("maximumPoolSize", 3);
ShardingSpherePipelineDataSourceConfiguration pipelineDataSourceConfig
= mock(ShardingSpherePipelineDataSourceConfiguration.class);
-
when(pipelineDataSourceConfig.getRootConfig()).thenReturn(createYamlRootConfiguration(dataSourceProps));
+ Map<String, DataSourcePoolProperties> dataSources = new
LinkedHashMap<>(1, 1F);
+ dataSources.put("ds_0", new
DataSourcePoolProperties("com.zaxxer.hikari.HikariDataSource",
dataSourceProps));
+
when(pipelineDataSourceConfig.getDataSourcePoolPropertiesMap()).thenReturn(dataSources);
PipelineDataSourceConfigurationUtils.transformPipelineDataSourceConfiguration("foo_job",
pipelineDataSourceConfig, Collections.singletonMap("ds_0", mockStorageUnit()));
- assertThat(dataSourceProps.get("maxPoolSize"), is(10));
- assertThat(dataSourceProps.get("maximumPoolSize"), is(20));
+
assertThat(dataSources.get("ds_0").getPoolPropertySynonyms().getStandardProperties().get("maxPoolSize"),
is(10));
}
@Test
@@ -141,13 +141,6 @@ class PipelineDataSourceConfigurationUtilsTest {
return result;
}
- private YamlRootConfiguration createYamlRootConfiguration(final
Map<String, Object> dataSourceProps) {
- YamlRootConfiguration result = new YamlRootConfiguration();
- result.setDatabaseName("foo_db");
- result.setDataSources(Collections.singletonMap("ds_0",
dataSourceProps));
- return result;
- }
-
private StorageUnit mockStorageUnit() {
Map<String, Object> standardProps = new LinkedHashMap<>(2, 1F);
standardProps.put("maxPoolSize", 10);
diff --git
a/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviser.java
b/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviser.java
new file mode 100644
index 00000000000..d6f8e9b515c
--- /dev/null
+++
b/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviser.java
@@ -0,0 +1,66 @@
+/*
+ * 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.data.pipeline.sharding;
+
+import
org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser;
+import
org.apache.shardingsphere.infra.algorithm.core.config.AlgorithmConfiguration;
+import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
+import
org.apache.shardingsphere.sharding.api.config.rule.ShardingAutoTableRuleConfiguration;
+import
org.apache.shardingsphere.sharding.api.config.rule.ShardingTableRuleConfiguration;
+import org.apache.shardingsphere.sharding.constant.ShardingOrder;
+
+/**
+ * Sharding pipeline rule configuration reviser.
+ */
+public final class ShardingPipelineRuleConfigurationReviser implements
PipelineRuleConfigurationReviser<ShardingRuleConfiguration> {
+
+ @Override
+ public void revise(final ShardingRuleConfiguration ruleConfig) {
+ enableRangeQueryForInline(ruleConfig);
+ removeAuditStrategy(ruleConfig);
+ }
+
+ private void enableRangeQueryForInline(final ShardingRuleConfiguration
ruleConfig) {
+ for (AlgorithmConfiguration each :
ruleConfig.getShardingAlgorithms().values()) {
+ if ("INLINE".equalsIgnoreCase(each.getType()) ||
"COMPLEX_INLINE".equalsIgnoreCase(each.getType())) {
+
each.getProps().setProperty("allow-range-query-with-inline-sharding",
Boolean.TRUE.toString());
+ }
+ }
+ }
+
+ private void removeAuditStrategy(final ShardingRuleConfiguration
ruleConfig) {
+ ruleConfig.setDefaultAuditStrategy(null);
+ ruleConfig.getAuditors().clear();
+ for (ShardingTableRuleConfiguration each : ruleConfig.getTables()) {
+ each.setAuditStrategy(null);
+ }
+ for (ShardingAutoTableRuleConfiguration each :
ruleConfig.getAutoTables()) {
+ each.setAuditStrategy(null);
+ }
+ }
+
+ @Override
+ public int getOrder() {
+ return ShardingOrder.ORDER;
+ }
+
+ @Override
+ public Class<ShardingRuleConfiguration> getTypeClass() {
+ return ShardingRuleConfiguration.class;
+ }
+}
diff --git
a/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviser.java
b/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviser.java
deleted file mode 100644
index 535dd9cb16e..00000000000
---
a/kernel/data-pipeline/feature/sharding/src/main/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviser.java
+++ /dev/null
@@ -1,64 +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.data.pipeline.sharding;
-
-import
org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser;
-import
org.apache.shardingsphere.infra.algorithm.core.yaml.YamlAlgorithmConfiguration;
-import org.apache.shardingsphere.sharding.constant.ShardingOrder;
-import
org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
-
-/**
- * Sharding pipeline YAML rule configuration reviser.
- */
-public final class ShardingPipelineYamlRuleConfigurationReviser implements
PipelineYamlRuleConfigurationReviser<YamlShardingRuleConfiguration> {
-
- @Override
- public void revise(final YamlShardingRuleConfiguration yamlRuleConfig) {
- enableRangeQueryForInline(yamlRuleConfig);
- removeAuditStrategy(yamlRuleConfig);
- }
-
- private void enableRangeQueryForInline(final YamlShardingRuleConfiguration
yamlRuleConfig) {
- for (YamlAlgorithmConfiguration each :
yamlRuleConfig.getShardingAlgorithms().values()) {
- if ("INLINE".equalsIgnoreCase(each.getType()) ||
"COMPLEX_INLINE".equalsIgnoreCase(each.getType())) {
-
each.getProps().setProperty("allow-range-query-with-inline-sharding",
Boolean.TRUE.toString());
- }
- }
- }
-
- private void removeAuditStrategy(final YamlShardingRuleConfiguration
yamlRuleConfig) {
- yamlRuleConfig.setDefaultAuditStrategy(null);
- yamlRuleConfig.setAuditors(null);
- if (null != yamlRuleConfig.getTables()) {
- yamlRuleConfig.getTables().forEach((key, value) ->
value.setAuditStrategy(null));
- }
- if (null != yamlRuleConfig.getAutoTables()) {
- yamlRuleConfig.getAutoTables().forEach((key, value) ->
value.setAuditStrategy(null));
- }
- }
-
- @Override
- public int getOrder() {
- return ShardingOrder.ORDER;
- }
-
- @Override
- public Class<YamlShardingRuleConfiguration> getTypeClass() {
- return YamlShardingRuleConfiguration.class;
- }
-}
diff --git
a/kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser
b/kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser
similarity index 90%
rename from
kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser
rename to
kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser
index a6fdb924cb6..3286f547068 100644
---
a/kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser
+++
b/kernel/data-pipeline/feature/sharding/src/main/resources/META-INF/services/org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser
@@ -1,4 +1,3 @@
-#
# 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.
@@ -6,13 +5,12 @@
# (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
+# 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.
-#
-org.apache.shardingsphere.data.pipeline.sharding.ShardingPipelineYamlRuleConfigurationReviser
+org.apache.shardingsphere.data.pipeline.sharding.ShardingPipelineRuleConfigurationReviser
diff --git
a/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviserTest.java
b/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviserTest.java
new file mode 100644
index 00000000000..b11ec1e11ef
--- /dev/null
+++
b/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineRuleConfigurationReviserTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.data.pipeline.sharding;
+
+import
org.apache.shardingsphere.data.pipeline.api.type.ShardingSpherePipelineDataSourceConfiguration;
+import
org.apache.shardingsphere.data.pipeline.core.datasource.rule.PipelineRuleConfigurationReviser;
+import
org.apache.shardingsphere.infra.algorithm.core.config.AlgorithmConfiguration;
+import
org.apache.shardingsphere.infra.algorithm.core.yaml.YamlAlgorithmConfiguration;
+import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
+import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
+import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
+import
org.apache.shardingsphere.sharding.api.config.rule.ShardingAutoTableRuleConfiguration;
+import
org.apache.shardingsphere.sharding.api.config.rule.ShardingTableRuleConfiguration;
+import
org.apache.shardingsphere.sharding.api.config.strategy.audit.ShardingAuditStrategyConfiguration;
+import
org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ShardingPipelineRuleConfigurationReviserTest {
+
+ @SuppressWarnings("unchecked")
+ private final PipelineRuleConfigurationReviser<ShardingRuleConfiguration>
reviser = (PipelineRuleConfigurationReviser<ShardingRuleConfiguration>)
OrderedSPILoader
+ .getServicesByClass(PipelineRuleConfigurationReviser.class,
Collections.singleton(ShardingRuleConfiguration.class)).get(ShardingRuleConfiguration.class);
+
+ @Test
+ void assertReviseInlineAlgorithmsAndAuditStrategies() {
+ ShardingRuleConfiguration ruleConfig = new ShardingRuleConfiguration();
+ ruleConfig.getShardingAlgorithms().put("inline", new
AlgorithmConfiguration("INLINE", new Properties()));
+ ruleConfig.getShardingAlgorithms().put("complex_inline", new
AlgorithmConfiguration("complex_inline", new Properties()));
+ ruleConfig.getShardingAlgorithms().put("hash_mod", new
AlgorithmConfiguration("HASH_MOD", new Properties()));
+ ShardingAuditStrategyConfiguration auditStrategy = new
ShardingAuditStrategyConfiguration(Collections.singleton("fixture_auditor"),
false);
+ ruleConfig.setDefaultAuditStrategy(auditStrategy);
+ ruleConfig.getAuditors().put("fixture_auditor", new
AlgorithmConfiguration("SQL_HINT", new Properties()));
+ ShardingTableRuleConfiguration tableRule = new
ShardingTableRuleConfiguration("t_order", "ds_0.t_order");
+ tableRule.setAuditStrategy(auditStrategy);
+ ruleConfig.getTables().add(tableRule);
+ ShardingAutoTableRuleConfiguration autoTableRule = new
ShardingAutoTableRuleConfiguration("t_auto", "ds_0");
+ autoTableRule.setAuditStrategy(auditStrategy);
+ ruleConfig.getAutoTables().add(autoTableRule);
+ reviser.revise(ruleConfig);
+
assertThat(ruleConfig.getShardingAlgorithms().get("inline").getProps().getProperty("allow-range-query-with-inline-sharding"),
is(Boolean.TRUE.toString()));
+
assertThat(ruleConfig.getShardingAlgorithms().get("complex_inline").getProps().getProperty("allow-range-query-with-inline-sharding"),
is(Boolean.TRUE.toString()));
+
assertNull(ruleConfig.getShardingAlgorithms().get("hash_mod").getProps().getProperty("allow-range-query-with-inline-sharding"));
+ assertNull(ruleConfig.getDefaultAuditStrategy());
+ assertTrue(ruleConfig.getAuditors().isEmpty());
+ assertNull(tableRule.getAuditStrategy());
+ assertNull(autoTableRule.getAuditStrategy());
+ }
+
+ @Test
+ void assertRevisionDoesNotChangeOriginalRuleConfiguration() {
+ YamlRootConfiguration rootConfig = new YamlRootConfiguration();
+ rootConfig.setDatabaseName("foo_db");
+ Map<String, Object> dataSource = new HashMap<>(2, 1F);
+ dataSource.put("dataSourceClassName",
"com.zaxxer.hikari.HikariDataSource");
+ dataSource.put("url", "jdbc:h2:mem:foo_db");
+ rootConfig.setDataSources(Collections.singletonMap("ds_0",
dataSource));
+ YamlAlgorithmConfiguration algorithm = new
YamlAlgorithmConfiguration();
+ algorithm.setType("INLINE");
+ YamlShardingRuleConfiguration yamlRuleConfig = new
YamlShardingRuleConfiguration();
+ yamlRuleConfig.getShardingAlgorithms().put("inline", algorithm);
+ rootConfig.setRules(Collections.singleton(yamlRuleConfig));
+ ShardingSpherePipelineDataSourceConfiguration config = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig);
+ ShardingRuleConfiguration original = (ShardingRuleConfiguration)
config.getRuleConfigurations().iterator().next();
+ ShardingRuleConfiguration forCreation = (ShardingRuleConfiguration)
config.getCreationRuleConfigurations().iterator().next();
+ reviser.revise(forCreation);
+
assertNull(original.getShardingAlgorithms().get("inline").getProps().getProperty("allow-range-query-with-inline-sharding"));
+
assertThat(forCreation.getShardingAlgorithms().get("inline").getProps().getProperty("allow-range-query-with-inline-sharding"),
is(Boolean.TRUE.toString()));
+ ShardingRuleConfiguration nextCreation = (ShardingRuleConfiguration)
config.getCreationRuleConfigurations().iterator().next();
+
assertNull(nextCreation.getShardingAlgorithms().get("inline").getProps().getProperty("allow-range-query-with-inline-sharding"));
+ }
+}
diff --git
a/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviserTest.java
b/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviserTest.java
deleted file mode 100644
index 86e26cfa974..00000000000
---
a/kernel/data-pipeline/feature/sharding/src/test/java/org/apache/shardingsphere/data/pipeline/sharding/ShardingPipelineYamlRuleConfigurationReviserTest.java
+++ /dev/null
@@ -1,94 +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.data.pipeline.sharding;
-
-import
org.apache.shardingsphere.data.pipeline.core.datasource.yaml.PipelineYamlRuleConfigurationReviser;
-import
org.apache.shardingsphere.infra.algorithm.core.yaml.YamlAlgorithmConfiguration;
-import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
-import
org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.rule.YamlShardingAutoTableRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.rule.YamlTableRuleConfiguration;
-import
org.apache.shardingsphere.sharding.yaml.config.strategy.audit.YamlShardingAuditStrategyConfiguration;
-import org.junit.jupiter.api.Test;
-
-import java.util.Collections;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-class ShardingPipelineYamlRuleConfigurationReviserTest {
-
- @SuppressWarnings("unchecked")
- private final
PipelineYamlRuleConfigurationReviser<YamlShardingRuleConfiguration> reviser =
(PipelineYamlRuleConfigurationReviser<YamlShardingRuleConfiguration>)
OrderedSPILoader
- .getServicesByClass(PipelineYamlRuleConfigurationReviser.class,
Collections.singleton(YamlShardingRuleConfiguration.class)).get(YamlShardingRuleConfiguration.class);
-
- @Test
- void assertReviseInlineAlgorithmsAndAuditStrategies() {
- YamlShardingRuleConfiguration yamlConfig = new
YamlShardingRuleConfiguration();
- yamlConfig.getShardingAlgorithms().put("inline",
createAlgorithm("INLINE"));
- yamlConfig.getShardingAlgorithms().put("complex_inline",
createAlgorithm("complex_inline"));
- yamlConfig.getShardingAlgorithms().put("hash_mod",
createAlgorithm("HASH_MOD"));
- yamlConfig.setDefaultAuditStrategy(new
YamlShardingAuditStrategyConfiguration());
- yamlConfig.getAuditors().put("fixture_auditor",
createAlgorithm("SQL_HINT"));
- yamlConfig.getTables().put("t_order", createTableRuleWithAudit());
- yamlConfig.getAutoTables().put("t_auto",
createAutoTableRuleWithAudit());
- reviser.revise(yamlConfig);
-
assertThat(yamlConfig.getShardingAlgorithms().get("inline").getProps().getProperty("allow-range-query-with-inline-sharding"),
is(Boolean.TRUE.toString()));
-
assertThat(yamlConfig.getShardingAlgorithms().get("complex_inline").getProps().getProperty("allow-range-query-with-inline-sharding"),
is(Boolean.TRUE.toString()));
-
assertNull(yamlConfig.getShardingAlgorithms().get("hash_mod").getProps().getProperty("allow-range-query-with-inline-sharding"));
- assertNull(yamlConfig.getDefaultAuditStrategy());
- assertNull(yamlConfig.getAuditors());
- assertNull(yamlConfig.getTables().get("t_order").getAuditStrategy());
-
assertNull(yamlConfig.getAutoTables().get("t_auto").getAuditStrategy());
- }
-
- @Test
- void assertReviseWhenTablesAbsent() {
- YamlShardingRuleConfiguration yamlConfig = new
YamlShardingRuleConfiguration();
- yamlConfig.setTables(null);
- yamlConfig.setAutoTables(null);
- yamlConfig.setDefaultAuditStrategy(new
YamlShardingAuditStrategyConfiguration());
- yamlConfig.setAuditors(Collections.singletonMap("auditor",
createAlgorithm("SQL_HINT")));
- reviser.revise(yamlConfig);
- assertNull(yamlConfig.getTables());
- assertNull(yamlConfig.getAutoTables());
- assertNull(yamlConfig.getDefaultAuditStrategy());
- assertNull(yamlConfig.getAuditors());
- }
-
- private YamlAlgorithmConfiguration createAlgorithm(final String type) {
- YamlAlgorithmConfiguration result = new YamlAlgorithmConfiguration();
- result.setType(type);
- return result;
- }
-
- private YamlTableRuleConfiguration createTableRuleWithAudit() {
- YamlTableRuleConfiguration result = new YamlTableRuleConfiguration();
- result.setLogicTable("t_order");
- result.setAuditStrategy(new YamlShardingAuditStrategyConfiguration());
- return result;
- }
-
- private YamlShardingAutoTableRuleConfiguration
createAutoTableRuleWithAudit() {
- YamlShardingAutoTableRuleConfiguration result = new
YamlShardingAutoTableRuleConfiguration();
- result.setLogicTable("t_auto");
- result.setAuditStrategy(new YamlShardingAuditStrategyConfiguration());
- return result;
- }
-}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
index d7a54de4d30..90f84a10fd3 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
@@ -68,7 +68,6 @@ import
org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import java.util.Collection;
import java.util.HashMap;
@@ -164,7 +163,7 @@ public final class CDCJob implements PipelineJob {
@SuppressWarnings({"rawtypes", "unchecked"})
private Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final CDCJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(jobConfig.getDataSourceConfig().getRootConfig().getRules());
+ Collection<RuleConfiguration> ruleConfigs =
jobConfig.getDataSourceConfig().getRuleConfigurations();
Collection<ShardingSphereIdentifier> targetTableNames =
jobConfig.getSchemaTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
index 839f8ce8724..e6a158655fa 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
@@ -35,7 +35,6 @@ import
org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeEntry;
import org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeLine;
import
org.apache.shardingsphere.data.pipeline.core.datanode.JobDataNodeLineConvertUtils;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
-import
org.apache.shardingsphere.data.pipeline.core.datasource.config.PipelineDataSourceConfigurationFactory;
import
org.apache.shardingsphere.data.pipeline.core.datasource.yaml.swapper.YamlPipelineDataSourceConfigurationSwapper;
import
org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
import
org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobCreationWithInvalidShardingCountException;
@@ -65,6 +64,7 @@ import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoa
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import
org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.OneOffJobBootstrap;
import org.apache.shardingsphere.infra.datanode.DataNode;
+import
org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
@@ -156,7 +156,7 @@ public final class CDCJobAPI implements TransmissionJobAPI {
}
private void checkDataSources(final CDCJobConfiguration jobConfig) {
- Map<String, Map<String, Object>> dataSources =
jobConfig.getDataSourceConfig().getRootConfig().getDataSources();
+ Map<String, DataSourcePoolProperties> dataSources =
jobConfig.getDataSourceConfig().getDataSourcePoolPropertiesMap();
for (DataNode each : getDataNodes(jobConfig)) {
ShardingSpherePreconditions.checkContainsKey(dataSources,
each.getDataSourceName(),
() -> new
PipelineInvalidParameterException(String.format("Data source `%s` does not
exist in job data source configuration.", each.getDataSourceName())));
@@ -190,14 +190,14 @@ public final class CDCJobAPI implements
TransmissionJobAPI {
sinkConfig.setProps(sinkProps);
result.setSinkConfig(sinkConfig);
ShardingSphereDatabase database =
PipelineContextManager.getContext(contextKey).getMetaDataContexts().getMetaData().getDatabase(param.getDatabaseName());
-
result.setDataSourceConfiguration(pipelineDataSourceConfigSwapper.swapToYamlConfiguration(getDataSourceConfiguration(database)));
+ ShardingSpherePipelineDataSourceConfiguration dataSourceConfig =
getDataSourceConfiguration(database);
+
result.setDataSourceConfiguration(pipelineDataSourceConfigSwapper.swapToYamlConfiguration(dataSourceConfig));
List<JobDataNodeLine> jobDataNodeLines =
JobDataNodeLineConvertUtils.convertDataNodesToLines(param.getTableAndDataNodesMap());
result.setJobShardingDataNodes(jobDataNodeLines.stream().map(JobDataNodeLine::marshal).collect(Collectors.toList()));
JobDataNodeLine tableFirstDataNodes = new
JobDataNodeLine(param.getTableAndDataNodesMap().entrySet().stream()
.map(entry -> new JobDataNodeEntry(entry.getKey(),
entry.getValue().subList(0, 1))).collect(Collectors.toList()));
result.setTablesFirstDataNodes(tableFirstDataNodes.marshal());
-
result.setSourceDatabaseType(PipelineDataSourceConfigurationFactory.newInstance(
- result.getDataSourceConfiguration().getType(),
result.getDataSourceConfiguration().getParameter()).getDatabaseType().getType());
+
result.setSourceDatabaseType(dataSourceConfig.getDatabaseType().getType());
return result;
}
@@ -327,12 +327,12 @@ public final class CDCJobAPI implements
TransmissionJobAPI {
}
private void cleanup(final CDCJobConfiguration jobConfig) {
- for (Entry<String, Map<String, Object>> entry :
jobConfig.getDataSourceConfig().getRootConfig().getDataSources().entrySet()) {
+ for (String each :
jobConfig.getDataSourceConfig().getDataSourcePoolPropertiesMap().keySet()) {
try {
- StandardPipelineDataSourceConfiguration
pipelineDataSourceConfig = new
StandardPipelineDataSourceConfiguration(entry.getValue());
+ StandardPipelineDataSourceConfiguration
pipelineDataSourceConfig =
jobConfig.getDataSourceConfig().getActualDataSourceConfiguration(each);
new
IncrementalTaskPositionManager(pipelineDataSourceConfig.getDatabaseType()).destroyPosition(jobConfig.getJobId(),
pipelineDataSourceConfig);
} catch (final SQLException ex) {
- log.warn("job destroying failed, jobId={}, dataSourceName={}",
jobConfig.getJobId(), entry.getKey(), ex);
+ log.warn("job destroying failed, jobId={}, dataSourceName={}",
jobConfig.getJobId(), each, ex);
}
}
}
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
index a2f161eae5b..7e3a5df1251 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobTest.java
@@ -66,9 +66,6 @@ import
org.apache.shardingsphere.infra.instance.metadata.InstanceType;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
-import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
import org.junit.jupiter.api.Test;
@@ -198,11 +195,9 @@ class CDCJobTest {
when(PipelineAPIFactory.getPipelineGovernanceFacade(CONTEXT_KEY)).thenReturn(mock(PipelineGovernanceFacade.class,
RETURNS_DEEP_STUBS));
when(PipelineDistributedBarrier.getInstance(CONTEXT_KEY)).thenReturn(mock(PipelineDistributedBarrier.class));
when(PipelineDataSourceConfigurationFactory.newInstance(anyString(),
anyString())).thenReturn(mock(PipelineDataSourceConfiguration.class));
- YamlRuleConfiguration yamlRuleConfig =
mock(YamlRuleConfiguration.class);
- Collection<YamlRuleConfiguration> yamlRuleConfigs =
Collections.singleton(yamlRuleConfig);
-
jobConfig.getDataSourceConfig().getRootConfig().setRules(yamlRuleConfigs);
RuleConfiguration ruleConfig = mock(RuleConfiguration.class);
Collection<RuleConfiguration> ruleConfigs =
Collections.singleton(ruleConfig);
+
when(jobConfig.getDataSourceConfig().getRuleConfigurations()).thenReturn(ruleConfigs);
PipelineRequiredColumnsExtractor extractor =
mock(PipelineRequiredColumnsExtractor.class);
Map<ShardingSphereIdentifier, Collection<String>> requiredColumns =
Collections.singletonMap(new ShardingSphereIdentifier("logic_tbl"),
Collections.singleton("id"));
when(extractor.getTableAndRequiredColumnsMap(eq(ruleConfig),
anyCollection())).thenReturn(requiredColumns);
@@ -211,15 +206,9 @@ class CDCJobTest {
capturedRuleConfigs.set(invocation.getArgument(1));
return Collections.singletonMap(ruleConfig, extractor);
});
- AtomicReference<Collection<YamlRuleConfiguration>>
capturedYamlRuleConfigs = new AtomicReference<>();
AtomicReference<CDCJobItemContext> capturedContext = new
AtomicReference<>();
try (
MockedStatic<TypedSPILoader> typedSPILoader =
mockStatic(TypedSPILoader.class);
- MockedConstruction<YamlRuleConfigurationSwapperEngine>
ignoredRuleConfigSwapper = mockConstruction(
- YamlRuleConfigurationSwapperEngine.class, (mock,
context) ->
when(mock.swapToRuleConfigurations(anyCollection())).thenAnswer(invocation -> {
-
capturedYamlRuleConfigs.set(invocation.getArgument(0));
- return ruleConfigs;
- }));
MockedConstruction<PipelineProcessConfigurationPersistService>
ignoredProcess = mockPersistService(processConfig);
MockedConstruction<CDCJobPreparer> ignoredPreparer =
mockConstruction(CDCJobPreparer.class, (mock, context) -> doAnswer(invocation
-> {
CDCJobItemContext jobItemContext =
((Collection<CDCJobItemContext>) invocation.getArgument(0)).iterator().next();
@@ -239,7 +228,6 @@ class CDCJobTest {
new CDCJob(mock(PipelineSink.class)).execute(shardingContext);
}
assertThat(capturedContext.get().getStatus(),
is(JobStatus.EXECUTE_INCREMENTAL_TASK));
- assertThat(capturedYamlRuleConfigs.get(),
sameInstance(yamlRuleConfigs));
assertThat(capturedRuleConfigs.get(), sameInstance(ruleConfigs));
assertThat(capturedContext.get().getTaskConfig().getImporterConfig().getShardingColumns("logic_tbl"),
is(Collections.singleton("id")));
}
@@ -405,18 +393,10 @@ class CDCJobTest {
ShardingSpherePipelineDataSourceConfiguration dataSourceConfig =
mock(ShardingSpherePipelineDataSourceConfiguration.class);
when(dataSourceConfig.getType()).thenReturn("JDBC");
when(dataSourceConfig.getParameter()).thenReturn("param");
-
when(dataSourceConfig.getRootConfig()).thenReturn(createYAMLRootConfiguration());
when(result.getDataSourceConfig()).thenReturn(dataSourceConfig);
return result;
}
- private YamlRootConfiguration createYAMLRootConfiguration() {
- YamlRootConfiguration result = new YamlRootConfiguration();
- result.setDatabaseName("logic_db");
- result.setDataSources(Collections.singletonMap("ds_0",
Collections.emptyMap()));
- return result;
- }
-
@SuppressWarnings({"rawtypes", "unchecked"})
private void prepareJobTypeAndContext(final CDCJobConfiguration jobConfig)
{
YamlPipelineJobConfigurationSwapper jobConfigSwapper =
mock(YamlPipelineJobConfigurationSwapper.class);
diff --git
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
index 9151fb38557..9d54fc11372 100644
---
a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
+++
b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPITest.java
@@ -72,6 +72,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import org.mockito.internal.configuration.plugins.Plugins;
@@ -273,6 +274,9 @@ class CDCJobAPITest {
StreamDataParameter param = new StreamDataParameter("foo_db", new
LinkedList<>(Collections.singletonList("foo_schema.foo_tbl")), true,
Collections.singletonMap("foo_schema.foo_tbl",
Collections.singletonList(new DataNode("foo_ds" + ".foo_tbl"))), false);
assertThat(jobAPI.create(param, CDCSinkType.SOCKET, new
Properties()), is("foo_job"));
+ ArgumentCaptor<YamlCDCJobConfiguration> yamlJobConfig =
ArgumentCaptor.forClass(YamlCDCJobConfiguration.class);
+
verify(ignored.constructed().get(0)).swapToObject(yamlJobConfig.capture());
+ assertThat(yamlJobConfig.getValue().getSourceDatabaseType(),
is("H2"));
assertTrue(positionManagerConstruction.constructed().isEmpty());
}
}
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
index 15723ed22ff..c73469b9397 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallback.java
@@ -54,7 +54,6 @@ import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUn
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedTable;
import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentifier;
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import java.sql.SQLException;
import java.util.Collection;
@@ -128,8 +127,7 @@ public final class MigrationJobExecutorCallback implements
DistributedPipelineJo
@SuppressWarnings({"rawtypes", "unchecked"})
private Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final MigrationJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(
- ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules());
+ Collection<RuleConfiguration> ruleConfigs =
((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRuleConfigurations();
Collection<ShardingSphereIdentifier> targetTableNames =
jobConfig.getTargetTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
index bd0ec4a0cf7..ceaa77d6d91 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJobExecutorCallbackTest.java
@@ -55,8 +55,6 @@ import
org.apache.shardingsphere.infra.metadata.identifier.ShardingSphereIdentif
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.pojo.rule.YamlRuleConfiguration;
-import
org.apache.shardingsphere.infra.yaml.config.swapper.rule.YamlRuleConfigurationSwapperEngine;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
@@ -122,22 +120,13 @@ class MigrationJobExecutorCallbackTest {
@SuppressWarnings({"rawtypes", "unchecked"})
@Test
- void assertBuildJobItemContextWithConvertedRuleConfigurations() {
- YamlRuleConfiguration yamlRuleConfig =
mock(YamlRuleConfiguration.class);
- Collection<YamlRuleConfiguration> yamlRuleConfigs =
Collections.singleton(yamlRuleConfig);
+ void assertBuildJobItemContextWithRuleConfigurations() {
RuleConfiguration ruleConfig = mock(RuleConfiguration.class);
- Collection<RuleConfiguration> ruleConfigs =
Collections.singleton(ruleConfig);
PipelineRequiredColumnsExtractor extractor =
mock(PipelineRequiredColumnsExtractor.class);
Map<ShardingSphereIdentifier, Collection<String>> requiredColumns =
Collections.singletonMap(new ShardingSphereIdentifier("t_order"),
Collections.singleton("id"));
when(extractor.getTableAndRequiredColumnsMap(eq(ruleConfig),
anyCollection())).thenReturn(requiredColumns);
- AtomicReference<Collection<YamlRuleConfiguration>>
capturedYamlRuleConfigs = new AtomicReference<>();
AtomicReference<Collection<RuleConfiguration>> capturedRuleConfigs =
new AtomicReference<>();
try (
- MockedConstruction<YamlRuleConfigurationSwapperEngine>
ignoredRuleConfigSwapper = mockConstruction(
- YamlRuleConfigurationSwapperEngine.class, (mock,
context) ->
when(mock.swapToRuleConfigurations(anyCollection())).thenAnswer(invocation -> {
-
capturedYamlRuleConfigs.set(invocation.getArgument(0));
- return ruleConfigs;
- }));
MockedStatic<OrderedSPILoader> orderedSPILoader =
mockStatic(OrderedSPILoader.class);
MockedStatic<DatabaseTypeFactory> databaseTypeFactory =
mockStatic(DatabaseTypeFactory.class);
MockedStatic<PipelineAPIFactory> pipelineAPIFactory =
mockStatic(PipelineAPIFactory.class);
@@ -148,12 +137,12 @@ class MigrationJobExecutorCallbackTest {
});
mockDatabaseTypeFactory(databaseTypeFactory);
MigrationJobConfiguration jobConfig = createJobConfiguration();
- ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().setRules(yamlRuleConfigs);
+ Collection<RuleConfiguration> ruleConfigs =
((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRuleConfigurations();
+ ruleConfigs.add(ruleConfig);
mockGovernanceFacade(pipelineAPIFactory,
createSourceDataSourceYaml(10, 20));
mockProxyContext(pipelineContextManager);
MigrationJobItemContext actual = new
MigrationJobExecutorCallback().buildJobItemContext(
jobConfig, 0, null, mockProcessContext(),
mock(PipelineDataSourceManager.class));
- assertThat(capturedYamlRuleConfigs.get(),
sameInstance(yamlRuleConfigs));
assertThat(capturedRuleConfigs.get(), sameInstance(ruleConfigs));
assertThat(actual.getTaskConfig().getImporterConfig().getShardingColumns("t_order"),
is(Collections.singleton("id")));
}
@@ -315,7 +304,8 @@ class MigrationJobExecutorCallbackTest {
}
private Map<String, Object> createDataSourceProperties(final String
jdbcUrl, final int maxPoolSize, final int maximumPoolSize) {
- Map<String, Object> result = new LinkedHashMap<>(5, 1F);
+ Map<String, Object> result = new LinkedHashMap<>(6, 1F);
+ result.put("dataSourceClassName",
"com.zaxxer.hikari.HikariDataSource");
result.put("url", jdbcUrl);
result.put("username", "root");
result.put("password", "root");
@@ -337,7 +327,7 @@ class MigrationJobExecutorCallbackTest {
}
private String getRootDataSourceMaxPoolSize(final
ShardingSpherePipelineDataSourceConfiguration dataSourceConfig) {
- Collection<Map<String, Object>> dataSources =
dataSourceConfig.getRootConfig().getDataSources().values();
- return
String.valueOf(dataSources.iterator().next().get("maxPoolSize"));
+ Collection<DataSourcePoolProperties> dataSources =
dataSourceConfig.getDataSourcePoolPropertiesMap().values();
+ return
String.valueOf(dataSources.iterator().next().getPoolPropertySynonyms().getStandardProperties().get("maxPoolSize"));
}
}
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
index 19eb0dfc313..f981d1d3999 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java
@@ -59,6 +59,7 @@ import
org.apache.shardingsphere.infra.metadata.statistics.builder.ShardingSpher
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.infra.util.file.SystemResourceFileUtils;
+import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.infra.yaml.config.pojo.YamlRootConfiguration;
import
org.apache.shardingsphere.infra.yaml.config.swapper.mode.YamlModeConfigurationSwapper;
import
org.apache.shardingsphere.infra.yaml.config.swapper.resource.YamlDataSourceConfigurationSwapper;
@@ -78,6 +79,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Map.Entry;
@@ -104,9 +106,13 @@ public final class PipelineContextUtils {
if (null != PipelineContextManager.getContext(contextKey)) {
return;
}
- ShardingSpherePipelineDataSourceConfiguration pipelineDataSourceConfig
= new ShardingSpherePipelineDataSourceConfiguration(
-
SystemResourceFileUtils.readFile("config_sharding_sphere_jdbc_source.yaml"));
- YamlRootConfiguration rootConfig = (YamlRootConfiguration)
pipelineDataSourceConfig.getDataSourceConfiguration();
+ String yaml =
SystemResourceFileUtils.readFile("config_sharding_sphere_jdbc_source.yaml");
+ ShardingSpherePipelineDataSourceConfiguration pipelineDataSourceConfig
= new ShardingSpherePipelineDataSourceConfiguration(yaml);
+ YamlRootConfiguration rootConfig =
YamlEngine.unmarshal(pipelineDataSourceConfig.getParameter(),
YamlRootConfiguration.class, true);
+ Map<String, Map<String, Object>> dataSources = new
LinkedHashMap<>(pipelineDataSourceConfig.getDataSourcePoolPropertiesMap().size(),
1F);
+ YamlDataSourceConfigurationSwapper swapper = new
YamlDataSourceConfigurationSwapper();
+
pipelineDataSourceConfig.getDataSourcePoolPropertiesMap().forEach((name, props)
-> dataSources.put(name, swapper.swapToMap(props)));
+ rootConfig.setDataSources(dataSources);
ContextManager contextManager = getContextManager(rootConfig);
ClusterPersistRepository persistRepository =
getClusterPersistRepository(
(ClusterPersistRepositoryConfiguration)
contextManager.getComputeNodeInstanceContext().getModeConfiguration().getRepository());
@@ -220,8 +226,7 @@ public final class PipelineContextUtils {
@SuppressWarnings({"rawtypes", "unchecked"})
private static Map<ShardingSphereIdentifier, Collection<String>>
getTableAndRequiredColumnsMap(final MigrationJobConfiguration jobConfig) {
Map<ShardingSphereIdentifier, Collection<String>> result = new
HashMap<>();
- Collection<RuleConfiguration> ruleConfigs = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(
- ((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRootConfig().getRules());
+ Collection<RuleConfiguration> ruleConfigs =
((ShardingSpherePipelineDataSourceConfiguration)
jobConfig.getTarget()).getRuleConfigurations();
Set<ShardingSphereIdentifier> targetTableNames =
jobConfig.getTargetTableNames().stream().map(ShardingSphereIdentifier::new).collect(Collectors.toSet());
for (Entry<RuleConfiguration, PipelineRequiredColumnsExtractor> entry
: OrderedSPILoader.getServices(PipelineRequiredColumnsExtractor.class,
ruleConfigs).entrySet()) {
result.putAll(entry.getValue().getTableAndRequiredColumnsMap(entry.getKey(),
targetTableNames));
