Use PipelineJobAPI.getPipelineJobClass() to instead of AbstractPipeli…

…neJobAPIImpl.getJobClassName() (#29018)
apache · Nov 12, 2023 · e70c639 · e70c639
1 parent d4379eb
commit e70c639
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 12 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
@@ -24,6 +24,7 @@
 import org.apache.shardingsphere.data.pipeline.common.context.PipelineJobItemContext;
 import org.apache.shardingsphere.data.pipeline.common.context.PipelineProcessContext;
 import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
+import org.apache.shardingsphere.data.pipeline.common.job.PipelineJob;
 import org.apache.shardingsphere.data.pipeline.common.job.progress.PipelineJobItemProgress;
 import org.apache.shardingsphere.data.pipeline.common.pojo.PipelineJobInfo;
 import org.apache.shardingsphere.data.pipeline.core.task.config.PipelineTaskConfiguration;
@@ -160,6 +161,13 @@ public interface PipelineJobAPI extends TypedSPI {
      */
     void cleanJobItemErrorMessage(String jobId, int shardingItem);
 
+    /**
+     * Get pipeline job class.
+     * 
+     * @return pipeline job class
+     */
+    Class<? extends PipelineJob> getPipelineJobClass();
+
     @Override
     String getType();
 }
diff --git a/...apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractPipelineJobAPIImpl.java b/...apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractPipelineJobAPIImpl.java
@@ -60,13 +60,11 @@ 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), getJobClassName());
+        repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobId), getPipelineJobClass().getName());
         repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(convertJobConfiguration(jobConfig)));
         return Optional.of(jobId);
     }
 
-    protected abstract String getJobClassName();
-
     protected JobConfigurationPOJO convertJobConfiguration(final PipelineJobConfiguration jobConfig) {
         JobConfigurationPOJO result = new JobConfigurationPOJO();
         result.setJobName(jobConfig.getJobId());

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
@@ -122,7 +122,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()), getJobClassName());
+            repositoryAPI.persist(PipelineMetaDataNode.getJobRootPath(jobConfig.getJobId()), getPipelineJobClass().getName());
             JobConfigurationPOJO jobConfigPOJO = convertJobConfiguration(jobConfig);
             jobConfigPOJO.setDisabled(true);
             repositoryAPI.persist(jobConfigKey, YamlEngine.marshal(jobConfigPOJO));
@@ -345,8 +345,8 @@ public PipelineDataConsistencyChecker buildPipelineDataConsistencyChecker(final
     }
 
     @Override
-    protected String getJobClassName() {
-        return CDCJob.class.getName();
+    public Class<CDCJob> getPipelineJobClass() {
+        return CDCJob.class;
     }
 
     @Override

diff --git a/...ardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java b/...ardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
@@ -390,8 +390,8 @@ public PipelineJobInfo getJobInfo(final String jobId) {
     }
 
     @Override
-    protected String getJobClassName() {
-        return ConsistencyCheckJob.class.getName();
+    public Class<ConsistencyCheckJob> getPipelineJobClass() {
+        return ConsistencyCheckJob.class;
     }
 
     @Override

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
@@ -473,12 +473,12 @@ public void refreshTableMetadata(final String jobId, final String databaseName)
     }
 
     @Override
-    public String getType() {
-        return "MIGRATION";
+    public Class<MigrationJob> getPipelineJobClass() {
+        return MigrationJob.class;
     }
 
     @Override
-    protected String getJobClassName() {
-        return MigrationJob.class.getName();
+    public String getType() {
+        return "MIGRATION";
     }
 }