This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new cda69534f0 [Fix][Connector-V2][MySQL-CDC] Fix BinlogOffset.compareTo
ignoring restartSkipRows when GTID sets are equal (#10811)
cda69534f0 is described below
commit cda69534f0b35d36edc15af6e54c79a1a515955e
Author: zoo-code <[email protected]>
AuthorDate: Mon Jun 22 22:03:01 2026 +0900
[Fix][Connector-V2][MySQL-CDC] Fix BinlogOffset.compareTo ignoring
restartSkipRows when GTID sets are equal (#10811)
---
.../cdc/mysql/source/offset/BinlogOffset.java | 11 ++-
.../cdc/mysql/source/offset/BinlogOffsetTest.java | 88 ++++++++++++++++++++++
2 files changed, 96 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffset.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffset.java
index 06e71dd4a6..fc55565078 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffset.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffset.java
@@ -141,9 +141,14 @@ public class BinlogOffset extends Offset {
GtidSet gtidSet = new GtidSet(gtidSetStr);
GtidSet targetGtidSet = new GtidSet(targetGtidSetStr);
if (gtidSet.equals(targetGtidSet)) {
- long restartSkipEvents = this.getRestartSkipEvents();
- long targetRestartSkipEvents = that.getRestartSkipEvents();
- return Long.compare(restartSkipEvents,
targetRestartSkipEvents);
+ // Same GTID set means both offsets are within the same
transaction
+ // boundary, so ordering must also consider per-event and
per-row progress.
+ int eventCompare =
+ Long.compare(this.getRestartSkipEvents(),
that.getRestartSkipEvents());
+ if (eventCompare != 0) {
+ return eventCompare;
+ }
+ return Long.compare(this.getRestartSkipRows(),
that.getRestartSkipRows());
}
// The GTIDs are not an exact match, so figure out if this is
a subset of the target
// offset
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffsetTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffsetTest.java
new file mode 100644
index 0000000000..4889308644
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/offset/BinlogOffsetTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BinlogOffsetTest {
+
+ private static final String GTID_SET_A =
"036d85a9-64e5-11e6-9b48-42010af0000c:1-100";
+ private static final String GTID_SET_B =
"036d85a9-64e5-11e6-9b48-42010af0000c:1-200";
+
+ @Test
+ public void testCompareToWithEqualGtidSetConsidersRestartSkipRows() {
+ BinlogOffset lower = new BinlogOffset("mysql-bin.000001", 4L, 1L, 5L,
0L, GTID_SET_A, 1);
+ BinlogOffset higher = new BinlogOffset("mysql-bin.000001", 4L, 1L, 9L,
0L, GTID_SET_A, 1);
+
+ Assertions.assertTrue(
+ lower.compareTo(higher) < 0,
+ "offset with smaller restartSkipRows must be ordered before a
larger one "
+ + "when the GTID set and restartSkipEvents are equal");
+ Assertions.assertTrue(higher.compareTo(lower) > 0);
+ Assertions.assertEquals(0, lower.compareTo(lower));
+ }
+
+ @Test
+ public void testCompareToWithEqualGtidSetPrefersRestartSkipEvents() {
+ BinlogOffset earlierEvent =
+ new BinlogOffset("mysql-bin.000001", 4L, 1L, 9L, 0L,
GTID_SET_A, 1);
+ BinlogOffset laterEvent =
+ new BinlogOffset("mysql-bin.000001", 4L, 2L, 0L, 0L,
GTID_SET_A, 1);
+
+ Assertions.assertTrue(
+ earlierEvent.compareTo(laterEvent) < 0,
+ "restartSkipEvents must take precedence over restartSkipRows");
+ }
+
+ @Test
+ public void testCompareToWithEqualGtidSetAndEqualProgress() {
+ BinlogOffset a = new BinlogOffset("mysql-bin.000001", 4L, 1L, 5L, 0L,
GTID_SET_A, 1);
+ BinlogOffset b = new BinlogOffset("mysql-bin.000001", 4L, 1L, 5L, 0L,
GTID_SET_A, 1);
+
+ Assertions.assertEquals(0, a.compareTo(b));
+ }
+
+ @Test
+ public void testCompareToWithGtidSubsetAndSuperset() {
+ BinlogOffset subset = new BinlogOffset("mysql-bin.000001", 4L, 0L, 0L,
0L, GTID_SET_A, 1);
+ BinlogOffset superset = new BinlogOffset("mysql-bin.000001", 4L, 0L,
0L, 0L, GTID_SET_B, 1);
+
+ Assertions.assertTrue(subset.compareTo(superset) < 0);
+ Assertions.assertTrue(superset.compareTo(subset) > 0);
+ }
+
+ @Test
+ public void testCompareToWithoutGtidFallsBackToEventsAndRows() {
+ BinlogOffset lowerRow = new BinlogOffset("mysql-bin.000001", 4L, 1L,
3L, 0L, null, 1);
+ BinlogOffset higherRow = new BinlogOffset("mysql-bin.000001", 4L, 1L,
7L, 0L, null, 1);
+
+ Assertions.assertTrue(lowerRow.compareTo(higherRow) < 0);
+ Assertions.assertTrue(higherRow.compareTo(lowerRow) > 0);
+ }
+
+ @Test
+ public void testNoStoppingOffsetIsAlwaysMaximum() {
+ BinlogOffset regular = new BinlogOffset("mysql-bin.000001", 4L, 1L,
5L, 0L, GTID_SET_A, 1);
+
+
Assertions.assertTrue(BinlogOffset.NO_STOPPING_OFFSET.compareTo(regular) > 0);
+
Assertions.assertTrue(regular.compareTo(BinlogOffset.NO_STOPPING_OFFSET) < 0);
+ Assertions.assertEquals(
+ 0,
BinlogOffset.NO_STOPPING_OFFSET.compareTo(BinlogOffset.NO_STOPPING_OFFSET));
+ }
+}
