@Override
        public void onDisconnect(BinaryLogClient client) {
            if (LOGGER.isInfoEnabled()) {
                taskContext.temporaryLoggingContext(connectorConfig, "binlog", () -> {
                    Map<String, ?> offset = lastOffset;
                    if (offset != null) {
                        LOGGER.info("Stopped reading binlog after {} events, last recorded offset: {}",
                                totalRecordCounter, offset);
                    } else {
                        LOGGER.info("Stopped reading binlog after {} events, no new offset was recorded",
                                totalRecordCounter);
                    }
                });
            }
        }
Wraps the specified exception in a {@link DebeziumException}, ensuring that all useful state is captured inside
the new exception's message.

@param error the exception; may not be null
@return the wrapped Kafka Connect exception
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java