[mysql] Optimize pure binlog phase check logic to improve performance (…

…apache#1392) Co-authored-by: sammieliu <sammieliu@tencent.com>
ruanhang1993 · Jul 28, 2022 · 74f0a65 · 74f0a65
1 parent 2c3962b
commit 74f0a65
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/...c/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java b/...c/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
@@ -43,9 +43,11 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
@@ -74,6 +76,7 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecord, MySqlSpli
     private Map<TableId, List<FinishedSnapshotSplitInfo>> finishedSplitsInfo;
     // tableId -> the max splitHighWatermark
     private Map<TableId, BinlogOffset> maxSplitHighWatermarkMap;
+    private final Set<TableId> pureBinlogPhaseTables;
     private Tables.TableFilter capturedTableFilter;
 
     public BinlogSplitReader(StatefulTaskContext statefulTaskContext, int subTaskId) {
@@ -82,6 +85,7 @@ public BinlogSplitReader(StatefulTaskContext statefulTaskContext, int subTaskId)
                 new ThreadFactoryBuilder().setNameFormat("debezium-reader-" + subTaskId).build();
         this.executor = Executors.newSingleThreadExecutor(threadFactory);
         this.currentTaskRunning = true;
+        this.pureBinlogPhaseTables = new HashSet<>();
     }
 
     public void submitSplit(MySqlSplit mySqlSplit) {
@@ -226,9 +230,13 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
     }
 
     private boolean hasEnterPureBinlogPhase(TableId tableId, BinlogOffset position) {
+        if (pureBinlogPhaseTables.contains(tableId)) {
+            return true;
+        }
         // the existed tables those have finished snapshot reading
         if (maxSplitHighWatermarkMap.containsKey(tableId)
                 && position.isAtOrAfter(maxSplitHighWatermarkMap.get(tableId))) {
+            pureBinlogPhaseTables.add(tableId);
             return true;
         }
         // capture dynamically new added tables
@@ -269,6 +277,7 @@ private void configureFilter() {
         }
         this.finishedSplitsInfo = splitsInfoMap;
         this.maxSplitHighWatermarkMap = tableIdBinlogPositionMap;
+        this.pureBinlogPhaseTables.clear();
     }
 
     public void stopBinlogReadTask() {