This is an automated email from the ASF dual-hosted git repository. leonardBang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit b8edfcfd357f87774fe055f966deb07672373cfd Author: Leonard Xu <[email protected]> AuthorDate: Fri Jul 3 17:26:25 2026 +0800 [fix][mysql-source] Compare varbinary split keys by byte value MySQL chunk splitting must order VARBINARY split keys by their binary contents instead of Java object identity so incremental snapshot boundaries stay stable and varbinary rows are not missed. Co-Authored-By: Claude Opus 4.7 <[email protected]> --- .../cdc/connectors/mysql/source/utils/ObjectUtils.java | 4 ++++ .../connectors/mysql/source/utils/ObjectUtilsTest.java | 18 ++++++++++++++++++ 2 files changed, 22 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/ObjectUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtils.java index a3a7281a8..7986cfaa5 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtils.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtils.java @@ -19,6 +19,7 @@ package org.apache.flink.cdc.connectors.mysql.source.utils; import java.math.BigDecimal; import java.math.BigInteger; +import java.util.Arrays; /** Utilities for operation on {@link Object}. */ public class ObjectUtils { @@ -82,6 +83,9 @@ public class ObjectUtils { */ @SuppressWarnings("unchecked") public static int compare(Object obj1, Object obj2) { + if (obj1 instanceof byte[] && obj2 instanceof byte[]) { + return Arrays.compareUnsigned((byte[]) obj1, (byte[]) obj2); + } if (obj1 instanceof Comparable && obj1.getClass().equals(obj2.getClass())) { return ((Comparable) obj1).compareTo(obj2); } else { 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/ObjectUtilsTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtilsTest.java index ad73dbc83..6c85df427 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtilsTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/ObjectUtilsTest.java @@ -41,4 +41,22 @@ class ObjectUtilsTest { ObjectUtils.minus(new BigDecimal("100.12345"), new BigDecimal("1.00001"))) .isEqualTo(new BigDecimal("99.12344")); } + + @Test + void testCompareByteArraysByValue() { + Assertions.assertThat(ObjectUtils.compare(new byte[] {4, 4, 4, 5}, new byte[] {4, 4, 4, 6})) + .isLessThan(0); + Assertions.assertThat(ObjectUtils.compare(new byte[] {4, 4, 4, 6}, new byte[] {4, 4, 4, 5})) + .isGreaterThan(0); + Assertions.assertThat(ObjectUtils.compare(new byte[] {1, 2, 3}, new byte[] {1, 2, 3})) + .isZero(); + } + + @Test + void testCompareByteArraysAsUnsignedValues() { + Assertions.assertThat(ObjectUtils.compare(new byte[] {0x7f}, new byte[] {(byte) 0x80})) + .isLessThan(0); + Assertions.assertThat(ObjectUtils.compare(new byte[] {(byte) 0x80}, new byte[] {0x7f})) + .isGreaterThan(0); + } }
