Debezium 개요

programmer-study/api-practice/debezium-embedded at main · lkdcode/programmer-study

🎯 [Growth]. Contribute to lkdcode/programmer-study development by creating an account on GitHub.

https://github.com/lkdcode/programmer-study/tree/main/api-practice/debezium-embedded

🔥 Debezium?

🚀 필터, 마스킹, 변환(SMT)?

🚀 Embedded mode?

🔥 시나리오

🚀 Polling

🚀 CDC

🚀 Polling vs CDC

🔥 Debezium Embedded

🚀 의존성

# 임베디드 모드 (앱 내 Kafka Connect 없이 커넥터 구동)
implementation("io.debezium:debezium-embedded:3.1.0.Final")

# 임베디드 실행을 위한 API 인터페이스 (DebeziumEngine, ChangeEvent)
implementation("io.debezium:debezium-api:3.1.0.Final")

# MySQL 용 소스 커넥터(binlog 를 읽어 이벤트 전달)
implementation("io.debezium:debezium-connector-mysql:3.1.0.Final")

# 오프셋, 스키마 히스토리를 DB에서 관리하기 위한 스토리지 모듈
implementation("io.debezium:debezium-storage-jdbc:3.1.0.Final")

🚀 Config

• Debezium connector for MySQL :: Debezium Documentation

  Debezium connector for MySQL :: Debezium Documentation

  MySQL has a binary log (binlog) that records all operations in the order in which they are committed to the database. This includes changes to table schemas as well as changes to the data in tables. MySQL uses the binlog for replication and recovery.

  

• Storing state of a Debezium connector :: Debezium Documentation

  Storing state of a Debezium connector :: Debezium Documentation

  Debezium connectors require persistent storage to preserve their state between restarts. All connectors require a mechanism to provide persistent storage for the offsets. In addition, connectors such as Db2, MySQL, Oracle, and SQL Server, require additional storage for their so-called internal schema history, which records changes to table schema in the database.

  

@Configuration
class DebeziumConfig {

    @Bean("debeziumConnector")
    fun debeziumConnector(): Properties = Properties()
        .apply {
            put("name", "lkdcode-mysql")
            put("connector.class", "io.debezium.connector.mysql.MySqlConnector")
            put("topic.prefix", "lkdcode")

            put("snapshot.mode", "initial")
            put("snapshot.locking.mode", "none")

            put("database.hostname", "localhost")
            put("database.port", "13306")
            put("database.user", "root")
            put("database.password", "lkdcode")
            put("database.server.id", "1234")
            put("database.connectionTimeZone", "Asia/Seoul")

            put("database.include.list", "lkdcode")
            put("table.include.list", "lkdcode.banana")

            put("offset.storage", "io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore")
            put("offset.storage.jdbc.connection.url", "jdbc:mysql://localhost:13306/lkdcode")
            put("offset.storage.jdbc.connection.user", "root")
            put("offset.storage.jdbc.connection.password", "lkdcode")
            put("offset.storage.jdbc.table.name", "debezium_offsets")

            put("schema.history.internal", "io.debezium.storage.jdbc.history.JdbcSchemaHistory")
            put("schema.history.internal.jdbc.connection.url", "jdbc:mysql://localhost:13306/lkdcode")
            put("schema.history.internal.jdbc.connection.user", "root")
            put("schema.history.internal.jdbc.connection.password", "lkdcode")
            put("schema.history.internal.jdbc.table.name", "debezium_schema_history")
        }
}

🎯 name

🎯 connector.class

🎯 topic.prefix

🎯 offset.storage

🎯 schema.history

🎯 snapshot.mode

• Debezium connector for MySQL :: Debezium Documentation

  Debezium connector for MySQL :: Debezium Documentation

  MySQL has a binary log (binlog) that records all operations in the order in which they are committed to the database. This includes changes to table schemas as well as changes to the data in tables. MySQL uses the binlog for replication and recovery.

  

• initial : 기본 설정. 오프셋이 없을 때 한 번만 전체 스냅샷 후 스티리밍 전환

• no_data : 스키마만 캡처. 데이터 스냅샷 생략 후 바로 스트리밍(과거 풀데이터 불필요할 때)

• when_needed : 시작 시 오프셋이 없거나 로그 위치가 유실된 경우에만 필요 시 스냅샷

• initial_only : 스냅샷만 수행하고 종료(일회성 덤프 용도)

• recovery : 스키마 히스토리 토픽/파일 복구용 스냅샷

🎯 offset storage DDL

• Storing state of a Debezium connector :: Debezium Documentation

  Storing state of a Debezium connector :: Debezium Documentation

  Debezium connectors require persistent storage to preserve their state between restarts. All connectors require a mechanism to provide persistent storage for the offsets. In addition, connectors such as Db2, MySQL, Oracle, and SQL Server, require additional storage for their so-called internal schema history, which records changes to table schema in the database.

  

🎯 schema history DDL

• Storing state of a Debezium connector :: Debezium Documentation

  Storing state of a Debezium connector :: Debezium Documentation

  Debezium connectors require persistent storage to preserve their state between restarts. All connectors require a mechanism to provide persistent storage for the offsets. In addition, connectors such as Db2, MySQL, Oracle, and SQL Server, require additional storage for their so-called internal schema history, which records changes to table schema in the database.

  

🚀 Listener

