ricky2129 opened a new issue, #12554: URL: https://github.com/apache/seatunnel/issues/12554
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened On a MySQL source that carries a GTID lineage inherited from a previous master, a MySQL-CDC job restored from a checkpoint re-reads everything the server still retains for that inherited lineage. In our case four jobs walked back roughly 100 binlog files and took about two and a half days to catch up. Nothing was lost, but the duplicate volume and the time to re-consume it were significant. The cause is in `MySqlStreamingChangeEventSource#filterGtidSet`: ```java mergedGtidSet = GtidUtils.fixRestoredGtidSet( GtidUtils.mergeGtidSetInto( relevantAvailableServerGtidSet.retainAll( uuid -> knownGtidSet.forServerWithId(uuid) != null), purgedServerGtid), filteredGtidSet); ``` `retainAll` drops any lineage that the restored GTID set does not mention. Only `purgedServerGtid` brings it back, and it reaches no further than `gtid_purged`, so everything the server retains past the purge point looks unconsumed and is sent again. ### How the checkpoint ends up without the lineage `startup.mode = specific` builds the starting offset from `startup.specific-offset.file` and `startup.specific-offset.pos`, with no GTID set. From then on the recorded set only grows from the GTID events the job actually reads, so a lineage that stopped receiving writes before the start position is never recorded. `startup.mode = earliest` and `timestamp` reach the same state. This needs a server that carries an inherited lineage next to its own, which is the normal state after a replica promotion, a restore from backup, or an RDS Blue/Green cutover. A job started with `initial` or `latest` seeds the server's full `gtid_executed` and is not affected. `startup.specific-offset.gtid-set` (#11123) lets an operator supply a complete set at first submit, which avoids the trigger, but it does not change the merge. A set can also stop being complete later, and the merge runs on every restore. ### Reproducing it `A` is the lineage the server is writing and the checkpoint tracks. `B` is inherited and dormant. ```java GtidSet server = new GtidSet("A:1-100,B:1-1000"); GtidSet purged = new GtidSet("B:1-900"); GtidSet restored = new GtidSet("A:9-90"); GtidSet merged = GtidUtils.fixRestoredGtidSet( GtidUtils.mergeGtidSetInto( server.retainAll(uuid -> restored.forServerWithId(uuid) != null), purged), restored); System.out.println(merged); // A:1-90,B:1-900 ``` `B` is claimed to 900 while the server holds 1000, so those 100 transactions are re-delivered. At production scale that difference was around 15 million. In the logs the signature is the lineage the checkpoint tracks sitting at its recorded position while the other one climbs a binlog file at a time. It also stops happening once a job has replayed through the lineage, since by then it has recorded it. ### What I would expect A lineage that was fully executed before the position being resumed from should be treated as consumed. A lineage that is genuinely new and active, as after a failover, should still be read from its earliest available position. Those two cases cannot be told apart from the GTID sets alone, since the lineage is missing from the checkpoint in both. What separates them is binlog order, and the `Previous_gtids` event of the resume file is exactly the set executed before that file: ```sql SHOW BINLOG EVENTS IN '<resume file>' LIMIT 3; -- row 2 is Previous_gtids, the set is in Info ``` I checked this on MySQL 8.4.5. A dormant lineage rotated into an older file does come back at its full range. The value carries a newline after the comma when more than one lineage is present. The same merge exists in upstream Debezium and I have reported it there as debezium/dbz#2731 with a fix in debezium/debezium#8135. SeaTunnel's copy needs its own change because `GtidUtils.mergeGtidSetInto` keeps the entry already held for a UUID rather than replacing it, so `Previous_gtids` has to be merged before the purged set or it is silently discarded. ### SeaTunnel Version dev, and 2.3.x. The merge is unchanged since #8528. ### SeaTunnel Config ```conf source { MySQL-CDC { base-url = "jdbc:mysql://<host>:3306/<db>" startup.mode = "specific" startup.specific-offset.file = "mysql-bin.000110" startup.specific-offset.pos = 437083895 } } ``` ### Running Command ```shell ./bin/seatunnel.sh --config <job>.conf ``` ### Error Exception ```log No exception. The job runs and re-reads the inherited lineage from the oldest binlog the server still retains. ``` ### Zeta or Flink or Spark Version Zeta ### Java or Scala Version Java 8 ### Screenshots _No response_ ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
