Kiểm tra xem MySQL đã bật chế độ ghi log nhị phân hay chưa bằng lệnh:

SHOW VARIABLES LIKE 'log_bin';

Nếu kết quả là 'ON', binlog đang được bật.

Liệt kê các tệp binlog:

SHOW BINARY LOGS;

Xem nội dung binlog:

tungnt@MacBook-Pro-cua-Nguyen-2 mysql % mysqlbinlog /opt/homebrew/var/mysql/binlog.000217
# The proper term IS pseudo_replica_mode, but we USE this compatibility alias
# TO make the statement usable ON server versions 8.0.24 AND older.
/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;
/*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
DELIMITER /*!*/;
# at 4
#241021 13:19:55 server id 1  end_log_pos 127 CRC32 0x609227a3 	START: binlog v 4, server v 8.4.2 created 241021 13:19:55 at startup
# Warning: this binlog IS either IN USE OR was NOT closed properly.
ROLLBACK/*!*/;
BINLOG '
i/IVZw8BAAAAewAAAH8AAAABAAQAOC40LjIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
AAAAAAAAAAAAAAAAAACL8hVnEwANAAgAAAAABAAEAAAAYwAEGggAAAAAAAACAAAACgoKKioAEjQA
CigAAAGjJ5Jg
'/*!*/;
# at 127
#241021 13:19:55 server id 1  end_log_pos 158 CRC32 0xcf101e2a 	Previous-GTIDs
# [empty]
SET @@SESSION.GTID_NEXT= 'AUTOMATIC' /* added by mysqlbinlog */ /*!*/;
DELIMITER ;
# END OF log file
/*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/;
/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;

https://kafka.apache.org/downloads

Khởi động Zookeeper:

% zookeeper-server-start config/zookeeper.properties

Khởi động Kafka:

% kafka-server-start config/server.properties

https://debezium.io/releases/

wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.9.5.Final/debezium-connector-mysql-1.9.5.Final-plugin.tar.gz
 
tar -xvf debezium-connector-mysql-1.9.5.Final-plugin.tar.gz
 
mkdir kafka_2.12-3.8.0/plugins/debezium-connector-mysql
 
cp -r debezium-connector-mysql kafka_2.12-3.8.0/plugins/debezium-connector-mysql

vim kafka_2.12-3.8.0/config/connect-standalone.properties
 
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
plugin.path=/Users/tungnt/Downloads/Libs/kafka_2.12-3.8.0/plugins

vim kafka_2.12-3.8.0/config/mysql-connector.properties
 
name=mysql-connector
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=localhost
database.port=3306
database.user=xxx
database.password=xxx
database.server.id=184054
database.server.name=dbserver1
database.include.list=testdb
database.history.kafka.bootstrap.servers=localhost:9092
database.history.kafka.topic=schema-changes.testdb

./bin/connect-standalone.sh config/connect-standalone.properties config/mysql-connector.properties