Add YamlPipelineJobItemProgressSwapper

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/InventoryIncrementalJobAPI.java

@@ -45,7 +45,7 @@
 public interface InventoryIncrementalJobAPI extends PipelineJobAPI {
 
     @Override
-    default YamlInventoryIncrementalJobItemProgressSwapper getYamlPipelineJobItemProgressSwapper() {
+    default YamlInventoryIncrementalJobItemProgressSwapper getYamlJobItemProgressSwapper() {
         return new YamlInventoryIncrementalJobItemProgressSwapper();
     }
 

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobAPI.java

@@ -46,7 +46,7 @@ public interface PipelineJobAPI extends TypedSPI {
      * @return YAML pipeline job item progress swapper
      */
     @SuppressWarnings("rawtypes")
-    YamlPipelineJobItemProgressSwapper getYamlPipelineJobItemProgressSwapper();
+    YamlPipelineJobItemProgressSwapper getYamlJobItemProgressSwapper();
 
     /**
      * Whether to ignore to start disabled job when job item progress is finished.

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java

@@ -223,7 +223,7 @@ public void updateJobItemProgress(final PipelineJobItemContext jobItemContext) {
 
     @SuppressWarnings("unchecked")
     private String convertJobItemProgress(final PipelineJobItemContext jobItemContext) {
-        return YamlEngine.marshal(jobAPI.getYamlPipelineJobItemProgressSwapper().swapToYamlConfiguration(jobItemContext.toProgress()));
+        return YamlEngine.marshal(jobAPI.getYamlJobItemProgressSwapper().swapToYamlConfiguration(jobItemContext.toProgress()));
     }
 
     /**

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/impl/AbstractInventoryIncrementalJobAPIImpl.java

@@ -133,7 +133,7 @@ public JobOffsetInfo getJobOffsetInfo(final String jobId) {
     @Override
     public Optional<InventoryIncrementalJobItemProgress> getJobItemProgress(final String jobId, final int shardingItem) {
         Optional<String> progress = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProgress(jobId, shardingItem);
-        return progress.map(optional -> getYamlPipelineJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(optional, YamlInventoryIncrementalJobItemProgress.class)));
+        return progress.map(optional -> getYamlJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(optional, YamlInventoryIncrementalJobItemProgress.class)));
     }
 
     @Override
@@ -144,7 +144,7 @@ public void updateJobItemStatus(final String jobId, final int shardingItem, fina
         }
         jobItemProgress.get().setStatus(status);
         PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).updateJobItemProgress(jobId, shardingItem,
-                YamlEngine.marshal(getYamlPipelineJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress.get())));
+                YamlEngine.marshal(getYamlJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress.get())));
     }
 
     @Override

kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java

@@ -176,7 +176,7 @@ private void initIncrementalPosition(final CDCJobConfiguration jobConfig) {
                 IncrementalDumperContext dumperContext = buildDumperContext(jobConfig, i, new TableAndSchemaNameMapper(jobConfig.getSchemaTableNames()));
                 InventoryIncrementalJobItemProgress jobItemProgress = getInventoryIncrementalJobItemProgress(jobConfig, pipelineDataSourceManager, dumperContext);
                 PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).persistJobItemProgress(
-                        jobId, i, YamlEngine.marshal(getYamlPipelineJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress)));
+                        jobId, i, YamlEngine.marshal(getYamlJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress)));
             }
         } catch (final SQLException ex) {
             throw new PrepareJobWithGetBinlogPositionException(jobId, ex);

kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java

@@ -120,7 +120,7 @@ public boolean isIgnoreToStartDisabledJobWhenJobItemProgressIsFinished() {
     @Override
     public Optional<ConsistencyCheckJobItemProgress> getJobItemProgress(final String jobId, final int shardingItem) {
         Optional<String> progress = PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProgress(jobId, shardingItem);
-        return progress.map(s -> getYamlPipelineJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(s, YamlConsistencyCheckJobItemProgress.class, true)));
+        return progress.map(s -> getYamlJobItemProgressSwapper().swapToObject(YamlEngine.unmarshal(s, YamlConsistencyCheckJobItemProgress.class, true)));
     }
 
     @Override
@@ -132,7 +132,7 @@ public void updateJobItemStatus(final String jobId, final int shardingItem, fina
         }
         jobItemProgress.get().setStatus(status);
         PipelineAPIFactory.getGovernanceRepositoryAPI(PipelineJobIdUtils.parseContextKey(jobId)).updateJobItemProgress(jobId, shardingItem,
-                YamlEngine.marshal(getYamlPipelineJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress.get())));
+                YamlEngine.marshal(getYamlJobItemProgressSwapper().swapToYamlConfiguration(jobItemProgress.get())));
     }
 
     /**
@@ -307,7 +307,7 @@ public YamlConsistencyCheckJobConfigurationSwapper getYamlJobConfigurationSwappe
     }
 
     @Override
-    public YamlConsistencyCheckJobItemProgressSwapper getYamlPipelineJobItemProgressSwapper() {
+    public YamlConsistencyCheckJobItemProgressSwapper getYamlJobItemProgressSwapper() {
         return new YamlConsistencyCheckJobItemProgressSwapper();
     }