• Debezium Engine :: Debezium Documentation

  Debezium Engine :: Debezium Documentation

  Debezium connectors are normally operated by deploying them to a Kafka Connect service, and configuring one or more connectors to monitor upstream databases and produce data change events for all changes that they see in the upstream databases. Those data change events are written to Kafka, where they can be independently consumed by many different applications. Kafka Connect provides excellent fault tolerance and scalability, since it runs as a distributed service and ensures that all registered and configured connectors are always running. For example, even if one of the Kafka Connect endpoints in a cluster goes down, the remaining Kafka Connect endpoints will restart any connectors that were previously running on the now-terminated endpoint, minimizing downtime and eliminating administrative activities.

  

@Service
class DebeziumListener(
    @Qualifier("debeziumConnector")
    private val properties: Properties
) {
    private val executor = Executors.newSingleThreadExecutor { r ->
        Thread(r, "debezium-lkdcode").apply { isDaemon = false }
    }

    private lateinit var engine: DebeziumEngine<ChangeEvent<String, String>>

    @PostConstruct
    fun start() {
        engine = DebeziumEngine.create(Json::class.java)
            .using(properties)
            .notifying { records: List<ChangeEvent<String, String>>,
                         committer: DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> ->
                records.forEach { record ->
                    try {
                        println("CDC payload: ${record.value()}")

                        committer.markProcessed(record)
                    } catch (e: Exception) {
                        println("ERROR: $e")
                    }
                }

                committer.markBatchFinished()
            }
            .build()

        executor.submit(engine)
    }

    @PreDestroy
    fun stop() {
        try {
            if (this::engine.isInitialized) engine.close()
        } finally {
            executor.shutdown()
        }
    }
}

콘솔에 출력되는 값을 JSONPath Online Evaluator

JSONPath Online Evaluator

에서 JSONPath 로 검색해볼 수 있다.

[
  {
    "id": 34,
    "name": "바나나",
    "sweetness_brix": "5.0",
    "brand": "lkdcode",
    "grade": "특급",
    "harvested_at": 20340
  }
]

🔥 체크 사항

🚀 무한 락?

🚀 Listener 가 이벤트를 수신 못한다면 아래의 설정들을 살펴보자.

• log_bin : 바이너리 로그 사용

• server_id : 서버 고유 ID

• binlog_format : 행 단위 로깅
• STATEMENT : 실행된 SQL 문 그 자체
• ROW : 변경된 행 데이터
• MIXED : 상황에 따라 혼합

• binlog_row_image : 행 이미지의 상세도
• FULL : 모든 컬럼을 기록
• MINIMAL : 변경된 컬럼 및 키 등 필수값만
• NOBLOB : BLOB/TEXT 등은 변경 시에만 포함

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'server_id';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';
# 결과
log_bin = ON
server_id ≠ 0
binlog_format = ROW
binlog_row_image = FULL

server-id = 1
log-bin = /var/lib/mysql/mysql-bin
binlog-format = ROW
binlog-row-image = FULL
expire_logs_days = 7

• 직접 config 파일에서 설정하기

cat mariadb-server.cnf
#
# These groups are read by MariaDB server.
# Use it for options that only the server (but not clients) should see
#
# See the examples of server my.cnf files in /usr/share/mysql/
#

# this is read by the standalone daemon and embedded servers
[server]

# this is only for the mysqld standalone daemon
# Settings user and group are ignored when systemd is used.
# If you need to run mysqld under a different user or group,
# customize your systemd unit file for mysqld/mariadb according to the
# instructions in http://fedoraproject.org/wiki/Systemd
[mysqld]
datadir=/var/lib/mysql
socket=/var/lib/mysql/mysql.sock
log-error=/var/log/mariadb/mariadb.log
pid-file=/run/mariadb/mariadb.pid
lower_case_table_names = 1

# 직접 config 파일에서 설정할 수 있다. 
# ✅ Debezium/CDC 필수 설정들
server_id=1
log_bin=/var/lib/mysql/mysql-bin
binlog_format=ROW
binlog_row_image=FULL
expire_logs_days=7

#
# * Galera-related settings
#
[galera]
# Mandatory settings
#wsrep_on=ON
#wsrep_provider=
#wsrep_cluster_address=
#binlog_format=row
#default_storage_engine=InnoDB
#innodb_autoinc_lock_mode=2
#
# Allow server to accept connections on all interfaces.
#
#bind-address=0.0.0.0
#
# Optional setting
#wsrep_slave_threads=1
#innodb_flush_log_at_trx_commit=0

# this is only for embedded server
[embedded]

# This group is only read by MariaDB servers, not by MySQL.
# If you use the same .cnf file for MySQL and MariaDB,
# you can put MariaDB-only options here
[mariadb]

# This group is only read by MariaDB-10.5 servers.
# If you use the same .cnf file for MariaDB of different versions,
# use this group for options that older servers don't understand
[mariadb-10.5]

• REPLICATION SLAVE : 바이너리 로그를 읽는 레플리카 권한 (MySQL 8에선 REPLICATION REPLICA)

• REPLICATION CLIENT : SHOW MASTER/SLAVE STATUS, SHOW BINARY LOGS 등 상태 조회 권한.

• SELECT ON : 스냅샷 시 테이블을 읽을 수 있는 권한

CREATE USER 'debezium'@'%' IDENTIFIED BY 'debezium_password';
GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT ON *.* TO 'debezium'@'%';

FLUSH PRIVILEGES;