datavane/tis#463 将isAfter()代码改回isAtOrAfter(),不然 stop之后再 恢复之后 就不会消费最新的更新消息了

Author: baisui1981

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -292,7 +292,8 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
     private boolean hasEnterPureBinlogPhase(TableId tableId, BinlogOffset position) {
         // the existed tables those have finished snapshot reading
         if (maxSplitHighWatermarkMap.containsKey(tableId)
-                && position.isAfter(maxSplitHighWatermarkMap.get(tableId))) {
+                // 百岁(baisui) 修改2025/07/27 将代码改回去了,不然 stop之后再 恢复之后 就不会消费最新的更新消息了，https://github.com/datavane/tis/issues/463
+                && position.isAtOrAfter(maxSplitHighWatermarkMap.get(tableId))) {
             pureBinlogPhaseTables.add(tableId);
             return true;
         }