Refactor usage of InventoryDumperContextSplitter (#32592)

apache · Aug 18, 2024 · a51323f · a51323f
1 parent 7c5c88f
commit a51323f
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/.../shardingsphere/data/pipeline/core/preparer/inventory/splitter/InventoryTaskSplitter.java b/.../shardingsphere/data/pipeline/core/preparer/inventory/splitter/InventoryTaskSplitter.java
@@ -61,7 +61,8 @@ public List<InventoryTask> split(final TransmissionJobItemContext jobItemContext
         List<InventoryTask> result = new LinkedList<>();
         long startTimeMillis = System.currentTimeMillis();
         TransmissionProcessContext processContext = jobItemContext.getJobProcessContext();
-        for (InventoryDumperContext each : new InventoryDumperContextSplitter(sourceDataSource, dumperContext).split(jobItemContext)) {
+        InventoryDumperContextSplitter dumperContextSplitter = new InventoryDumperContextSplitter(sourceDataSource, dumperContext);
+        for (InventoryDumperContext each : dumperContextSplitter.split(jobItemContext)) {
             AtomicReference<IngestPosition> position = new AtomicReference<>(each.getCommonContext().getPosition());
             PipelineChannel channel = InventoryChannelCreator.create(processContext.getProcessConfiguration().getStreamChannel(), importerConfig.getBatchSize(), position);
             Dumper dumper = new InventoryDumper(each, channel, sourceDataSource, jobItemContext.getSourceMetaDataLoader());

diff --git a/...rc/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java b/...rc/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/prepare/CDCJobPreparer.java
@@ -117,8 +117,9 @@ private void initInventoryTasks(final CDCJobItemContext jobItemContext, final At
         CDCTaskConfiguration taskConfig = jobItemContext.getTaskConfig();
         ImporterConfiguration importerConfig = taskConfig.getImporterConfig();
         TransmissionProcessContext processContext = jobItemContext.getJobProcessContext();
-        for (InventoryDumperContext each : new InventoryDumperContextSplitter(jobItemContext.getSourceDataSource(), new InventoryDumperContext(taskConfig.getDumperContext().getCommonContext()))
-                .split(jobItemContext)) {
+        InventoryDumperContextSplitter dumperContextSplitter = new InventoryDumperContextSplitter(
+                jobItemContext.getSourceDataSource(), new InventoryDumperContext(taskConfig.getDumperContext().getCommonContext()));
+        for (InventoryDumperContext each : dumperContextSplitter.split(jobItemContext)) {
             AtomicReference<IngestPosition> position = new AtomicReference<>(each.getCommonContext().getPosition());
             PipelineChannel channel = InventoryChannelCreator.create(processContext.getProcessConfiguration().getStreamChannel(), importerConfig.getBatchSize(), position);
             if (!(position.get() instanceof IngestFinishedPosition)) {