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 86c324080a2 Move Pipeline data source YAML conversion to construction
(#39933)
86c324080a2 is described below
commit 86c324080a2d73d6f0fe5f19daa3383c9ca41e9d
Author: Liang Zhang <[email protected]>
AuthorDate: Thu Sep 24 00:04:55 2026 +0800
Move Pipeline data source YAML conversion to construction (#39933)
---
...rdingSpherePipelineDataSourceConfiguration.java | 60 +++++-----------------
...gSpherePipelineDataSourceConfigurationTest.java | 31 +++++++++++
...PipelineDataSourceConfigurationFactoryTest.java | 3 +-
3 files changed, 46 insertions(+), 48 deletions(-)
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 03ffb396c6d..928935b9128 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
@@ -18,6 +18,7 @@
package org.apache.shardingsphere.data.pipeline.api.type;
import com.google.common.base.Preconditions;
+import lombok.AccessLevel;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
@@ -60,9 +61,12 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
private final String databaseName;
- private volatile Map<String, DataSourcePoolProperties>
dataSourcePoolPropertiesMap;
+ @Getter(AccessLevel.NONE)
+ private final Map<String, Map<String, Object>> dataSources;
- private volatile Collection<RuleConfiguration> ruleConfigurations;
+ private final Map<String, DataSourcePoolProperties>
dataSourcePoolPropertiesMap;
+
+ private final Collection<RuleConfiguration> ruleConfigurations;
private final DatabaseType databaseType;
@@ -76,6 +80,9 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
Map<String, Object> props =
rootConfig.getDataSources().values().iterator().next();
databaseType = DatabaseTypeFactory.get(getJdbcUrl(props));
databaseName = rootConfig.getDatabaseName();
+ dataSources = rootConfig.getDataSources();
+ dataSourcePoolPropertiesMap =
createDataSourcePoolPropertiesMap(dataSources);
+ ruleConfigurations = new
YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(rootConfig.getRules());
}
public ShardingSpherePipelineDataSourceConfiguration(final
YamlRootConfiguration rootConfig) {
@@ -123,51 +130,11 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
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));
+ result.put(key,
adjustDataSourcePoolProperties(swapper.swapToDataSourcePoolProperties(value),
extension, null == customPoolProps ? Collections.emptyMap() : customPoolProps));
});
return result;
}
- /**
- * 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.
*
@@ -179,7 +146,6 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
@Override
public Object getDataSourceConfiguration() {
- getDataSourcePoolPropertiesMap();
return this;
}
@@ -195,9 +161,9 @@ public final class
ShardingSpherePipelineDataSourceConfiguration implements Pipe
* @return actual data source configuration
*/
public StandardPipelineDataSourceConfiguration
getActualDataSourceConfiguration(final String actualDataSourceName) {
- YamlRootConfiguration rootConfig = YamlEngine.unmarshal(parameter,
YamlRootConfiguration.class, true);
- Map<String, Object> dataSourceConfig =
rootConfig.getDataSources().get(actualDataSourceName);
- Preconditions.checkNotNull(dataSourceConfig, "actualDataSourceName
'{}' does not exist", actualDataSourceName);
+ Map<String, Object> sourceDataSourceConfig =
dataSources.get(actualDataSourceName);
+ Preconditions.checkNotNull(sourceDataSourceConfig,
"actualDataSourceName '{}' does not exist", actualDataSourceName);
+ Map<String, Object> dataSourceConfig = new
LinkedHashMap<>(sourceDataSourceConfig);
StandardPipelineDataSourceConfiguration adjusted = new
StandardPipelineDataSourceConfiguration(dataSourceConfig);
String jdbcUrlKey = dataSourceConfig.containsKey("url") ? "url" :
"jdbcUrl";
dataSourceConfig.put(jdbcUrlKey, adjusted.getUrl());
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 af73f44e4b5..c8db64f77a9 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
@@ -31,6 +31,7 @@ import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockS
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import java.util.Collections;
import java.util.HashMap;
@@ -47,9 +48,13 @@ 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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
@ExtendWith(AutoMockExtension.class)
@@ -129,6 +134,32 @@ class ShardingSpherePipelineDataSourceConfigurationTest {
assertThat(config.getDataSourcePoolPropertiesMap().get("ds_0").getAllLocalProperties().get("minimumIdle"),
is(7));
}
+ @Test
+ void assertRuntimeReadsDoNotParsePersistedParameter() {
+ ShardingSpherePipelineDataSourceConfiguration config = new
ShardingSpherePipelineDataSourceConfiguration(
+
SystemResourceFileUtils.readFile("yaml/shardingsphere-pipeline-datasource-config.yaml"));
+ try (MockedStatic<YamlEngine> yamlEngine =
mockStatic(YamlEngine.class, CALLS_REAL_METHODS)) {
+ config.getDataSourceConfiguration();
+ config.getDataSourcePoolPropertiesMap();
+ config.getRuleConfigurations();
+ config.getActualDataSourceConfiguration("ds_0");
+ yamlEngine.verify(() -> YamlEngine.unmarshal(anyString(),
eq(YamlRootConfiguration.class), eq(true)), never());
+ }
+ }
+
+ @Test
+ void assertPersistedParameterRestoresRuntimeConfiguration() {
+ 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 original = new
ShardingSpherePipelineDataSourceConfiguration(rootConfig);
+ ShardingSpherePipelineDataSourceConfiguration actual = new
ShardingSpherePipelineDataSourceConfiguration(original.getParameter());
+
assertThat(actual.getDataSourcePoolPropertiesMap().get("ds_0").getAllLocalProperties().get("connectionTimeout"),
is(30000));
+ assertThat(actual.getActualDataSourceConfiguration("ds_0").getUrl(),
is("jdbc:mock://127.0.0.1/ds_0"));
+ YamlRootConfiguration persisted =
YamlEngine.unmarshal(actual.getParameter(), YamlRootConfiguration.class, true);
+ assertThat(persisted.getDataSources().get("ds_0").get("minPoolSize"),
is(20));
+ assertThat(((Map<?, ?>)
persisted.getDataSources().get("ds_0").get("customPoolProps")).get("connectionTimeout"),
is(30000));
+ }
+
@Test
void assertAppendJdbcQueryPropertiesAfterPersistingParameter() {
JdbcQueryPropertiesExtension queryPropsExtension =
mock(JdbcQueryPropertiesExtension.class);
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/config/PipelineDataSourceConfigurationFactoryTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/config/PipelineDataSourceConfigurationFactoryTest.java
index 0ded7d251c7..e89f4437172 100644
---
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/config/PipelineDataSourceConfigurationFactoryTest.java
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/datasource/config/PipelineDataSourceConfigurationFactoryTest.java
@@ -36,7 +36,8 @@ class PipelineDataSourceConfigurationFactoryTest {
@Test
void assertNewInstanceForShardingSpherePipelineDataSourceConfiguration() {
-
assertThat(PipelineDataSourceConfigurationFactory.newInstance(ShardingSpherePipelineDataSourceConfiguration.TYPE,
"dataSources:\n" + " foo_ds:\n" + " url: jdbc:mock://127.0.0.1/foo_db"),
+
assertThat(PipelineDataSourceConfigurationFactory.newInstance(ShardingSpherePipelineDataSourceConfiguration.TYPE,
+ "dataSources:\n" + " foo_ds:\n" + " url:
jdbc:mock://127.0.0.1/foo_db\n" + " dataSourceClassName:
com.zaxxer.hikari.HikariDataSource"),
isA(ShardingSpherePipelineDataSourceConfiguration.class));
}
