ruanhang1993 commented on code in PR #4334:
URL: https://github.com/apache/flink-cdc/pull/4334#discussion_r4069231538
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java:
##########
@@ -1106,23 +1108,30 @@ public void execute(
(event) -> handleRowsQuery(effectiveOffsetContext, event));
}
- BinaryLogClient.EventListener listener;
+ BinaryLogClient.EventListener eventListener;
if (connectorConfig.bufferSizeForStreamingChangeEventSource() == 0) {
- listener = (event) -> handleEvent(partition,
effectiveOffsetContext, event);
+ eventListener = (event) -> handleEvent(partition,
effectiveOffsetContext, event);
} else {
EventBuffer buffer =
new EventBuffer(
connectorConfig.bufferSizeForStreamingChangeEventSource(),
this,
context);
- listener = (event) -> buffer.add(partition,
effectiveOffsetContext, event);
+ eventListener = (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 metricsEventListener =
Review Comment:
The name metricsEventListener isn't appropriate here. The corresponding
listener doesn't only handle metrics updates — it also registers some other
information. Consider renaming it to `onEventListener`.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java:
##########
@@ -1264,6 +1272,13 @@ public void execute(
} catch (Exception e) {
LOGGER.info("Exception while stopping binary log client", e);
}
+
+ client.unregisterEventListener(eventListener);
+ client.unregisterEventListener(metricsEventListener);
+ client.unregisterLifecycleListener(lifecycleListener);
+ if (logEventListener != null) {
+ client.unregisterEventListener(logEventListener);
Review Comment:
I'd suggest moving `unregister` into the finally block:
1. The exception-masking concern doesn't apply here: unregister* is just
CopyOnWriteArrayList.remove() (mysql-binlog-connector-java 0.27.2, L1236/L1287)
and cannot throw.
2. With the current placement, the "Stopped reading binlog after N events,
last recorded offset" log is deterministically lost on the normal path:
disconnect() triggers onDisconnect asynchronously from the blc-* reader thread
for listeners still registered at that moment (BinaryLogClient.java L632-639,
notifyWhenDisconnected is true once the connection was established), but the
lifecycle listener has already been removed by then. Today (without this PR)
that log is printed; after this PR it would only survive on the exception path.
Unregistering after disconnect() in the finally block keeps the current
behavior in the common case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
