[FLINK-37191][mysql] Avoid back filling if the lowWatermark of BinlogSplit is equal to highWatermark.

lvyanquan · lvyanquan · commit 0312e6a4357d · 2025-02-04T21:35:44.000+08:00
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java
@@ -255,6 +255,11 @@ public int compareTo(BinlogOffset that) {
             return Long.compare(this.getRestartSkipRows(), that.getRestartSkipRows());
         }
 
+        // serverId of BinlogOffset in snapshot phase is 0.
+        if (serverId == 0) {
+            return 0;
+        }
+
         // The skip rows are the same, so compare the timestamp ...
         return Long.compare(this.getTimestampSec(), that.getTimestampSec());
     }

Original file line number	Diff line number	Diff line change
`@@ -255,6 +255,11 @@ public int compareTo(BinlogOffset that) {`
`255`	`255`	`return Long.compare(this.getRestartSkipRows(), that.getRestartSkipRows());`
`256`	`256`	`}`
`257`	`257`
	`258`	`+ // serverId of BinlogOffset in snapshot phase is 0.`
	`259`	`+ if (serverId == 0) {`
	`260`	`+ return 0;`
	`261`	`+ }`
	`262`	`+`
`258`	`263`	`// The skip rows are the same, so compare the timestamp ...`
`259`	`264`	`return Long.compare(this.getTimestampSec(), that.getTimestampSec());`
`260`	`265`	`}`