You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/08/13 03:13:45 UTC
(flink-cdc) branch master updated: [FLINK-37835] Fix NoPointException when start with latest-offset. (#4091)
This is an automated email from the ASF dual-hosted git repository.
kunni 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 b937fdcf6 [FLINK-37835] Fix NoPointException when start with latest-offset. (#4091)
b937fdcf6 is described below
commit b937fdcf6ad7ebdefa2a2a3ee23b55a1ddbfa5da
Author: proletarians <75...@users.noreply.github.com>
AuthorDate: Wed Aug 13 11:13:42 2025 +0800
[FLINK-37835] Fix NoPointException when start with latest-offset. (#4091)
---------
Co-authored-by: wuzexian <sh...@alibaba-inc.com>
---
.../mysql/source/offset/BinlogOffset.java | 4 +-
.../debezium/reader/BinlogSplitReaderTest.java | 55 +++++++++++++++++++++-
2 files changed, 56 insertions(+), 3 deletions(-)
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java
index d9334c49f..c6511a1e7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java
@@ -238,7 +238,9 @@ public class BinlogOffset implements Comparable<BinlogOffset>, Serializable {
}
// First compare the MySQL binlog filenames
- if (this.getFilename().compareToIgnoreCase(that.getFilename()) != 0) {
+ if (this.getFilename() != null
+ && that.getFilename() != null
+ && this.getFilename().compareToIgnoreCase(that.getFilename()) != 0) {
return this.getFilename().compareToIgnoreCase(that.getFilename());
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java
index d254d94ad..e9205d2c5 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java
@@ -118,6 +118,9 @@ class BinlogSplitReaderTest extends MySqlSourceTestBase {
LOG.info("Starting MySql8 containers...");
Startables.deepStart(Stream.of(MYSQL8_CONTAINER)).join();
LOG.info("Container MySql8 is started.");
+ LOG.info("Starting MySqlNoGtid containers...");
+ Startables.deepStart(Stream.of(MYSQL_CONTAINER_NOGTID)).join();
+ LOG.info("Container MySqlNoGtid is started.");
}
@AfterAll
@@ -125,6 +128,9 @@ class BinlogSplitReaderTest extends MySqlSourceTestBase {
LOG.info("Stopping MySql8 containers...");
MYSQL8_CONTAINER.stop();
LOG.info("Container MySql8 is stopped.");
+ LOG.info("Stopping MySqlNoGtid containers...");
+ MYSQL_CONTAINER_NOGTID.stop();
+ LOG.info("Container MySqlNoGtid is stopped.");
}
@AfterEach
@@ -472,6 +478,53 @@ class BinlogSplitReaderTest extends MySqlSourceTestBase {
reader.close();
}
+ @Test
+ void testReadBinlogWithoutGtidFromLatestOffset() throws Exception {
+ customerDatabaseNoGtid.createAndInitialize();
+ MySqlSourceConfig sourceConfig =
+ getConfig(
+ MYSQL_CONTAINER_NOGTID,
+ customerDatabaseNoGtid,
+ StartupOptions.latest(),
+ new String[] {"customers"});
+ binaryLogClient = DebeziumUtils.createBinaryClient(sourceConfig.getDbzConfiguration());
+ mySqlConnection = DebeziumUtils.createMySqlConnection(sourceConfig);
+
+ // Create reader and submit splits
+ MySqlBinlogSplit split = createBinlogSplit(sourceConfig);
+ BinlogSplitReader reader = createBinlogReader(sourceConfig);
+ reader.submitSplit(split);
+
+ // Create some binlog events
+ makeCustomersBinlogEvents(
+ mySqlConnection, customerDatabaseNoGtid.qualifiedTableName("customers"), false);
+
+ final DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.BIGINT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("address", DataTypes.STRING()),
+ DataTypes.FIELD("phone_number", DataTypes.STRING()));
+ String[] expected =
+ new String[] {
+ "-U[103, user_3, Shanghai, 123567891234]",
+ "+U[103, user_3, Hangzhou, 123567891234]",
+ "-D[102, user_2, Shanghai, 123567891234]",
+ "+I[102, user_2, Shanghai, 123567891234]",
+ "-U[103, user_3, Hangzhou, 123567891234]",
+ "+U[103, user_3, Shanghai, 123567891234]",
+ "-U[1010, user_11, Shanghai, 123567891234]",
+ "+U[1010, Hangzhou, Shanghai, 123567891234]",
+ "+I[2001, user_22, Shanghai, 123567891234]",
+ "+I[2002, user_23, Shanghai, 123567891234]",
+ "+I[2003, user_24, Shanghai, 123567891234]"
+ };
+ List<String> actual = readBinlogSplits(dataType, reader, expected.length);
+ assertEqualsInOrder(Arrays.asList(expected), actual);
+
+ reader.close();
+ }
+
@Test
void testReadBinlogFromEarliestOffset() throws Exception {
customerDatabase.createAndInitialize();
@@ -1007,7 +1060,6 @@ class BinlogSplitReaderTest extends MySqlSourceTestBase {
@Test
public void testBinlogOffsetCompareWithSnapshotAndBinlogPhase() throws Exception {
- Startables.deepStart(Stream.of(MYSQL_CONTAINER_NOGTID)).join();
// Preparations
customerDatabaseNoGtid.createAndInitialize();
MySqlSourceConfig sourceConfig =
@@ -1064,7 +1116,6 @@ class BinlogSplitReaderTest extends MySqlSourceTestBase {
List<SourceRecord> sourceRecords =
pollRecordsFromReader(binlogReader, RecordUtils::isDataChangeRecord);
- MYSQL_CONTAINER_NOGTID.stop();
Assertions.assertThat(sourceRecords).isEmpty();
}