sandynz commented on code in PR #16854: URL: https://github.com/apache/shardingsphere/pull/16854#discussion_r851121531
########## shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/node/PipelineMetaDataNode.java: ########## @@ -0,0 +1,52 @@ +/* + * 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.metadata.node; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import java.util.StringJoiner; + +/** + * Scaling meta data node. + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class PipelineMetaDataNode { + + public static final String ROOT_NODE = "scaling"; + + /** + * Get job config path. + * + * @param jobId job id. + * @return job config path. + */ + public static String getJobConfigPath(final String jobId) { + StringJoiner joiner = new StringJoiner("/"); + return joiner.add(getScalingRootPath()).add(jobId).add("config").toString(); + } + + /** + * get scaling root path. + * Review Comment: The first character of javadoc should be UPPERCASE. ########## shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/node/PipelineMetaDataNode.java: ########## @@ -0,0 +1,52 @@ +/* + * 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.metadata.node; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import java.util.StringJoiner; + +/** + * Scaling meta data node. + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class PipelineMetaDataNode { + + public static final String ROOT_NODE = "scaling"; + + /** + * Get job config path. + * + * @param jobId job id. + * @return job config path. + */ + public static String getJobConfigPath(final String jobId) { + StringJoiner joiner = new StringJoiner("/"); + return joiner.add(getScalingRootPath()).add(jobId).add("config").toString(); + } Review Comment: It's better to use the unified `Joiner` from Guava library. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorkerTest.java: ########## @@ -18,17 +18,40 @@ package org.apache.shardingsphere.data.pipeline.scenario.rulealtered; import com.google.common.collect.ImmutableMap; +import org.apache.commons.io.FileUtils; import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.JobConfiguration; import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.WorkflowConfiguration; +import org.apache.shardingsphere.data.pipeline.api.job.JobStatus; +import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress; +import org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI; +import org.apache.shardingsphere.data.pipeline.core.api.PipelineAPIFactory; import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException; +import org.apache.shardingsphere.data.pipeline.core.metadata.node.PipelineMetaDataNode; import org.apache.shardingsphere.data.pipeline.core.util.JobConfigurationBuilder; import org.apache.shardingsphere.data.pipeline.core.util.PipelineContextUtil; -import org.junit.Assert; +import org.apache.shardingsphere.data.pipeline.core.util.ReflectionUtil; +import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.cache.event.StartScalingEvent; +import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration; +import org.apache.shardingsphere.sharding.schedule.ShardingRuleAlteredDetector; +import org.apache.shardingsphere.spi.ShardingSphereServiceLoader; import org.junit.BeforeClass; import org.junit.Test; +import java.io.File; +import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.net.URL; +import java.util.Optional; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + public final class RuleAlteredJobWorkerTest { + static { + ShardingSphereServiceLoader.register(ShardingRuleAlteredDetector.class); + } Review Comment: Seems `RuleAlteredDetectorFactory` do the same registration, if we do not register it here, does unit test work? ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorkerTest.java: ########## @@ -44,7 +67,47 @@ public void assertCreateRuleAlteredContextNoAlteredRule() { @Test public void assertCreateRuleAlteredContextSuccess() { JobConfiguration jobConfig = JobConfigurationBuilder.createJobConfiguration(); Review Comment: `jobConfig` could be embedded into `createRuleAlteredContext` method parameter. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobPreparerTest.java: ########## @@ -0,0 +1,37 @@ +/* + * 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.scenario.rulealtered; + +import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobCreationException; +import org.apache.shardingsphere.data.pipeline.core.util.JobConfigurationBuilder; +import org.apache.shardingsphere.data.pipeline.core.util.PipelineContextUtil; +import org.junit.BeforeClass; +import org.junit.Test; + +public final class RuleAlteredJobPreparerTest { + + @BeforeClass + public static void beforeClass() { + PipelineContextUtil.mockModeConfigAndContextManager(); + } + + @Test(expected = PipelineJobCreationException.class) + public void assertPrepareFailedOfNoPrimaryKey() { + new RuleAlteredJobPreparer().prepare(new RuleAlteredJobContext(JobConfigurationBuilder.createJobConfiguration())); + } Review Comment: Why is there no primary key, is it because H2 database? It's not obvious to check it. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorkerTest.java: ########## @@ -44,7 +67,47 @@ public void assertCreateRuleAlteredContextNoAlteredRule() { @Test public void assertCreateRuleAlteredContextSuccess() { JobConfiguration jobConfig = JobConfigurationBuilder.createJobConfiguration(); - final RuleAlteredContext ruleAlteredContext = RuleAlteredJobWorker.createRuleAlteredContext(jobConfig); - Assert.assertNotNull(ruleAlteredContext.getOnRuleAlteredActionConfig()); + RuleAlteredContext ruleAlteredContext = RuleAlteredJobWorker.createRuleAlteredContext(jobConfig); + assertNotNull(ruleAlteredContext.getOnRuleAlteredActionConfig()); Review Comment: Naming: `ruleAlteredContext` could be `actual`. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/scaling/rule_alter/source_rules_config.yaml: ########## @@ -0,0 +1,68 @@ +# +# 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. +# + +- !SHARDING + defaultDatabaseStrategy: + standard: + shardingAlgorithmName: database_inline + shardingColumn: user_id + defaultTableStrategy: + none: '' + keyGenerators: + snowflake: + type: SNOWFLAKE + scaling: + default_scaling: + completionDetector: + props: + incremental-task-idle-minute-threshold: 30 + type: IDLE + dataConsistencyChecker: + props: + chunk-size: 1000 + type: DATA_MATCH + input: + batchSize: 1000 + workerThread: 4 + output: + batchSize: 1000 + workerThread: 4 + streamChannel: + props: + block-queue-size: 10000 + type: MEMORY Review Comment: `input/output/streamChannel` is not necessary to configure, we could just keep the required configuration. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/resources/scaling/rule_alter/target_rules_config.yaml: ########## @@ -0,0 +1,75 @@ +# +# 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. +# + +- !SHARDING + autoTables: + t_order: + actualDataSources: ds_2,ds_3,ds_4 + keyGenerateStrategy: + column: order_id + keyGeneratorName: t_order_snowflake + logicTable: t_order + shardingStrategy: + standard: + shardingAlgorithmName: t_order_hash_mod + shardingColumn: order_id + defaultDatabaseStrategy: + standard: Review Comment: Could we reuse `target_rules_config.yaml`, seems there's the same content yaml file. ########## shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/RuleAlteredJobWorkerTest.java: ########## @@ -44,7 +67,47 @@ public void assertCreateRuleAlteredContextNoAlteredRule() { @Test public void assertCreateRuleAlteredContextSuccess() { JobConfiguration jobConfig = JobConfigurationBuilder.createJobConfiguration(); - final RuleAlteredContext ruleAlteredContext = RuleAlteredJobWorker.createRuleAlteredContext(jobConfig); - Assert.assertNotNull(ruleAlteredContext.getOnRuleAlteredActionConfig()); + RuleAlteredContext ruleAlteredContext = RuleAlteredJobWorker.createRuleAlteredContext(jobConfig); + assertNotNull(ruleAlteredContext.getOnRuleAlteredActionConfig()); + } + + @Test + public void assertRuleAlteredActionEnabled() { + ShardingRuleConfiguration ruleConfiguration = new ShardingRuleConfiguration(); + ruleConfiguration.setScalingName("default_scaling"); + assertTrue(RuleAlteredJobWorker.isOnRuleAlteredActionEnabled(ruleConfiguration)); + } + + @Test + public void assertRuleAlteredActionDisabled() throws IOException, InvocationTargetException, NoSuchMethodException, IllegalAccessException { + URL dataSourceUrl = getClass().getClassLoader().getResource("scaling/detector/datasource_config.yaml"); + assertNotNull(dataSourceUrl); + URL sourceRuleUrl = getClass().getClassLoader().getResource("scaling/rule_alter/source_rules_config.yaml"); + assertNotNull(sourceRuleUrl); + URL targetRuleUrl = getClass().getClassLoader().getResource("scaling/rule_alter/target_rules_config.yaml"); + assertNotNull(targetRuleUrl); + StartScalingEvent startScalingEvent = new StartScalingEvent("logic_db", FileUtils.readFileToString(new File(dataSourceUrl.getFile())), + FileUtils.readFileToString(new File(sourceRuleUrl.getFile())), FileUtils.readFileToString(new File(dataSourceUrl.getFile())), + FileUtils.readFileToString(new File(targetRuleUrl.getFile())), 0, 1); + RuleAlteredJobWorker ruleAlteredJobWorker = new RuleAlteredJobWorker(); + Object result = ReflectionUtil.invokeMethod(ruleAlteredJobWorker, "createJobConfig", new Class[]{StartScalingEvent.class}, new Object[]{startScalingEvent}); + assertTrue(((Optional<?>) result).isPresent()); + } + + @Test + public void assertHasUncompletedJob() throws IOException, InvocationTargetException, NoSuchMethodException, IllegalAccessException { + final JobConfiguration jobConfiguration = JobConfigurationBuilder.createJobConfiguration(); + RuleAlteredJobContext jobContext = new RuleAlteredJobContext(jobConfiguration); + JobProgress finishProcess = new JobProgress(); + finishProcess.setStatus(JobStatus.FINISHED); + jobContext.setInitProgress(finishProcess); + GovernanceRepositoryAPI repositoryAPI = PipelineAPIFactory.getGovernanceRepositoryAPI(); + repositoryAPI.persistJobProgress(jobContext); + URL jobConfigUrl = getClass().getClassLoader().getResource("scaling/rule_alter/scaling_job_config.yaml"); + assertNotNull(jobConfigUrl); Review Comment: Could we just generate `scaling_job_config.yaml` by code, the job configuration structure might be changed frequently. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
