Revert "Refactor TransmissionJobItemProgress (apache#32736)" (apache#…

…32737) This reverts commit e98467e.
terrymanu · Aug 30, 2024 · 90b1f6b · 90b1f6b
1 parent e98467e
commit 90b1f6b
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 25 deletions.
diff --git a/...n/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java b/...n/java/org/apache/shardingsphere/data/pipeline/core/job/AbstractSeparablePipelineJob.java
@@ -124,7 +124,7 @@ private boolean execute(final I jobItemContext) {
 
     protected abstract PipelineTasksRunner buildTasksRunner(I jobItemContext);
 
-    private void prepare(final I jobItemContext) {
+    protected final void prepare(final I jobItemContext) {
         try {
             doPrepare(jobItemContext);
             // CHECKSTYLE:OFF

diff --git a/...rg/apache/shardingsphere/data/pipeline/core/job/progress/TransmissionJobItemProgress.java b/...rg/apache/shardingsphere/data/pipeline/core/job/progress/TransmissionJobItemProgress.java
@@ -18,13 +18,13 @@
 package org.apache.shardingsphere.data.pipeline.core.job.progress;
 
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 import org.apache.shardingsphere.data.pipeline.core.context.TransmissionJobItemContext;
 import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
-import org.apache.shardingsphere.data.pipeline.core.task.PipelineTask;
 import org.apache.shardingsphere.data.pipeline.core.task.progress.IncrementalTaskProgress;
 import org.apache.shardingsphere.data.pipeline.core.task.progress.InventoryTaskProgress;
+import org.apache.shardingsphere.data.pipeline.core.task.PipelineTask;
 import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
 
 import java.util.Collection;
@@ -34,22 +34,22 @@
 /**
  * Transmission job item progress.
  */
-@RequiredArgsConstructor
+@NoArgsConstructor
 @Getter
 @Setter
 public final class TransmissionJobItemProgress implements PipelineJobItemProgress {
 
-    private final DatabaseType sourceDatabaseType;
+    private DatabaseType sourceDatabaseType;
 
-    private final String dataSourceName;
+    private String dataSourceName;
 
-    private final JobItemInventoryTasksProgress inventory;
+    private JobItemInventoryTasksProgress inventory;
 
-    private final JobItemIncrementalTasksProgress incremental;
+    private JobItemIncrementalTasksProgress incremental;
 
-    private final long inventoryRecordsCount;
+    private long inventoryRecordsCount;
 
-    private final long processedRecordsCount;
+    private long processedRecordsCount;
 
     private boolean active;
 

diff --git a/.../data/pipeline/core/job/progress/yaml/swapper/YamlTransmissionJobItemProgressSwapper.java b/.../data/pipeline/core/job/progress/yaml/swapper/YamlTransmissionJobItemProgressSwapper.java
@@ -47,12 +47,14 @@ public YamlTransmissionJobItemProgress swapToYamlConfiguration(final Transmissio
 
     @Override
     public TransmissionJobItemProgress swapToObject(final YamlTransmissionJobItemProgress yamlProgress) {
-        TransmissionJobItemProgress result = new TransmissionJobItemProgress(
-                TypedSPILoader.getService(DatabaseType.class, yamlProgress.getSourceDatabaseType()),
-                yamlProgress.getDataSourceName(), inventoryTasksProgressSwapper.swapToObject(yamlProgress.getInventory()),
-                incrementalTasksProgressSwapper.swapToObject(yamlProgress.getSourceDatabaseType(), yamlProgress.getIncremental()),
-                yamlProgress.getInventoryRecordsCount(), yamlProgress.getProcessedRecordsCount());
+        TransmissionJobItemProgress result = new TransmissionJobItemProgress();
         result.setStatus(JobStatus.valueOf(yamlProgress.getStatus()));
+        result.setSourceDatabaseType(TypedSPILoader.getService(DatabaseType.class, yamlProgress.getSourceDatabaseType()));
+        result.setDataSourceName(yamlProgress.getDataSourceName());
+        result.setInventory(inventoryTasksProgressSwapper.swapToObject(yamlProgress.getInventory()));
+        result.setIncremental(incrementalTasksProgressSwapper.swapToObject(yamlProgress.getSourceDatabaseType(), yamlProgress.getIncremental()));
+        result.setProcessedRecordsCount(yamlProgress.getProcessedRecordsCount());
+        result.setInventoryRecordsCount(yamlProgress.getInventoryRecordsCount());
         return result;
     }
 

diff --git a/...pache/shardingsphere/data/pipeline/core/job/progress/PipelineJobProgressDetectorTest.java b/...pache/shardingsphere/data/pipeline/core/job/progress/PipelineJobProgressDetectorTest.java
@@ -34,7 +34,6 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 class PipelineJobProgressDetectorTest {
 
@@ -68,30 +67,31 @@ void assertIsInventoryFinishedWhenCollectionElementIsNull() {
 
     @Test
     void assertIsInventoryFinishedWhenJobCountDoesNotMatchJobItemProgresses() {
-        assertFalse(PipelineJobProgressDetector.isInventoryFinished(2, Collections.singleton(mock(TransmissionJobItemProgress.class))));
+        TransmissionJobItemProgress transmissionJobItemProgress = new TransmissionJobItemProgress();
+        assertFalse(PipelineJobProgressDetector.isInventoryFinished(2, Collections.singleton(transmissionJobItemProgress)));
     }
 
     @Test
     void assertIsInventoryFinishedWhenInventoryTaskProgressHasEmptyMap() {
         JobItemInventoryTasksProgress jobItemInventoryTasksProgress = new JobItemInventoryTasksProgress(Collections.emptyMap());
-        TransmissionJobItemProgress transmissionJobItemProgress = mock(TransmissionJobItemProgress.class);
-        when(transmissionJobItemProgress.getInventory()).thenReturn(jobItemInventoryTasksProgress);
+        TransmissionJobItemProgress transmissionJobItemProgress = new TransmissionJobItemProgress();
+        transmissionJobItemProgress.setInventory(jobItemInventoryTasksProgress);
         assertFalse(PipelineJobProgressDetector.isInventoryFinished(1, Collections.singleton(transmissionJobItemProgress)));
     }
 
     @Test
     void assertIsInventoryFinishedWhenNotAllInventoryTasksCompleted() {
         JobItemInventoryTasksProgress inventoryTasksProgress = new JobItemInventoryTasksProgress(Collections.singletonMap("TEST", new InventoryTaskProgress(new IngestPlaceholderPosition())));
-        TransmissionJobItemProgress transmissionJobItemProgress = mock(TransmissionJobItemProgress.class);
-        when(transmissionJobItemProgress.getInventory()).thenReturn(inventoryTasksProgress);
+        TransmissionJobItemProgress transmissionJobItemProgress = new TransmissionJobItemProgress();
+        transmissionJobItemProgress.setInventory(inventoryTasksProgress);
         assertFalse(PipelineJobProgressDetector.isInventoryFinished(1, Collections.singleton(transmissionJobItemProgress)));
     }
 
     @Test
     void assertIsInventoryFinished() {
         JobItemInventoryTasksProgress inventoryTasksProgress = new JobItemInventoryTasksProgress(Collections.singletonMap("TEST", new InventoryTaskProgress(new IngestFinishedPosition())));
-        TransmissionJobItemProgress transmissionJobItemProgress = mock(TransmissionJobItemProgress.class);
-        when(transmissionJobItemProgress.getInventory()).thenReturn(inventoryTasksProgress);
+        TransmissionJobItemProgress transmissionJobItemProgress = new TransmissionJobItemProgress();
+        transmissionJobItemProgress.setInventory(inventoryTasksProgress);
         assertTrue(PipelineJobProgressDetector.isInventoryFinished(1, Collections.singleton(transmissionJobItemProgress)));
     }
 }
diff --git a/...rio/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java b/...rio/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/CDCJobAPI.java
@@ -207,10 +207,13 @@ private IncrementalDumperContext buildDumperContext(final CDCJobConfiguration jo
 
     private TransmissionJobItemProgress getTransmissionJobItemProgress(final CDCJobConfiguration jobConfig, final PipelineDataSourceManager dataSourceManager,
                                                                        final IncrementalDumperContext incrementalDumperContext) throws SQLException {
+        TransmissionJobItemProgress result = new TransmissionJobItemProgress();
+        result.setSourceDatabaseType(jobConfig.getSourceDatabaseType());
+        result.setDataSourceName(incrementalDumperContext.getCommonContext().getDataSourceName());
         IncrementalTaskPositionManager positionManager = new IncrementalTaskPositionManager(incrementalDumperContext.getCommonContext().getDataSourceConfig().getDatabaseType());
         IncrementalTaskProgress incrementalTaskProgress = new IncrementalTaskProgress(positionManager.getPosition(null, incrementalDumperContext, dataSourceManager));
-        return new TransmissionJobItemProgress(jobConfig.getSourceDatabaseType(), incrementalDumperContext.getCommonContext().getDataSourceName(), null,
-                new JobItemIncrementalTasksProgress(incrementalTaskProgress), 0L, 0L);
+        result.setIncremental(new JobItemIncrementalTasksProgress(incrementalTaskProgress));
+        return result;
     }
 
     /**