This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 93a40138d [hotfix][connect/mysql] Fix potential comparison failure
when PK is VARBINARY (#4532)
93a40138d is described below
commit 93a40138d7a55fa4f30107bf183f9950fdc36e38
Author: yuxiqian <[email protected]>
AuthorDate: Thu Sep 17 14:03:01 2026 +0800
[hotfix][connect/mysql] Fix potential comparison failure when PK is
VARBINARY (#4532)
---
.../mysql/source/utils/SplitKeyUtils.java | 3 +
.../mysql/source/utils/SplitKeyUtilsTest.java | 67 ++++++++++++++++++++++
2 files changed, 70 insertions(+)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtils.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtils.java
index d19de5df5..3ed9089e5 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtils.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtils.java
@@ -25,6 +25,7 @@ import org.apache.kafka.connect.data.Struct;
import java.math.BigDecimal;
import java.math.BigInteger;
+import java.util.Arrays;
import java.util.List;
/** Utility class to deal split keys and split key ranges. */
@@ -42,6 +43,8 @@ public class SplitKeyUtils {
return ((Comparable) o1).compareTo(o2);
} else if (isNumericObject(o1) && isNumericObject(o2)) {
return toBigDecimal(o1).compareTo(toBigDecimal(o2));
+ } else if (o1 instanceof byte[] && o2 instanceof byte[]) {
+ return Arrays.compareUnsigned((byte[]) o1, (byte[]) o2);
} else {
return o1.toString().compareTo(o2.toString());
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtilsTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtilsTest.java
index 2a2b0421e..98f9bcc05 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtilsTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/SplitKeyUtilsTest.java
@@ -57,6 +57,23 @@ class SplitKeyUtilsTest {
Assertions.assertThat(splits.get(3).getSplitStart()).isEqualTo(new
Object[] {300L});
}
+ @Test
+ void testSortFinishedSplitInfosWithByteArraySplitKey() {
+ TableId tableId = new TableId("test_db", null, "test_table");
+
+ List<FinishedSnapshotSplitInfo> splits = new ArrayList<>();
+ splits.add(createSplit(tableId, "split-80", new Object[]
{bytes(0x80)}, null));
+ splits.add(createSplit(tableId, "split-7f", new Object[]
{bytes(0x7f)}, null));
+ splits.add(createSplit(tableId, "split-ff", new Object[]
{bytes(0xff)}, null));
+ splits.add(createSplit(tableId, "split-00", null, null));
+
+ SplitKeyUtils.sortFinishedSplitInfos(splits);
+
+ Assertions.assertThat(splits)
+ .extracting(FinishedSnapshotSplitInfo::getSplitId)
+ .containsExactly("split-00", "split-7f", "split-80",
"split-ff");
+ }
+
@Test
void testFindSplitByKeyBinary() {
TableId tableId = new TableId("test_db", null, "test_table");
@@ -101,6 +118,48 @@ class SplitKeyUtilsTest {
Assertions.assertThat(result.getSplitId()).isEqualTo("split-0");
}
+ @Test
+ void testFindSplitByKeyBinaryWithByteArraySplitKey() {
+ TableId tableId = new TableId("test_db", null, "test_table");
+
+ List<FinishedSnapshotSplitInfo> sortedSplits = new ArrayList<>();
+ sortedSplits.add(createSplit(tableId, "split-before-7f", null, new
Object[] {bytes(0x7f)}));
+ sortedSplits.add(
+ createSplit(
+ tableId,
+ "split-7f",
+ new Object[] {bytes(0x7f)},
+ new Object[] {bytes(0x80)}));
+ sortedSplits.add(
+ createSplit(
+ tableId,
+ "split-80",
+ new Object[] {bytes(0x80)},
+ new Object[] {bytes(0xff)}));
+ sortedSplits.add(createSplit(tableId, "split-ff", new Object[]
{bytes(0xff)}, null));
+
+ FinishedSnapshotSplitInfo result =
+ SplitKeyUtils.findSplitByKeyBinary(sortedSplits, new Object[]
{bytes(0x00)});
+ Assertions.assertThat(result).isNotNull();
+
Assertions.assertThat(result.getSplitId()).isEqualTo("split-before-7f");
+
+ result = SplitKeyUtils.findSplitByKeyBinary(sortedSplits, new Object[]
{bytes(0x7f)});
+ Assertions.assertThat(result).isNotNull();
+ Assertions.assertThat(result.getSplitId()).isEqualTo("split-7f");
+
+ result = SplitKeyUtils.findSplitByKeyBinary(sortedSplits, new Object[]
{bytes(0x80)});
+ Assertions.assertThat(result).isNotNull();
+ Assertions.assertThat(result.getSplitId()).isEqualTo("split-80");
+
+ result = SplitKeyUtils.findSplitByKeyBinary(sortedSplits, new Object[]
{bytes(0xfe)});
+ Assertions.assertThat(result).isNotNull();
+ Assertions.assertThat(result.getSplitId()).isEqualTo("split-80");
+
+ result = SplitKeyUtils.findSplitByKeyBinary(sortedSplits, new Object[]
{bytes(0xff)});
+ Assertions.assertThat(result).isNotNull();
+ Assertions.assertThat(result.getSplitId()).isEqualTo("split-ff");
+ }
+
@Test
void testFindSplitByKeyBinaryWithOnlyOneSplit() {
TableId tableId = new TableId("test_db", null, "test_table");
@@ -203,4 +262,12 @@ class SplitKeyUtilsTest {
return new FinishedSnapshotSplitInfo(
tableId, splitId, splitStart, splitEnd,
BinlogOffset.ofEarliest());
}
+
+ private static byte[] bytes(int... values) {
+ byte[] bytes = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ bytes[i] = (byte) values[i];
+ }
+ return bytes;
+ }
}
