Improve pipeline incremental task persist position (#29749)

apache · Jan 17, 2024 · a8e57a0 · a8e57a0
1 parent 9590654
commit a8e57a0
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 10 deletions.
diff --git a/.../org/apache/shardingsphere/data/pipeline/core/importer/SingleChannelConsumerImporter.java b/.../org/apache/shardingsphere/data/pipeline/core/importer/SingleChannelConsumerImporter.java
@@ -22,14 +22,12 @@
 import org.apache.shardingsphere.data.pipeline.core.execute.AbstractPipelineLifecycleRunnable;
 import org.apache.shardingsphere.data.pipeline.core.importer.sink.PipelineSink;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.FinishedRecord;
-import org.apache.shardingsphere.data.pipeline.core.ingest.record.PlaceholderRecord;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.Record;
 import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressListener;
 import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressUpdatedParameter;
 import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 /**
  * Single channel consumer importer.
@@ -50,7 +48,7 @@ public final class SingleChannelConsumerImporter extends AbstractPipelineLifecyc
     @Override
     protected void runBlocking() {
         while (isRunning()) {
-            List<Record> records = channel.fetch(batchSize, timeoutMillis).stream().filter(each -> !(each instanceof PlaceholderRecord)).collect(Collectors.toList());
+            List<Record> records = channel.fetch(batchSize, timeoutMillis);
             if (records.isEmpty()) {
                 continue;
             }

diff --git a/...shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java b/...shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java
@@ -81,7 +81,7 @@ protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final L
                 return;
             }
             if (binlogEvent.get() instanceof PlaceholderEvent) {
-                out.add(binlogEvent);
+                out.add(binlogEvent.get());
                 skipChecksum(binlogEventHeader.getEventType(), in);
                 return;
             }

diff --git a/.../src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/importer/CDCImporter.java b/.../src/main/java/org/apache/shardingsphere/data/pipeline/cdc/core/importer/CDCImporter.java
@@ -27,15 +27,14 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.shardingsphere.data.pipeline.cdc.core.ack.CDCAckId;
 import org.apache.shardingsphere.data.pipeline.cdc.core.ack.CDCAckPosition;
+import org.apache.shardingsphere.data.pipeline.core.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
 import org.apache.shardingsphere.data.pipeline.core.execute.AbstractPipelineLifecycleRunnable;
 import org.apache.shardingsphere.data.pipeline.core.importer.Importer;
 import org.apache.shardingsphere.data.pipeline.core.importer.sink.PipelineSink;
-import org.apache.shardingsphere.data.pipeline.core.channel.PipelineChannel;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.FinishedRecord;
-import org.apache.shardingsphere.data.pipeline.core.ingest.record.PlaceholderRecord;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.Record;
-import org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
 import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressUpdatedParameter;
 import org.apache.shardingsphere.data.pipeline.core.ratelimit.JobRateLimitAlgorithm;
 
@@ -217,15 +216,17 @@ private void doWithoutSorting() {
 
     private void doWithoutSorting(final CDCChannelProgressPair progressPair) {
         PipelineChannel channel = progressPair.getChannel();
-        List<Record> records = channel.fetch(batchSize, timeoutMillis).stream().filter(each -> !(each instanceof PlaceholderRecord)).collect(Collectors.toList());
+        List<Record> records = channel.fetch(batchSize, timeoutMillis);
         if (records.isEmpty()) {
             return;
         }
         Record lastRecord = records.get(records.size() - 1);
-        if (lastRecord instanceof FinishedRecord && records.stream().noneMatch(DataRecord.class::isInstance)) {
+        if (records.stream().noneMatch(DataRecord.class::isInstance)) {
             channel.ack(records);
             progressPair.getJobProgressListener().onProgressUpdated(new PipelineJobProgressUpdatedParameter(0));
-            originalChannelProgressPairs.remove(progressPair);
+            if (lastRecord instanceof FinishedRecord) {
+                originalChannelProgressPairs.remove(progressPair);
+            }
             return;
         }
         if (null != rateLimitAlgorithm) {