This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new ea71b2302 [FLINK-35874][cdc-connector][mysql] Check
pureBinlogPhaseTables set before call getBinlogPosition method
ea71b2302 is described below
commit ea71b2302ddc5f9b7be65843dbf3f5bed4ca9d8e
Author: Zhongmin Qiao <[email protected]>
AuthorDate: Wed Jul 24 11:30:40 2024 +0800
[FLINK-35874][cdc-connector][mysql] Check pureBinlogPhaseTables set before
call getBinlogPosition method
This closes #3488.
---
.../cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
index 61f4ce9a9..e1ef3895f 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
@@ -215,6 +215,9 @@ public class BinlogSplitReader implements
DebeziumReader<SourceRecords, MySqlSpl
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 @@ public class BinlogSplitReader implements
DebeziumReader<SourceRecords, MySqlSpl
}
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))) {
