[FLINK-35874][cdc-connector][mysql] Check pureBinlogPhaseTables set b…

…efore call getBinlogPosition method This closes apache#3488.
ChaomingZhangCN · Jan 3, 2025 · b6c22b5 · b6c22b5
1 parent 3b190b5
commit b6c22b5
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/...rc/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java b/...rc/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
@@ -215,6 +215,9 @@ public void close() {
     private boolean shouldEmit(SourceRecord sourceRecord) {
         if (RecordUtils.isDataChangeRecord(sourceRecord)) {
             TableId tableId = RecordUtils.getTableId(sourceRecord);
+            if (pureBinlogPhaseTables.contains(tableId)) {
+                return true;
+            }
             BinlogOffset position = RecordUtils.getBinlogPosition(sourceRecord);
             if (hasEnterPureBinlogPhase(tableId, position)) {
                 return true;
@@ -254,9 +257,6 @@ 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))) {