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 2023/11/21 15:04:38 UTC
(shardingsphere) branch master updated: Rename GovernanceRepositoryAPI.isJobConfigurationExisted() (#29115)
This is an automated email from the ASF dual-hosted git repository.
zhaojinchao 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 2715b71a0e9 Rename GovernanceRepositoryAPI.isJobConfigurationExisted() (#29115)
2715b71a0e9 is described below
commit 2715b71a0e9c09b32a3c828a4d5d288d7a234961
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Tue Nov 21 23:04:31 2023 +0800
Rename GovernanceRepositoryAPI.isJobConfigurationExisted() (#29115)
---
.../data/pipeline/common/metadata/node/PipelineMetaDataNode.java | 4 ++--
.../registrycenter/repository/GovernanceRepositoryAPI.java | 8 ++++----
.../registrycenter/repository/GovernanceRepositoryAPIImpl.java | 5 ++---
.../data/pipeline/core/job/service/PipelineJobManager.java | 7 +++----
.../pipeline/common/metadata/node/PipelineMetaDataNodeTest.java | 2 +-
.../shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java | 7 +++----
.../core/job/service/GovernanceRepositoryAPIImplTest.java | 9 ++++-----
7 files changed, 19 insertions(+), 23 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNode.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNode.java
index 19d71a8d791..278c99744b5 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNode.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNode.java
@@ -106,12 +106,12 @@ public final class PipelineMetaDataNode {
}
/**
- * Get job config path.
+ * Get job configuration path.
*
* @param jobId job id
* @return job configuration path
*/
- public static String getJobConfigPath(final String jobId) {
+ public static String getJobConfigurationPath(final String jobId) {
return String.join("/", getJobRootPath(jobId), "config");
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPI.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPI.java
index 2a9e0447fbb..83904a361a9 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPI.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPI.java
@@ -32,12 +32,12 @@ import java.util.Optional;
public interface GovernanceRepositoryAPI {
/**
- * Whether key existing or not.
+ * Whether job configuration existed.
*
- * @param key registry center key
- * @return true if job exists, else false
+ * @param jobId jobId
+ * @return job configuration exist or not
*/
- boolean isExisted(String key);
+ boolean isJobConfigurationExisted(String jobId);
/**
* Persist job offset info.
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPIImpl.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPIImpl.java
index 9071709e745..9e12ee8d9a6 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPIImpl.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/registrycenter/repository/GovernanceRepositoryAPIImpl.java
@@ -51,9 +51,8 @@ public final class GovernanceRepositoryAPIImpl implements GovernanceRepositoryAP
private final ClusterPersistRepository repository;
@Override
- public boolean isExisted(final String key) {
- // TODO delegate to repository isExisted
- return null != repository.getDirectly(key);
+ public boolean isJobConfigurationExisted(final String jobId) {
+ return null != repository.getDirectly(PipelineMetaDataNode.getJobConfigurationPath(jobId));
}
@Override
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
index b01ce97c1f2..e8e1c821481 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -76,13 +76,12 @@ public final class PipelineJobManager {
String jobId = jobConfig.getJobId();
ShardingSpherePreconditions.checkState(0 != jobConfig.getJobShardingCount(), () -> new PipelineJobCreationWithInvalidShardingCountException(jobId));
GovernanceRepositoryAPI repositoryAPI = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId));
- String jobConfigKey = PipelineMetaDataNode.getJobConfigPath(jobId);
- if (repositoryAPI.isExisted(jobConfigKey)) {
- log.warn("jobId already exists in registry center, ignore, jobConfigKey={}", jobConfigKey);
+ if (repositoryAPI.isJobConfigurationExisted(jobId)) {
+ log.warn("jobId already exists in registry center, ignore, job id is `{}`", jobId);
return Optional.of(jobId);
}
repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobId), jobAPI.getJobClass().getName());
- repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(jobConfig.convertToJobConfigurationPOJO()));
+ repositoryAPI.persist(PipelineMetaDataNode.getJobConfigurationPath(jobId), YamlEngine.marshal(jobConfig.convertToJobConfigurationPOJO()));
return Optional.of(jobId);
}
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNodeTest.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNodeTest.java
index d33cf56282c..0dbbc688841 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNodeTest.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/common/metadata/node/PipelineMetaDataNodeTest.java
@@ -67,7 +67,7 @@ class PipelineMetaDataNodeTest {
@Test
void assertGetJobConfigPath() {
- assertThat(PipelineMetaDataNode.getJobConfigPath(jobId), is(jobRootPath + "/config"));
+ assertThat(PipelineMetaDataNode.getJobConfigurationPath(jobId), is(jobRootPath + "/config"));
}
@Test
diff --git a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
index 405d0f5a5ad..17c3daa7a1f 100644
--- a/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
+++ b/kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
@@ -122,14 +122,13 @@ public final class CDCJobAPI implements InventoryIncrementalJobAPI {
CDCJobConfiguration jobConfig = new YamlCDCJobConfigurationSwapper().swapToObject(yamlJobConfig);
ShardingSpherePreconditions.checkState(0 != jobConfig.getJobShardingCount(), () -> new PipelineJobCreationWithInvalidShardingCountException(jobConfig.getJobId()));
GovernanceRepositoryAPI repositoryAPI = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobConfig.getJobId()));
- String jobConfigKey = PipelineMetaDataNode.getJobConfigPath(jobConfig.getJobId());
- if (repositoryAPI.isExisted(jobConfigKey)) {
- log.warn("CDC job already exists in registry center, ignore, jobConfigKey={}", jobConfigKey);
+ if (repositoryAPI.isJobConfigurationExisted(jobConfig.getJobId())) {
+ log.warn("CDC job already exists in registry center, ignore, job id is `{}`", jobConfig.getJobId());
} else {
repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobConfig.getJobId()), getJobClass().getName());
JobConfigurationPOJO jobConfigPOJO = jobConfig.convertToJobConfigurationPOJO();
jobConfigPOJO.setDisabled(true);
- repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(jobConfigPOJO));
+ repositoryAPI.persist(PipelineMetaDataNode.getJobConfigurationPath(jobConfig.getJobId()), YamlEngine.marshal(jobConfigPOJO));
if (!param.isFull()) {
initIncrementalPosition(jobConfig);
}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/job/service/GovernanceRepositoryAPIImplTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/job/service/GovernanceRepositoryAPIImplTest.java
index 60e31f0811c..9c5bca8cf1d 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/job/service/GovernanceRepositoryAPIImplTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/job/service/GovernanceRepositoryAPIImplTest.java
@@ -80,11 +80,10 @@ class GovernanceRepositoryAPIImplTest {
}
@Test
- void assertIsExisted() {
- String testKey = "/testKey1";
- assertFalse(governanceRepositoryAPI.isExisted(testKey));
- governanceRepositoryAPI.persist(testKey, "testValue1");
- assertTrue(governanceRepositoryAPI.isExisted(testKey));
+ void assertIsJobConfigurationExisted() {
+ assertFalse(governanceRepositoryAPI.isJobConfigurationExisted("foo_job"));
+ governanceRepositoryAPI.persist("/pipeline/jobs/foo_job/config", "foo");
+ assertTrue(governanceRepositoryAPI.isJobConfigurationExisted("foo_job"));
}
@Test