[cdc-base] Optimize pure binlog phase check logic to improve performance (apache#1620)

This closes apache#1620.

Author: ruanhang1993

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/reader/external/JdbcSourceStreamFetcher.java

@@ -39,9 +39,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;
@@ -56,6 +58,7 @@ public class JdbcSourceStreamFetcher implements Fetcher<SourceRecords, SourceSpl
 
     private final JdbcSourceFetchTaskContext taskContext;
     private final ExecutorService executor;
+    private final Set<TableId> pureBinlogPhaseTables;
 
     private volatile ChangeEventQueue<DataChangeEvent> queue;
     private volatile Throwable readException;
@@ -72,6 +75,7 @@ public JdbcSourceStreamFetcher(JdbcSourceFetchTaskContext taskContext, int subTa
         ThreadFactory threadFactory =
                 new ThreadFactoryBuilder().setNameFormat("debezium-reader-" + subTaskId).build();
         this.executor = Executors.newSingleThreadExecutor(threadFactory);
+        this.pureBinlogPhaseTables = new HashSet<>();
     }
 
     @Override
@@ -178,9 +182,13 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
     }
 
     private boolean hasEnterPureBinlogPhase(TableId tableId, Offset 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;
         }
 
@@ -217,5 +225,6 @@ private void configureFilter() {
         }
         this.finishedSplitsInfo = splitsInfoMap;
         this.maxSplitHighWatermarkMap = tableIdBinlogPositionMap;
+        this.pureBinlogPhaseTables.clear();
     }
 }