Add YamlPipelineJobItemProgressSwapper

apache · Nov 18, 2023 · 7b15f98 · 7b15f98
1 parent 6c974cf
commit 7b15f98
Show file tree

Hide file tree

Showing 5 changed files with 6 additions and 6 deletions.
diff --git a/...rc/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java b/...rc/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java
@@ -98,7 +98,7 @@ default Optional<String> getToBeStoppedPreviousJobType() {
      * 
      * @return pipeline job class
      */
-    Class<? extends PipelineJob> getPipelineJobClass();
+    Class<? extends PipelineJob> getJobClass();
 
     @Override
     String getType();

diff --git a/...ain/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java b/...ain/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -83,7 +83,7 @@ public Optional<String> start(final PipelineJobConfiguration jobConfig) {
             log.warn("jobId already exists in registry center, ignore, jobConfigKey={}", jobConfigKey);
             return Optional.of(jobId);
         }
-        repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobId), jobAPI.getPipelineJobClass().getName());
+        repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobId), jobAPI.getJobClass().getName());
         repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(jobConfig.convertToJobConfigurationPOJO()));
         return Optional.of(jobId);
     }

diff --git a/...dc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java b/...dc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java
@@ -123,7 +123,7 @@ public String createJob(final StreamDataParameter param, final CDCSinkType sinkT
         if (repositoryAPI.isExisted(jobConfigKey)) {
             log.warn("CDC job already exists in registry center, ignore, jobConfigKey={}", jobConfigKey);
         } else {
-            repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobConfig.getJobId()), getPipelineJobClass().getName());
+            repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobConfig.getJobId()), getJobClass().getName());
             JobConfigurationPOJO jobConfigPOJO = jobConfig.convertToJobConfigurationPOJO();
             jobConfigPOJO.setDisabled(true);
             repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(jobConfigPOJO));
@@ -329,7 +329,7 @@ public PipelineDataConsistencyChecker buildPipelineDataConsistencyChecker(final
     }
 
     @Override
-    public Class<CDCJob> getPipelineJobClass() {
+    public Class<CDCJob> getJobClass() {
         return CDCJob.class;
     }
 

diff --git a/...ardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java b/...ardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
@@ -312,7 +312,7 @@ public YamlConsistencyCheckJobItemProgressSwapper getYamlJobItemProgressSwapper(
     }
 
     @Override
-    public Class<ConsistencyCheckJob> getPipelineJobClass() {
+    public Class<ConsistencyCheckJob> getJobClass() {
         return ConsistencyCheckJob.class;
     }
 

diff --git a/.../org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java b/.../org/apache/shardingsphere/data/pipeline/scenario/migration/api/impl/MigrationJobAPI.java
@@ -445,7 +445,7 @@ public void refreshTableMetadata(final String jobId, final String databaseName)
     }
 
     @Override
-    public Class<MigrationJob> getPipelineJobClass() {
+    public Class<MigrationJob> getJobClass() {
         return MigrationJob.class;
     }