[FLINK-39197][jdbc&mysql] Fix NPE when finding chunk end

[ChengbingLiu]

Fix https://issues.apache.org/jira/browse/FLINK-39197

We encountered the following problem when using MySQL pipeline connector:

Caused by: org.apache.flink.util.FlinkRuntimeException: Chunk splitting has encountered exception
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.checkSplitterErrors(MySqlSnapshotSplitAssigner.java:698)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.getNext(MySqlSnapshotSplitAssigner.java:398)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlHybridSplitAssigner.getNext(MySqlHybridSplitAssigner.java:151)
    at org.apache.flink.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator.assignSplits(MySqlSourceEnumerator.java:223)
    at org.apache.flink.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator.handleSplitRequest(MySqlSourceEnumerator.java:117)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinator.handleRequestSplitEvent(SourceCoordinator.java:568)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$handleEventFromOperator$3(SourceCoordinator.java:295)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$10(SourceCoordinator.java:469)
    ... 8 more
Caused by: java.lang.IllegalStateException: Error when splitting chunks for db1.table1
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.splitTable(MySqlSnapshotSplitAssigner.java:359)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.splitChunksForRemainingTables(MySqlSnapshotSplitAssigner.java:676)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    ... 3 more
Caused by: java.lang.NullPointerException
    at org.apache.flink.cdc.connectors.mysql.source.utils.ObjectUtils.compare(ObjectUtils.java:88)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter.nextChunkEnd(MySqlChunkSplitter.java:360)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter.splitOneUnevenlySizedChunk(MySqlChunkSplitter.java:176)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter.splitChunks(MySqlChunkSplitter.java:138)
    at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.splitTable(MySqlSnapshotSplitAssigner.java:357)
    ... 6 more

This should be because max values are removed resulting in chunkEnd == null, as mentioned in

    // chunk end might be null when max values are removed
    Object chunkEnd =
            queryNextChunkMax(jdbc, tableId, splitColumn, chunkSize, previousChunkEnd);

[Copilot]

Pull request overview

Fixes FLINK-39197 by preventing a NullPointerException during snapshot chunk splitting when the computed next chunk end is null (e.g., rows containing max values are removed between min/max discovery and subsequent chunk boundary queries).

Changes:

• Add an early null return in MySqlChunkSplitter.nextChunkEnd(...) when queryNextChunkMax(...) returns null.
• Add the same early null return in the base JdbcSourceChunkSplitter.nextChunkEnd(...) to protect all JDBC-based connectors.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java	Prevents NPE in MySQL snapshot chunk splitting when the next chunk end cannot be determined (returns null).
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java	Applies the same null-guard to the shared JDBC chunk splitting logic used by multiple connectors.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Consider adding a regression test that covers the scenario fixed here: queryNextChunkMax(...) returning null (e.g., max rows deleted between the earlier MIN/MAX query and subsequent chunk-end queries). This change prevents an NPE, but without a test it’s easy for the null-handling path in nextChunkEnd to regress again. A unit test could stub the JDBC/StatementUtils interaction, or an ITCase could delete the max row after MIN/MAX is determined and assert chunk splitting completes without exceptions.