Refactor CDCJob and MigrationJob (#29361)

* Refactor CDCJob * Refactor MigrationJob
apache · Dec 11, 2023 · bc68578 · bc68578
1 parent 4cb0cd0
commit bc68578
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 19 deletions.
diff --git a/...e/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java b/...e/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java
@@ -71,24 +71,20 @@
 @Slf4j
 public final class CDCJob extends AbstractInseparablePipelineJob<CDCJobItemContext> {
 
-    @Getter
-    private final PipelineSink sink;
+    private final CDCJobAPI jobAPI = (CDCJobAPI) TypedSPILoader.getService(TransmissionJobAPI.class, "STREAMING");
 
-    private final CDCJobAPI jobAPI;
+    private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager = new PipelineJobItemManager<>(new CDCJobType().getYamlJobItemProgressSwapper());
 
-    private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager;
+    private final PipelineProcessConfigurationPersistService processConfigPersistService = new PipelineProcessConfigurationPersistService();
 
-    private final PipelineProcessConfigurationPersistService processConfigPersistService;
+    private final CDCJobPreparer jobPreparer = new CDCJobPreparer();
 
-    private final CDCJobPreparer jobPreparer;
+    @Getter
+    private final PipelineSink sink;
 
     public CDCJob(final PipelineSink sink) {
         super(new PipelineJobRunnerManager(new CDCJobRunnerCleaner(sink)));
         this.sink = sink;
-        jobAPI = (CDCJobAPI) TypedSPILoader.getService(TransmissionJobAPI.class, "STREAMING");
-        jobItemManager = new PipelineJobItemManager<>(new CDCJobType().getYamlJobItemProgressSwapper());
-        processConfigPersistService = new PipelineProcessConfigurationPersistService();
-        jobPreparer = new CDCJobPreparer();
     }
 
     @Override

diff --git a/...rc/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java b/...rc/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/MigrationJob.java
@@ -63,18 +63,12 @@
 @Slf4j
 public final class MigrationJob extends AbstractSeparablePipelineJob<MigrationJobItemContext> {
 
-    private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager;
+    private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager = new PipelineJobItemManager<>(new MigrationJobType().getYamlJobItemProgressSwapper());
 
-    private final PipelineProcessConfigurationPersistService processConfigPersistService;
+    private final PipelineProcessConfigurationPersistService processConfigPersistService = new PipelineProcessConfigurationPersistService();
 
     // Shared by all sharding items
-    private final MigrationJobPreparer jobPreparer;
-
-    public MigrationJob() {
-        jobItemManager = new PipelineJobItemManager<>(new MigrationJobType().getYamlJobItemProgressSwapper());
-        processConfigPersistService = new PipelineProcessConfigurationPersistService();
-        jobPreparer = new MigrationJobPreparer();
-    }
+    private final MigrationJobPreparer jobPreparer = new MigrationJobPreparer();
 
     @Override
     protected MigrationJobItemContext buildJobItemContext(final ShardingContext shardingContext) {