mariadb debezium source connector 설정

debezium으로 적용

debezium은 mariadb(mysql) 의 binary log를 기반으로 cdc 를 수행한다.

• binary log 켜기 위한 수정 파일 위치 찾기 : mysql --help | grep maria
• maria bin log 켜주기 (/usr/local/etc/my.cnf)

log-bin = /usr/local/Cellar/mariadb/11.3.2/log/binary/maria-bin 
binlog_cache_size = 1M 
max_binlog_size = 512M
expire_logs_days = 7

connector 생성

{
    "name": "maria-cdc-test",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "localhost",
        "database.port": "3306",
        "database.user": "root",
        "database.password": "1234",
        "database.server.id": "1",
        "database.server.name": "mysql01",
        "database.include.list": "my_db",
        "table.include.list": "my_db.my_table",
        "database.ssl.mode":"disabled",
        "schema.history.internal.kafka.bootstrap.servers": "localhost:9092", 
        "schema.history.internal.kafka.topic": "schema.history.fullfillment", 
        "include.schema.changes": "true" ,
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "snapshot.mode": "schema_only_recovery",
        "topic.prefix" : "cdcing.source",
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.unwrap.delete.handling.mode": "rewrite",
        "transforms.unwrap.add.fields" : "op,table",
        "transforms.unwrap.add.fields.prefix": "cdc_meta_",
        "database.connectionTimeZone": "Asia/Seoul"
    }
}