You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mielientiev (via GitHub)" <gi...@apache.org> on 2025/01/08 12:04:40 UTC
[PR] [FLINK-37065]: MySQL cdc, fix method 'fixRestoredGtidSet' to support gt… [flink-cdc]
mielientiev opened a new pull request, #3845:
URL: https://github.com/apache/flink-cdc/pull/3845
This PR closes: [FLINK-37065](https://issues.apache.org/jira/browse/FLINK-37065) ticket.
The new `fixRestoredGtidSet` implementation preserves GTID gaps by merging the server and restored GTID sets interval-by-interval, ensuring no unprocessed transactions are skipped or flattened.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3185001849
@lvyanquan Do you know how to retrigger pipeline? seems like false failure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3172323028
Can we add another test to demonstrate that such GTIDs can be used to establish a binary log connection and read the intermediate transactions to avoid data loss?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2278142126
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
There is also a risk of data loss here, right?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3161708575
@leonardBang could you please review it as well?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2278148619
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
The operation here likely arises because the GTIDs in our state are merged from binlog events and do not include GTIDs that were executed prior to the current binlog entries.
Thus, if we can use the `PreviousGtidsEvent` to supplement the GTIDs that were already executed, would this operation become unnecessary? (Of course, we must also consider compatibility with historical jobs—removing it could lead to redundant data retrieval.)
Example code for handling `PreviousGtidsEvent` in `MySqlStreamingChangeEventSource`:
```
protected void handlePreviousGtidsEvent(MySqlOffsetContext offsetContext, Event event) {
PreviousGtidSetEventData previousGtidSetEventData = unwrapData(event);
LOGGER.info(
"Handling PREVIOUS_GTID_EVENT: {}. Current GTID set: {}",
previousGtidSetEventData,
gtidSet);
com.github.shyiko.mysql.binlog.GtidSet previousGtidSet =
new com.github.shyiko.mysql.binlog.GtidSet(previousGtidSetEventData.getGtidSet());
if (previousGtidSet.isContainedWithin(gtidSet)) {
LOGGER.info(
"Current GTID set already contains previous GTID set and will not be updated.");
return;
}
LOGGER.info(
"Updating GTID set with previous GTID set: {}",
previousGtidSetEventData.getGtidSet());
gtidSet = previousGtidSet;
offsetContext.setCompletedGtidSet(gtidSet.toString());
}
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303851624
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> > When starting with the 'scan.startup.mode' = 'earliest-offset', after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2".
>
> I'm not sure that its 100% correct. Because in case of `earliest-offset`, Pipeline should receive all data from the beginning, in this case `A:1-182571` and then 182580, 182579, 182578.
>
> So after 182580 tx, saved GTID Set should be `A:1-182571:182580-182580` "row": "1", "event": "2". And in this case after restart "merged" version should be still `A:1-182571:182580-182580`
I have did a test and get the above result.
```
2025-08-27 20:39:40.394 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127165000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=316, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182580'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127179000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=634, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182579'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127436000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=952, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182578'}}
```
<img width="1917" height="470" alt="image" src="https://github.com/user-attachments/assets/560b2404-50d0-492c-9bf7-bfb98f0c74dc" />
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc fix method 'fixRestoredGtidSet' to support gt… [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-2589582425
Hey @ruanhang1993
Just wanted to ask if you could review my PR.
Thanks
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2291165263
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
Hi, @mielientiev.
Can you help confirm my question? I am willing to push forward with this PR, but we need to consider all possible risks.
CC @lzshlzsh also.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303231371
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> Considering that the gtid in the state may be incomplete, I think the merging is acceptable. But we need to add handling for the `PreviousGtidsEvent` so that we can use cleaner methods to deal with this in future versions.
In my opinion, PreviousGtidsEvent cannot solve the problem of small risk of data loss. Merging the gtid on the server side that precedes the minimum restored gtid may result in small risk of data loss. This merging logic exists before this PR. I mentioned a case of data loss above. Let me briefly explain it here.
1. There is only one binlog file avaible in mysql, gtids are out of order, 182580182579182578, corresponding to three update data. PreviousGtidsEvent is A:1-182571. When starting with the `'scan.startup.mode' = 'earliest-offset'`, after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2". The gtid known to the server from the `show master status` command is A:1-1825771:182578-182580. That is, 182578 has not been read.
2. When recovering from the checkpoint, the gtid of the merged to connect to MySQL is A:1-1825771:182578-182580, and 182578 data will not be read, thus it is lost.
PreviousGtidsEvent A:1-182571 could not be used to avoid the data loss.
Beside that I think the merging is acceptable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3148900794
@mielientiev Thanks for your contribution, I'll review this PR ASAP. Before the review work, could you kindly rebase to latest master branch to fix current CI failed cases?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3146501691
@leonardBang @xiaom @ruanhang1993 Could you please review this PR?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303231371
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> Considering that the gtid in the state may be incomplete, I think the merging is acceptable. But we need to add handling for the `PreviousGtidsEvent` so that we can use cleaner methods to deal with this in future versions.
In my opinion, PreviousGtidsEvent cannot solve the problem of data loss with small probability. Merging the gtid on the server side that precedes the minimum restored gtid may result in small risk of data loss. This merging logic exists before this PR. I mentioned a case of data loss above. Let me briefly explain it here.
1. There is only one binlog file avaible in mysql, gtids are out of order, 182580182579182578, corresponding to three update data. PreviousGtidsEvent is A:1-182571. When starting with the `'scan.startup.mode' = 'earliest-offset'`, after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2". The gtid known to the server from the `show master status` command is A:1-1825771:182578-182580. That is, 182578 has not been read.
2. When recovering from the checkpoint, the gtid of the merged to connect to MySQL is A:1-1825771:182578-182580, and 182578 data will not be read, thus it is lost.
PreviousGtidsEvent A:1-182571 could not be used to avoid the data loss.
Beside that I think the merging is acceptable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3156717172
@leonardBang done - rebased! but now I don't see status of pipelines
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3163909892
Hi @mielientiev, I could understand the problem you are hoping to solve, but I don't understand why https://github.com/apache/flink-cdc/pull/2220 not addressing this issue, as it said:
```
When MySQL is under a dual-master architecture, GTIDs may have gaps, similar to A:1-102, 105-150.
Such gaps are temporary but will eventually be consistent.
But if the CDC happens to recover from the checkpoint when there are gaps in MySQL's GTID, it may access non-existent transactions, leading to recovery failure.
```
Is it because the previous analysis was not entirely accurate?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2313152939
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
@lvyanquan Let's maybe create a separate issue for this edge case and merge this PR, since this edge case is still possible in existing implementation. WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3306441850
Thanks @mielientiev @lzshlzsh for the fix.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2301824397
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
As far as I remember there shouldn't be any risk of data loss in this case.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303851624
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> > When starting with the 'scan.startup.mode' = 'earliest-offset', after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2".
>
> I'm not sure that its 100% correct. Because in case of `earliest-offset`, Pipeline should receive all data from the beginning, in this case `A:1-182571` and then 182580, 182579, 182578.
>
> So after 182580 tx, saved GTID Set should be `A:1-182571:182580-182580` "row": "1", "event": "2". And in this case after restart "merged" version should be still `A:1-182571:182580-182580`
I have did a test and get the above result. It seems that #MySqlStreamingChangeEventSource update effectiveOffsetContext on GTID event, and not receive `A:1-182571` that has been purged.
Maybe we should process PreviousGtidsEvent in case of `earliest-offset`. @lvyanquan
```
2025-08-27 20:39:40.394 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127165000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=316, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182580'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127179000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=634, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182579'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127436000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=952, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182578'}}
```
<img width="1917" height="470" alt="image" src="https://github.com/user-attachments/assets/560b2404-50d0-492c-9bf7-bfb98f0c74dc" />
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256038312
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
tip: Maybe simple source id like A that the original tests use is simple and clear.
use
```
A:1-100
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256011828
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
The function comment said: During the correction process, the restored GTID set is adjusted according to the server's GTID set to ensure it does not exceed the latter. Event though I do not kown why this could happen, it is not bad to dispose it. Maybe we can add the following test:
```
Arguments.of(
"Restored end exceeds server range",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-110",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100")
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2278142126
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
There is also a risk of data loss here as we cannot confirm whether this gtid has been executed(although the probability seems very small), right?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2357280751
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
I agree that before finding a complete fix, this is a better solution than the original implementation, which can avoid most of the data loss issues that existed before.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3307145083
The remaining unhandled edge cases will be traced in https://issues.apache.org/jira/browse/FLINK-38380.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303667843
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> When starting with the 'scan.startup.mode' = 'earliest-offset', after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2".
I'm not sure that its 100% correct. Because in case of `earliest-offset`, Pipeline should receive all data from the beginning, in this case `A:1-182571` and then 182580, 182579, 182578.
So after 182580 tx, saved GTID Set should be `A:1-182571:182580-182580` "row": "1", "event": "2". And in this case after restart "merged" version should be still `A:1-182571:182580-182580`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2297204132
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
@mielientiev @lzshlzsh Would you like to comment here when you have time ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2291165263
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
Hi, @mielientiev.
Can you help confirm my question? I am willing to push forward with this PR, but we need to consider all possible risks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256011828
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
The method comment said: During the correction process, the restored GTID set is adjusted according to the server's GTID set to ensure it does not exceed the latter. Event though I do not kown why this could happen, it is not bad to dispose it. Maybe we can add the following test:
```
Arguments.of(
"Restored end exceeds server range",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-110",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100")
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256038312
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
tips:
1. Maybe source id like A that the original tests use is simple and clear.
```
A:1-100
```
2. We may modify tests to have multiple sourceId like server: 'A:1-100:102-200,B:20-200' , restored 'A:106-150,C:1-100', result: 'A:1-100:102-150,B:20-200,C:1-100'
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc fix method 'fixRestoredGtidSet' to support gt… [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-2993462563
@leonardBang @xiaom @ruanhang1993 Can you please review it? I think this is a very important PR since MySQLCDC can lose/skip data during recovering from the checkpoint.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2301824397
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
As far as I remember there shouldn't be any risk of data loss in this case.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3172892389
@lvyanquan sure. Just added the test with binlog connector. Please review it again and let me know what do you think
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2298111455
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> There is also a risk of data loss here as we cannot confirm whether this gtid has been executed(although the probability seems very small), right?
Yes, you are right, we can construct the following data loss case:
```
mysql> show master status;
+------------------+----------+--------------+------------------+------------------------------------------------------------------------------------------+
| File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+------------------------------------------------------------------------------------------+
| mysql-bin.000197 | 237 | | | ad268c5d-2f18-11ef-8eac-0242ac120003:9-12,
fc992a75-c2a9-11ee-82e7-0242ac120004:1-182571 |
+------------------+----------+--------------+------------------+------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
mysql> set gtid_next='fc992a75-c2a9-11ee-82e7-0242ac120004:182580';
Query OK, 0 rows affected (0.01 sec)
mysql> update full_types_bitsail_2x set tiny_c = 1 where id = 10;
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> set gtid_next='fc992a75-c2a9-11ee-82e7-0242ac120004:182579';
Query OK, 0 rows affected (0.00 sec)
mysql> update full_types_bitsail_2x set tiny_c = 2 where id = 10;
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
-- do a savepoint: {"transaction_id":null,"ts_sec":"1756127179","file":"mysql-bin.000197","pos":"634","kind":"SPECIFIC","gtids":"fc992a75-c2a9-11ee-82e7-0242ac120004:182580-182580","row":"1","event":"2","server_id":"123"}
mysql> set gtid_next='fc992a75-c2a9-11ee-82e7-0242ac120004:182578';
Query OK, 0 rows affected (0.00 sec)
mysql> update full_types_bitsail_2x set tiny_c = 3 where id = 10;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> show master status;
+------------------+----------+--------------+------------------+--------------------------------------------------------------------------------------------------------+
| File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+--------------------------------------------------------------------------------------------------------+
| mysql-bin.000197 | 1191 | | | ad268c5d-2f18-11ef-8eac-0242ac120003:9-12,
fc992a75-c2a9-11ee-82e7-0242ac120004:1-182571:182578-182580 |
+------------------+----------+--------------+------------------+--------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)
-- restore from savepoint, update set tiny_c = 3 lost, and because of "event":"2", 2 following events are dropped also.
GTID set available on server: ad268c5d-2f18-11ef-8eac-0242ac120003:9-12,fc992a75-c2a9-11ee-82e7-0242ac120004:1-182571:182578-182580
Final merged GTID set to use when connecting to MySQL: ad268c5d-2f18-11ef-8eac-0242ac120003:9-12,fc992a75-c2a9-11ee-82e7-0242ac120004:1-182571:182578-182580
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303231371
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> Considering that the gtid in the state may be incomplete, I think the merging is acceptable. But we need to add handling for the `PreviousGtidsEvent` so that we can use cleaner methods to deal with this in future versions.
In my opinion, PreviousGtidsEvent cannot solve the problem of small risk of data loss. Merging the gtid on the server side that precedes the minimum restored gtid may result in small risk of data loss. This merging logic exists before this PR. I mentioned a case of data loss above. Let me briefly explain it here.
1. There is only one binlog file avaible in mysql, gtids are out of order, 182580, 182579, 182578, corresponding to three update data. PreviousGtidsEvent is A:1-182571. When starting with the `'scan.startup.mode' = 'earliest-offset'`, after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2". The gtid known to the server from the `show master status` command is A:1-1825771:182578-182580. That is, 182578 has not been read.
2. When recovering from the checkpoint, the gtid of the merged to connect to MySQL is A:1-1825771:182578-182580, and 182578 data will not be read, thus it is lost.
PreviousGtidsEvent A:1-182571 could not be used to avoid the data loss.
Beside that I think the merging is acceptable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3305198667
Hi @ruanhang1993 could you take a look at this?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2302998216
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
But why it will be lost? As far as I remember GTID will be added to "handled/processed" GTID Set only after transaction is complete (XID event).
https://github.com/apache/flink-cdc/blob/f6197d223fdc4710200e9e54b69ff3a57d2202f9/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java#L689
https://github.com/debezium/debezium/blob/v1.9.8.Final/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlOffsetContext.java#L339
Or maybe I'm missing use case? Could you please elaborate in more detail?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303851624
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> > When starting with the 'scan.startup.mode' = 'earliest-offset', after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2".
>
> I'm not sure that its 100% correct. Because in case of `earliest-offset`, Pipeline should receive all data from the beginning, in this case `A:1-182571` and then 182580, 182579, 182578.
>
> So after 182580 tx, saved GTID Set should be `A:1-182571:182580-182580` "row": "1", "event": "2". And in this case after restart "merged" version should be still `A:1-182571:182580-182580`
I have did a test and get the above result. It seems that #MySqlStreamingChangeEventSource update effectiveOffsetContext on GTID, and not receive `A:1-182571` that has been purged.
Maybe we should process PreviousGtidsEvent in case of `earliest-offset`. @lvyanquan
```
2025-08-27 20:39:40.394 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127165000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=316, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182580'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127179000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=634, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182579'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127436000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=952, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182578'}}
```
<img width="1917" height="470" alt="image" src="https://github.com/user-attachments/assets/560b2404-50d0-492c-9bf7-bfb98f0c74dc" />
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303851624
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> > When starting with the 'scan.startup.mode' = 'earliest-offset', after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2".
>
> I'm not sure that its 100% correct. Because in case of `earliest-offset`, Pipeline should receive all data from the beginning, in this case `A:1-182571` and then 182580, 182579, 182578.
>
> So after 182580 tx, saved GTID Set should be `A:1-182571:182580-182580` "row": "1", "event": "2". And in this case after restart "merged" version should be still `A:1-182571:182580-182580`
I have did a test and get the above result. It seems that #MySqlStreamingChangeEventSource update effectiveOffsetContext on GTID, and not receive `A:1-182571` that has been purged.
```
2025-08-27 20:39:40.394 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127165000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=316, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182580'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127179000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=634, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182579'}}
2025-08-27 20:39:40.401 [blc-127.0.0.1:3308] MySqlStreamingChangeEventSource.java:558 DEBUG io.debezium.connector.mysql.MySqlStreamingChangeEventSource - GTID transaction: Event{header=EventHeaderV4{timestamp=1756127436000, eventType=GTID, serverId=123, headerLength=19, dataLength=60, nextPosition=952, flags=0}, data=GtidEventData{flags=0, gtid='fc992a75-c2a9-11ee-82e7-0242ac120004:182578'}}
```
<img width="1917" height="470" alt="image" src="https://github.com/user-attachments/assets/560b2404-50d0-492c-9bf7-bfb98f0c74dc" />
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2300353087
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
Considering that the gtid in the state may be incomplete, I think the merging is acceptable. But we need to add handling for the `PreviousGtidsEvent` so that we can use cleaner methods to deal with this in future versions.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2258292595
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
@lzshlzsh Thanks for the feedback
It's actually a very valid point. I think it may happen if you have a pool of MySQL hosts, and some replication delay. And imagine that your pipeline got restarted and reconnected to "lagging" host that haven't yet applied processed transactions, that your pipeline "saw" in previous run
I implemented intersection logic with serverId range
Also I preserved original test logic like this `A:1-100`. Indeed, it looks simpler.
Please review it again
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2258292595
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
@lzshlzsh Thanks for the feedback
It's actually a very valid point. I think it may happen if you have a pool of MySQL hosts, and some replication delay. And imagine that your pipeline got restarted and reconnected to "lagging" host that haven't yet applied processed transactions.
I implemented intersection logic with serverId range
Also I preserved original test logic like this `A:1-100`. Indeed, it looks simpler.
Please review it again
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc, fix method 'fixRestoredGtidSet' to support gt… [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-2577532661
@ruanhang1993 Would you like to help review this PR?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256011828
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
The comments said: During the correction process, the restored GTID set is adjusted according to the server's GTID set to ensure it does not exceed the latter. Event though I do not kown why this could happen, it is not bad to dispose it. Maybe we can add the following test:
```
Arguments.of(
"Restored end exceeds server range",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-110",
"e21e8354-04b8-42a9-bfe3-12d71e809836:1-100")
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256038312
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
tip:
1. Maybe source id like A that the original tests use is simple and clear.
```
A:1-100
```
2. We may modify tests to have multiple sourceId like server: 'A:1-100:102-200,B:20-200' , restored 'A:106-150,C:1-100', result: 'A:1-100:102-150,B:20-200,C:1-100'
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2256038312
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,53 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
Review Comment:
tip: Maybe source id like A that the original tests use is simple and clear.
use
```
A:1-100
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3174606105
LGTM.
A small problem, you need to add license to the file MysqlGtidRecoveryTest.java
```
Error: Failed to execute goal org.apache.rat:apache-rat-plugin:0.12:check (default) on project flink-cdc-parent: Too many files with unapproved license: 1 See RAT report in: /home/runner/work/flink-cdc/flink-cdc/target/rat.txt -> [Help 1]
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#issuecomment-3165205505
@lvyanquan The analysis is partially correct. MySQL servers can have gaps in their executed transactions (e.g., `A:1-102,105-150`), and these gaps may or may not get filled later. This isn't a problem.
However, there's something the analysis missed. When MySQL processes transactions in parallel, they can appear out of order in the binlog. For example, even though the server executed transactions `A:1-10`, the binlog might show them like this when sending to Flink CDC:
```
-Tx, A:1
-Tx, A:2
-Tx, A:4
-Tx, A:6
-Tx, A:5
-Tx, A:3
-Tx, A:7
-Tx, A:8
-Tx, A:10
-Tx, A:9
```
Now imagine our application stopped and saved its state after processing transaction `A:6`. The restored GTID set would be `A:1-2:4:6` (we've seen 1, 2, 4, and 6, but not 3 or 5 yet).
When we restart, we need to tell MySQL which transactions we've already processed so it knows what to send next. The problem is that the current code merges these gaps and produces `A:1-6` as restored value. This tells MySQL we've processed ALL transactions from 1 to 6, including 3 and 5. MySQL then skips transactions 3 and 5, causing data loss.
I hope it's clear now
Also @lzshlzsh did a great description for this issue here https://issues.apache.org/jira/browse/FLINK-38183
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2278142126
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
There is also a risk of data loss here, as we cannot confirm whether this gtid has been executed, right?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
There is also a risk of data loss here as we cannot confirm whether this gtid has been executed, right?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lzshlzsh (via GitHub)" <gi...@apache.org>.
lzshlzsh commented on code in PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845#discussion_r2303231371
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java:
##########
@@ -36,36 +38,64 @@ public class GtidUtils {
public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restoredGtidSet) {
Map<String, GtidSet.UUIDSet> newSet = new HashMap<>();
serverGtidSet.getUUIDSets().forEach(uuidSet -> newSet.put(uuidSet.getUUID(), uuidSet));
- for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
- GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.getUUID());
+ for (GtidSet.UUIDSet restoredUuidSet : restoredGtidSet.getUUIDSets()) {
+ GtidSet.UUIDSet serverUuidSet = newSet.get(restoredUuidSet.getUUID());
if (serverUuidSet != null) {
- long restoredIntervalEnd = getIntervalEnd(uuidSet);
- List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
- new ArrayList<>();
- for (GtidSet.Interval serverInterval : serverUuidSet.getIntervals()) {
- if (serverInterval.getEnd() <= restoredIntervalEnd) {
- newIntervals.add(
- new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), serverInterval.getEnd()));
- } else if (serverInterval.getStart() <= restoredIntervalEnd
- && serverInterval.getEnd() > restoredIntervalEnd) {
- newIntervals.add(
+ List<GtidSet.Interval> serverIntervals = serverUuidSet.getIntervals();
+ List<GtidSet.Interval> restoredIntervals = restoredUuidSet.getIntervals();
+
+ long earliestRestoredTx = getMinIntervalStart(restoredIntervals);
+
+ List<com.github.shyiko.mysql.binlog.GtidSet.Interval> merged = new ArrayList<>();
+
+ // Process each server interval
+ for (GtidSet.Interval serverInterval : serverIntervals) {
+ // First, check if any part comes before earliest restored
+ if (serverInterval.getStart() < earliestRestoredTx) {
+ long end = Math.min(serverInterval.getEnd(), earliestRestoredTx - 1);
+ merged.add(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
- serverInterval.getStart(), restoredIntervalEnd));
+ serverInterval.getStart(), end));
Review Comment:
> Considering that the gtid in the state may be incomplete, I think the merging is acceptable. But we need to add handling for the `PreviousGtidsEvent` so that we can use cleaner methods to deal with this in future versions.
In my opinion, PreviousGtidsEvent cannot solve the problem of small risk of data loss. Merging the gtid on the server side that precedes the minimum restored gtid may result in small risk of data loss. This merging logic exists before this PR. I mentioned a case of data loss above. Let me briefly explain it here. cc @mielientiev
1. There is only one binlog file avaible in mysql, gtids are out of order, 182580, 182579, 182578, corresponding to three update data. PreviousGtidsEvent is A:1-182571. When starting with the `'scan.startup.mode' = 'earliest-offset'`, after reading 182580 and 182579 data, a checkpoint is made. The saved gtid is A:182580-182580, "row": "1", "event": "2". The gtid known to the server from the `show master status` command is A:1-1825771:182578-182580. That is, 182578 has not been read.
2. When recovering from the checkpoint, the gtid of the merged to connect to MySQL is A:1-1825771:182578-182580, and 182578 data will not be read, thus it is lost.
PreviousGtidsEvent A:1-182571 could not be used to avoid the data loss.
Beside that I think the merging is acceptable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37065]: MySQL cdc can lose/skip data during recovering from the checkpoint [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #3845:
URL: https://github.com/apache/flink-cdc/pull/3845
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org