This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 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 a87b2db89 [FLINK-39315][cdc-connector][mysql] Unregister listeners of
BinaryLogClient to prevent snapshot reader hang during backfill (#4334)
a87b2db89 is described below
commit a87b2db897293f7ff83f5a902678a816476197f9
Author: chengcongchina <[email protected]>
AuthorDate: Thu Sep 24 14:26:23 2026 +0800
[FLINK-39315][cdc-connector][mysql] Unregister listeners of BinaryLogClient
to prevent snapshot reader hang during backfill (#4334)
Co-authored-by: Hang Ruan <[email protected]>
---
.../mysql/MySqlStreamingChangeEventSource.java | 35 ++++++++++++++++------
.../debezium/reader/SnapshotSplitReaderTest.java | 15 ++++++++++
2 files changed, 41 insertions(+), 9 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
index 205582ca8..33329bf0e 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
@@ -87,18 +87,20 @@ import static io.debezium.util.Strings.isNullOrEmpty;
* Copied from Debezium project(1.9.8.Final) to fix
* https://github.com/ververica/flink-cdc-connectors/issues/1944.
*
- * <p>Line 1432-1443 : Adjust GTID merging logic to support recovering from
job which previously
+ * <p>Line 1467-1479 : Adjust GTID merging logic to support recovering from
job which previously
* specifying starting offset on start. Uses {@link
GtidUtils#fixOldChannelsGtidSet} for shared
* EARLIEST/LATEST logic.
*
- * <p>Line 1444-1452 : Fix LATEST mode GTID merging to avoid replaying
pre-checkpoint transactions
+ * <p>Line 1481-1487 : Fix LATEST mode GTID merging to avoid replaying
pre-checkpoint transactions
* when checkpoint GTID has non-contiguous ranges. Delegates to {@link
* GtidUtils#computeLatestModeGtidSet}. See FLINK-39149.
*
- * <p>Line 1490 : Add more error details for some exceptions.
+ * <p>Line 1539 : Add more error details for some exceptions.
*
- * <p>Line 951-963 : Use iterator instead of index-based loop to avoid O(n²)
complexity when
+ * <p>Line 972-984 : Use iterator instead of index-based loop to avoid O(n²)
complexity when
* processing LinkedList rows in handleChange method. See FLINK-38846.
+ *
+ * <p>Line 1289-1300 : Unregister listeners to avoid client reuse
interference. See FLINK-39315.
*/
public class MySqlStreamingChangeEventSource
implements StreamingChangeEventSource<MySqlPartition,
MySqlOffsetContext> {
@@ -1135,12 +1137,19 @@ public class MySqlStreamingChangeEventSource
context);
listener = (event) -> buffer.add(partition,
effectiveOffsetContext, event);
}
- client.registerEventListener(listener);
- client.registerLifecycleListener(new
ReaderThreadLifecycleListener(effectiveOffsetContext));
- client.registerEventListener((event) ->
onEvent(effectiveOffsetContext, event));
- if (LOGGER.isDebugEnabled()) {
- client.registerEventListener((event) ->
logEvent(effectiveOffsetContext, event));
+ ReaderThreadLifecycleListener lifecycleListener =
+ new ReaderThreadLifecycleListener(effectiveOffsetContext);
+ BinaryLogClient.EventListener onEventListener =
+ (event) -> onEvent(effectiveOffsetContext, event);
+ BinaryLogClient.EventListener logEventListener =
+ LOGGER.isDebugEnabled() ? (event) ->
logEvent(effectiveOffsetContext, event) : null;
+
+ client.registerEventListener(listener);
+ client.registerLifecycleListener(lifecycleListener);
+ client.registerEventListener(onEventListener);
+ if (logEventListener != null) {
+ client.registerEventListener(logEventListener);
}
final boolean isGtidModeEnabled = connection.isGtidModeEnabled();
@@ -1277,10 +1286,18 @@ public class MySqlStreamingChangeEventSource
Thread.sleep(100);
}
} finally {
+ // Unregister listeners to avoid client reuse interference
(FLINK-39315)
+ client.unregisterEventListener(listener);
+ client.unregisterEventListener(onEventListener);
+ if (logEventListener != null) {
+ client.unregisterEventListener(logEventListener);
+ }
try {
client.disconnect();
} catch (Exception e) {
LOGGER.info("Exception while stopping binary log client", e);
+ } finally {
+ client.unregisterLifecycleListener(lifecycleListener);
}
}
}
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/SnapshotSplitReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReaderTest.java
index 6598ed126..b41f6a077 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReaderTest.java
@@ -586,6 +586,21 @@ class SnapshotSplitReaderTest extends MySqlSourceTestBase {
"UPDATE " + tableId + " SET address =
'Beijing' WHERE id = 103");
mySqlConnection.commit();
} else if (split.splitId().equals(tableId + ":1")) {
+ // To verify that FLINK-39315 is fixed, generate
sufficient binlog events,
+ // so that the MySqlBinlogSplitReadTask runs long
enough to exercise the
+ // context-running checks in binlog reading backfill
phase.
+ for (int i = 0; i < 100; i++) {
+ mySqlConnection.execute(
+ "UPDATE "
+ + tableId
+ + " SET address = 'Beijing' WHERE
id = 106");
+ mySqlConnection.commit();
+ mySqlConnection.execute(
+ "UPDATE "
+ + tableId
+ + " SET address = 'Shanghai' WHERE
id = 106");
+ mySqlConnection.commit();
+ }
mySqlConnection.execute(
"UPDATE " + tableId + " SET address =
'Beijing' WHERE id = 106");
mySqlConnection.commit();
