Skip to content

[FLINK-38183] Fix the problem of data loss in the GTID disordered scenario #4072

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
+105 −36
Draft

[FLINK-38183] Fix the problem of data loss in the GTID disordered scenario #4072

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
37fcc78
Fix the problem of data loss in the GTID disordered scenario
Aug 2, 2025
835a2fd
use ParameterizedTest
Aug 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 61 additions & 14 deletions ...ectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/GtidUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,20 @@

package io.debezium.connector.mysql;

import java.util.ArrayList;
import com.google.common.collect.Range;
import com.google.common.collect.RangeSet;
import com.google.common.collect.TreeRangeSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/** Utils for handling GTIDs. */
public class GtidUtils {
private static final Logger LOG = LoggerFactory.getLogger(GtidUtils.class);

/**
* This method corrects the GTID set that has been restored from a state or checkpoint using the
Expand All @@ -39,21 +46,61 @@ public static GtidSet fixRestoredGtidSet(GtidSet serverGtidSet, GtidSet restored
for (GtidSet.UUIDSet uuidSet : restoredGtidSet.getUUIDSets()) {
GtidSet.UUIDSet serverUuidSet = newSet.get(uuidSet.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(
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
serverInterval.getStart(), restoredIntervalEnd));
long serverIntervalEnd = getIntervalEnd(serverUuidSet);

// The end of the restored interval cannot be greater than the end of the server
// span
RangeSet<Long> restoredRangeSet = TreeRangeSet.create();
for (GtidSet.Interval restoredInterval : uuidSet.getIntervals()) {
if (restoredInterval.getStart() > serverIntervalEnd) {
LOG.warn(
"Discard restored interval {} because it is greater than the end of the server span {}",
restoredInterval,
serverIntervalEnd);
} else if (restoredInterval.getEnd() > serverIntervalEnd) {
Range<Long> adjustedInterval =
Range.closed(restoredInterval.getStart(), serverIntervalEnd);
restoredRangeSet.add(adjustedInterval);
LOG.warn(
"Adjust restored interval {} to {} because it is greater than the end of the server span {}",
restoredInterval,
adjustedInterval,
serverIntervalEnd);
} else {
restoredRangeSet.add(
Range.closed(
restoredInterval.getStart(), restoredInterval.getEnd()));
}
}

// Add the server intervals that do not overlap with the middle gap of the restored
// intervals
final Range<Long> restoredSmallestRange =
restoredRangeSet.asRanges().iterator().next();
serverUuidSet.getIntervals().stream()
.filter(
interval ->
interval.getStart()
<= restoredSmallestRange.upperEndpoint())
.forEach(
interval ->
restoredRangeSet.add(
Range.closed(
interval.getStart(),
Math.min(
interval.getEnd(),
restoredSmallestRange
.upperEndpoint()))));

List<com.github.shyiko.mysql.binlog.GtidSet.Interval> newIntervals =
restoredRangeSet.asRanges().stream()
.map(
range ->
new com.github.shyiko.mysql.binlog.GtidSet.Interval(
range.lowerEndpoint(),
range.upperEndpoint()))
.collect(Collectors.toList());

newSet.put(
uuidSet.getUUID(),
new GtidSet.UUIDSet(
Expand Down
66 changes: 44 additions & 22 deletions ...rs/flink-connector-mysql-cdc/src/test/java/io/debezium/connector/mysql/GtidUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,38 +17,60 @@

package io.debezium.connector.mysql;

import org.junit.jupiter.api.Test;

import static io.debezium.connector.mysql.GtidUtils.fixRestoredGtidSet;
import static io.debezium.connector.mysql.GtidUtils.mergeGtidSetInto;

import static org.assertj.core.api.Assertions.assertThat;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

import java.util.stream.Stream;

/** Unit test for {@link GtidUtils}. */
class GtidUtilsTest {
@Test
void testFixingRestoredGtidSet() {
GtidSet serverGtidSet = new GtidSet("A:1-100");
GtidSet restoredGtidSet = new GtidSet("A:30-100");
assertThat(fixRestoredGtidSet(serverGtidSet, restoredGtidSet)).hasToString("A:1-100");

serverGtidSet = new GtidSet("A:1-100");
restoredGtidSet = new GtidSet("A:30-50");
assertThat(fixRestoredGtidSet(serverGtidSet, restoredGtidSet)).hasToString("A:1-50");
@ParameterizedTest(name = "{0}")
@MethodSource("gtidSetsProvider")
void testFixingRestoredGtidSet(
String description, String serverStr, String restoredStr, String expectedStr) {
GtidSet serverGtidSet = new GtidSet(serverStr);
GtidSet restoredGtidSet = new GtidSet(restoredStr);

serverGtidSet = new GtidSet("A:1-100:102-200,B:20-200");
restoredGtidSet = new GtidSet("A:106-150");
assertThat(fixRestoredGtidSet(serverGtidSet, restoredGtidSet))
.hasToString("A:1-100:102-150,B:20-200");
GtidSet result = fixRestoredGtidSet(serverGtidSet, restoredGtidSet);

serverGtidSet = new GtidSet("A:1-100:102-200,B:20-200");
restoredGtidSet = new GtidSet("A:106-150,C:1-100");
assertThat(fixRestoredGtidSet(serverGtidSet, restoredGtidSet))
.hasToString("A:1-100:102-150,B:20-200,C:1-100");
assertThat(result).hasToString(expectedStr);
}

serverGtidSet = new GtidSet("A:1-100:102-200,B:20-200");
restoredGtidSet = new GtidSet("A:106-150:152-200,C:1-100");
assertThat(fixRestoredGtidSet(serverGtidSet, restoredGtidSet))
.hasToString("A:1-100:102-200,B:20-200,C:1-100");
private static Stream<Arguments> gtidSetsProvider() {
return Stream.of(
Arguments.of(
"Basic example with a straightforward subset",
"A:1-100",
"A:1-50:63-100",
"A:1-50:63-100"),
Arguments.of(
"Restored starts midrange, single gap", "A:1-100", "A:45-80", "A:1-80"),
Arguments.of(
"Multiple intervals with gaps in restored",
"A:1-100,B:1-100",
"A:45-80:83-90:92-98,C:1-20",
"A:1-80:83-90:92-98,B:1-100,C:1-20"),
Arguments.of(
"Server has disjoint intervals, restored partially overlaps",
"A:1-50:60-90:95-200",
"A:45-50:65-70:96-100",
"A:1-50:65-70:96-100"),
Arguments.of(
"Restored completely covers server range", "A:1-100", "A:1-100", "A:1-100"),
Arguments.of(
"Restored partially covers server range",
"A:1-100:102-200",
"A:106-150:152-200",
"A:1-100:102-150:152-200"),
Arguments.of("Restored end exceeds server range", "A:1-100", "A:1-110", "A:1-100"));
}

@Test
Expand Down