You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2026/01/14 12:33:58 UTC
(shardingsphere) branch master updated: Improve PipelineProcessConfigurationPersistServiceTest code coverage rate (#37739)
This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng 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 279458aad7c Improve PipelineProcessConfigurationPersistServiceTest code coverage rate (#37739)
279458aad7c is described below
commit 279458aad7c26a44e408083b6644c4ef7ab41373
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Wed Jan 14 20:33:52 2026 +0800
Improve PipelineProcessConfigurationPersistServiceTest code coverage rate (#37739)
---
...lineProcessConfigurationPersistServiceTest.java | 150 ++++++++++++++++++---
1 file changed, 132 insertions(+), 18 deletions(-)
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/metadata/PipelineProcessConfigurationPersistServiceTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/metadata/PipelineProcessConfigurationPersistServiceTest.java
index ceab68d907e..013a4f07d5d 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/metadata/PipelineProcessConfigurationPersistServiceTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/core/metadata/PipelineProcessConfigurationPersistServiceTest.java
@@ -17,48 +17,162 @@
package org.apache.shardingsphere.data.pipeline.core.metadata;
-import org.apache.shardingsphere.data.pipeline.core.job.progress.config.PipelineProcessConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.yaml.config.YamlPipelineProcessConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.yaml.config.YamlPipelineReadConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.yaml.config.YamlPipelineWriteConfiguration;
import org.apache.shardingsphere.data.pipeline.core.job.progress.config.yaml.swapper.YamlPipelineProcessConfigurationSwapper;
import org.apache.shardingsphere.infra.algorithm.core.yaml.YamlAlgorithmConfiguration;
+import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
import org.apache.shardingsphere.test.it.data.pipeline.core.util.PipelineContextUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import java.util.Properties;
+
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNull;
class PipelineProcessConfigurationPersistServiceTest {
+ private static final YamlPipelineProcessConfigurationSwapper SWAPPER = new YamlPipelineProcessConfigurationSwapper();
+
@BeforeAll
static void beforeClass() {
PipelineContextUtils.initPipelineContextManager();
}
@Test
- void assertLoadAndPersist() {
- YamlPipelineProcessConfiguration yamlProcessConfig = createYamlPipelineProcessConfiguration();
- String expectedYamlText = YamlEngine.marshal(yamlProcessConfig);
- PipelineProcessConfiguration processConfig = new YamlPipelineProcessConfigurationSwapper().swapToObject(yamlProcessConfig);
- PipelineProcessConfigurationPersistService persistService = new PipelineProcessConfigurationPersistService();
- persistService.persist(PipelineContextUtils.getContextKey(), "MIGRATION", processConfig);
- String actualYamlText = YamlEngine.marshal(new YamlPipelineProcessConfigurationSwapper().swapToYamlConfiguration(persistService.load(PipelineContextUtils.getContextKey(), "MIGRATION")));
- assertThat(actualYamlText, is(expectedYamlText));
+ void assertPersistAndLoadReadConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("READ");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), expected.getRead());
+ assertYamlConfiguration(actual.getWrite(), new YamlPipelineWriteConfiguration());
+ assertNull(actual.getStreamChannel());
+ }
+
+ @Test
+ void assertPersistAndLoadWriteConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("WRITE");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), new YamlPipelineReadConfiguration());
+ assertYamlConfiguration(actual.getWrite(), expected.getWrite());
+ assertNull(actual.getStreamChannel());
+ }
+
+ @Test
+ void assertPersistAndLoadStreamChannelConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("STREAM_CHANNEL");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), new YamlPipelineReadConfiguration());
+ assertYamlConfiguration(actual.getWrite(), new YamlPipelineWriteConfiguration());
+ assertYamlConfiguration(actual.getStreamChannel(), expected.getStreamChannel());
+ }
+
+ @Test
+ void assertPersistAndLoadReadWriteConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("READ,WRITE");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), expected.getRead());
+ assertYamlConfiguration(actual.getWrite(), expected.getWrite());
+ assertNull(actual.getStreamChannel());
+ }
+
+ @Test
+ void assertPersistAndLoadReadStreamChannelConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("READ,STREAM_CHANNEL");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), expected.getRead());
+ assertYamlConfiguration(actual.getWrite(), new YamlPipelineWriteConfiguration());
+ assertYamlConfiguration(actual.getStreamChannel(), expected.getStreamChannel());
+ }
+
+ @Test
+ void assertPersistAndLoadWriteStreamChannelConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("WRITE,STREAM_CHANNEL");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), new YamlPipelineReadConfiguration());
+ assertYamlConfiguration(actual.getWrite(), expected.getWrite());
+ assertYamlConfiguration(actual.getStreamChannel(), expected.getStreamChannel());
}
- private YamlPipelineProcessConfiguration createYamlPipelineProcessConfiguration() {
+ @Test
+ void assertPersistAndLoadAllConfiguration() {
+ YamlPipelineProcessConfiguration expected = createYamlPipelineProcessConfiguration("READ,WRITE,STREAM_CHANNEL");
+ YamlPipelineProcessConfiguration actual = persistAndLoad(expected);
+ assertYamlConfiguration(actual.getRead(), expected.getRead());
+ assertYamlConfiguration(actual.getWrite(), expected.getWrite());
+ assertYamlConfiguration(actual.getStreamChannel(), expected.getStreamChannel());
+ }
+
+ private YamlPipelineProcessConfiguration createYamlPipelineProcessConfiguration(final String types) {
YamlPipelineProcessConfiguration result = new YamlPipelineProcessConfiguration();
- YamlPipelineReadConfiguration yamlReadConfig = new YamlPipelineReadConfiguration();
- yamlReadConfig.setShardingSize(10);
- result.setRead(yamlReadConfig);
- YamlPipelineWriteConfiguration yamlWriteConfig = new YamlPipelineWriteConfiguration();
- result.setWrite(yamlWriteConfig);
- YamlAlgorithmConfiguration yamlStreamChannel = new YamlAlgorithmConfiguration();
- yamlStreamChannel.setType("MEMORY");
- result.setStreamChannel(yamlStreamChannel);
+ // Ref AlterTransmissionRuleExecutor
+ result.setRead(null);
+ result.setWrite(null);
+ result.setStreamChannel(null);
+ for (String each : types.split(",")) {
+ switch (each.trim()) {
+ case "READ":
+ result.setRead(createYamlPipelineReadConfiguration());
+ break;
+ case "WRITE":
+ result.setWrite(createYamlPipelineWriteConfiguration());
+ break;
+ case "STREAM_CHANNEL":
+ result.setStreamChannel(createYamlStreamChannelConfiguration());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown type: " + each);
+ }
+ }
+ return result;
+ }
+
+ private YamlPipelineReadConfiguration createYamlPipelineReadConfiguration() {
+ YamlPipelineReadConfiguration result = new YamlPipelineReadConfiguration();
+ result.setWorkerThread(10);
+ result.setBatchSize(1000);
+ result.setShardingSize(10000000);
+ YamlAlgorithmConfiguration rateLimiter = new YamlAlgorithmConfiguration();
+ result.setRateLimiter(rateLimiter);
+ rateLimiter.setType("QPS");
+ Properties props = new Properties();
+ rateLimiter.setProps(props);
+ props.setProperty("qps", "500");
+ return result;
+ }
+
+ private YamlPipelineWriteConfiguration createYamlPipelineWriteConfiguration() {
+ YamlPipelineWriteConfiguration result = new YamlPipelineWriteConfiguration();
+ result.setWorkerThread(10);
+ result.setBatchSize(1000);
+ YamlAlgorithmConfiguration rateLimiter = new YamlAlgorithmConfiguration();
+ result.setRateLimiter(rateLimiter);
+ rateLimiter.setType("TPS");
+ Properties props = new Properties();
+ rateLimiter.setProps(props);
+ props.setProperty("tps", "2000");
return result;
}
+
+ private YamlAlgorithmConfiguration createYamlStreamChannelConfiguration() {
+ YamlAlgorithmConfiguration result = new YamlAlgorithmConfiguration();
+ result.setType("MEMORY");
+ Properties props = new Properties();
+ result.setProps(props);
+ props.setProperty("block-queue-size", "1000");
+ return result;
+ }
+
+ YamlPipelineProcessConfiguration persistAndLoad(final YamlPipelineProcessConfiguration yamlProcessConfig) {
+ PipelineProcessConfigurationPersistService persistService = new PipelineProcessConfigurationPersistService();
+ persistService.persist(PipelineContextUtils.getContextKey(), "MIGRATION", SWAPPER.swapToObject(yamlProcessConfig));
+ return SWAPPER.swapToYamlConfiguration(persistService.load(PipelineContextUtils.getContextKey(), "MIGRATION"));
+ }
+
+ private void assertYamlConfiguration(final YamlConfiguration actual, final YamlConfiguration expected) {
+ assertThat(YamlEngine.marshal(actual), is(YamlEngine.marshal(expected)));
+ }
}