Debezium MySQL Source Connector 제대로 이해하기

0. 들어가며

RDB의 데이터 변경을 실시간으로 Kafka로 보내야 할 때, CDC 방식을 많이 사용한다. Kafka의 Connector 도구로 가장 많이 쓰이고 잘 알려진 Debezium에 대해 자세히 파헤쳐보자.

1. CDC(Change Data Capture)와 Debezium에 대해

1-1. CDC(Change Data Capture)란?

데이터베이스의 변경(INSERT, UPDATE, DELETE)을 캡쳐해 다른 시스템으로 전파하는 방법이다.
구현 방식은 크게 폴링 기반, 트리거 기반, 로그 기반으로 나뉜다.
CDC는 거의 로그 기반을 택하는데, 그 이유는 DB가 복제/복구를 위해 어차피 남기는 트랜잭션 로그를 읽기 때문에 원본 DB에 추가 부하가 거의 없고, 커밋 순서ㅐ로 빠짐없이 변경을 가져올 수 있기 때문이다.

1-2. Debezium 이란?

Debezium은 로그 기반 CDC를 Kafka Connect 위에서 구현한 오픈소스 플랫폼이다.
Debezium은 변경 이벤트를 Apache Kafka와 같은 스트리밍 시스템으로 전달하여 다른 저장소로 실시간 데이터베이스 변경 사항을 처리할 수 있도록 한다.

<동작 방식>

Debezium 커넥터는 각각의 데이터베이스에서 변경 사항을 캡처한다.
각 Debezium 커넥터는 자신의 소스 데이터베이스에 연결을 설정한다.
- MySQL 커넥터는 binlog에 접근하기 위한 클라이언트 라이브러리 사용
- PostgreSQL 커넥터는 logical replication stream에서 읽어옴
하나의 데이터베이스 테이블에서 발생한 변경 사항은 하나의 kafka 토픽에 기록된다.
필요에 따라, Debezium의 토픽 라우팅 변환을 구성하여 대상 토픽 이름을 변경할 수 있다.
Apache Kafka 토픽에 변경 이벤트 레코드가 저장되면, Kafka Connect를 사용하여 해당 레코드를 여러 데이터베이스로 스트리밍 할 수 있다.
- 선택한 싱크 커넥터에 따라, 모든 변경 정보가 필요하지 않을 때는 변경 후의 정보만을 추출하여 다른 시스템으로 보낼 수 있다.

Sink Connector란? Debezium은 Apache Kafka 토픽과 같은 소스에서 이벤트를 소비할 수 있는 싱크 커넥터를 제공합니다. 싱크 커넥터는 데이터 형식을 표준화한 다음, 구성된 싱크 저장소에 이벤트 데이터를 저장합니다. 그러면 다른 시스템, 애플리케이션 또는 사용자가 데이터 싱크에서 이벤트에 접근할 수 있습니다.

2. MySQL Connector는 어떻게 동작할까?

2-1. binlog : 모든 것의 출발점

MySQL은 데이터베이스에 커밋된 모든 작업을 커밋된 순서대로 바이너리 로그(binlog)에 기록한다. 여기에는 테이블 데이터 변경뿐만 아니라 스키마 변경(DDL)도 포함된다. 원래 binlog는 복제(replication)와 복구(recovery)를 위해 존재한다.

Debezium MySQL 커넥터는 바로 이 binlog를 읽어 row 단위 INSERT, UPDATE, DELETE 작업에 대한 change event를 만들고 이를 Kafka 토픽으로 발행한다. 컨슈머 애플리케이션은 그 토픽을 구독해서 변경을 받아간다. 즉 커넥터는 MySQL 입장에서 보면 "하나의 복제 클라이언트"처럼 행동한다.

2-2. Snapshot : 시작 시점의 baseline 만들기

하지만 binlog는 보통 일정 기간이 지나면 삭제되도록 설정된다. 그러면 커넥터가 처음 붙었을 때 "현재 테이블에 이미 들어 있는 데이터"는 binlog에 더 이상 남아 있지 않을 수 있다.

그래서 커넥터는 처음 기동될 때 consistent snapshot(일관된 스냅샷)을 한 번 수행한다. 스냅샷은 현재 데이터베이스 상태의 baseline을 잡아주는 과정이며, 스냅샷을 뜬 binlog 위치를 기록해 두었다가 그 지점부터 스트리밍을 이어간다. 기본 동작(global read lock 사용)을 단순화하면 다음과 같다.

1. DB에 연결한다.
2. 캡처할 테이블을 결정한다.
3. global read lock을 잡아 다른 클라이언트의 쓰기를 잠시 막는다.
4. repeatable read 트랜잭션을 시작해 이후 읽기가 일관된 스냅샷을 보도록 한다.
5. 현재 binlog 위치를 읽어 기록한다. (이후 스트리밍의 시작점)
6. 캡쳐 대상 테이블의 구조를 읽어 내부 schema history 토픽에 저장한다.
7. read lock을 해제한다.
8. 5번에서 읽은 binlog 위치 기준으로 각 테이블을 스캔하며, 행마다 read 이벤트(op : "r")를 만들어 해당 테이블의 Kafka 토픽으로 보낸다.
9. 트랜잭션을 커밋한다.
10. 스냅샷 완료를 offset에 기록한다.

중간에 커넥터가 멈췄다 재시작하면 마지막으로 처리한 위치부터 이어서 스트리밍한다.
만약 재시작 시점에 binlog가 이미 정리되어 위치를 찾을 수 없으면 커넥터는 실패하고 새 스냅샷이 필요하다는 에러를 발생시킨다.
이 상황을 자동 처리하려면 snapshot.mode를 when_needed로 둔다.

<snapshot.mode의 주요 값>

값	동작
initial(기본)	최초 1회 스냅샷 후 스트리밍
initial_only	스냅샷만 수행하고 스트리밍은 하지 않음
no_data	모든 관련 테이블의 구조만 캡처하고 데이터 read 이벤트는 만들지 않음
never	스냅샷 없이 곧장 스트리밍 시작
when_needed	offset이 없거나 기록된 위치가 서버에 없을 때만 스냅샷
recovery	유실/손상된 schema history 토픽을 복구
always	시작할 때마다 매번 스냅샷

=> 대용량 테이블이라 초기 스냅샷이 부담된다면, 스트리밍과 병행 가능한 incremental snapshot(청크 단위, 기본 1024행)도 신호(signal)를 통해 트리거할 수 있다.

2-3. Shema history topic : 과거 이벤트에 올바른 스키마를 입히기

테이블 구조는 언제든 바뀔 수 있고, binlog에는 스키마 변경 전후의 변경 이벤트가 섞여 있다. 커넥터가 모든 이벤트에 현재 스키마를 그대로 적용하면, 오래된 이벤트는 잘못 해석될 수 있다.

MySQL은 binlog에 row 변경뿐만 아니라 DDL까지 기록한다. 커넥터는 binlog를 읽다가 DDL을 만나면 이를 파싱해 각 테이블 스키마의 in-memory 표현을 갱신하고, 동시에 모든 DDL을 그것이 등장한 binlog 위치와 함께 별도의 데이터베이스 스키마 history Kafka 토픽에 기록한다. 커넥터는 각 이벤트가 발생한 그 시점의 테이블 구조를 정확히 재현해 올바른 change event를 만든다.

⚠️ 재시작/장애 복구 시에도 schema history 토픽이 핵심!
커넥터는 특정 binlog 위치부터 다시 읽기 시작하는데, 그 위치까지의 DDL을 schema history 토픽에서 재생해 테이블 구조를 복원한 뒤 스트리밍을 재개한다. (이 schema history 토픽은 커넥터 내부 전용이며 컨슈머가 직접 읽으면 안 된다.)

2-4. 지원 토폴로지

커넥터는 binlog상 자신의 위치를 기록하는데, 이 위치는 클러스터 내 서버마다 다르다. 따라서 커넥터는 단 하나의 MySQL 인스턴스만 따라간다.

토폴로지	요지
Standalone	단일 서버. binlog만 켜져 있으면 됨
Primary&Replica	primary 또는 (binlog가 켜진)replica 하나를 따라감. 그 서버가 죽으면 복구 후에야 재개
HA(High-Availability)	대부분 GTID를 쓰므로 replica가 변경을 추적 가능
Multi-primary	NDB 클러스터 복제. GTID 필수, GTID source 필터로 페일오버 가능
Hosted	global read lock 불가 -> table-level lock으로 스냅샷

- GTID(Global Transaction Identifiler)
: 원본 서버(Source)에서 Commit된 각각의 트랜잭션들이 가지게되는 고유한 전역 식별자로 원본 서버에만 고유한것이 아니라 복제 설정의 모든 서버에서 고유한 값을 가진다. 모든 트랜잭션과 모든 GITD 사이에는 일대일 매핑이 있다.

- NDB 클러스터(MySQL NDB Cluster)
: 공유 없는(Shared-Nothing) 아키텍처를 기반으로 99.999%의 고가용성과 실시간 ACID 트랜잭션, 자동 샤딩을 제공하는 분산 데이터베이스 기술

Primay&Replica에 대해서

- Replica : MySQL 복제(replication) 구조에서 primary 서버의 데이터를 복사해 따라가는 서버.(master/slave 구조에서 slave에 해당)

- Primary(쓰기를 받는 원본 서버)가 모든 변경을 자기 binlog에 기록하면, replica는 그 binlog를 받아 자기 DB에 똑같이 재생(replay)한다. 그래서 replica는 primary와 거의 같은 데이터를 가진 읽기 전용 복제본 역할을 하고, 보통 읽기 부하 분산이나 장애 대비용으로 둔다.

- Debezium 입장에서 중요한 건, 커넥터가 primary가 아니라 replica에 붙어서 binlog를 읽어도 된다는 점이다. 운영중인 primary에 부하를 주기 싫을 때 유용하다. (단, replica에 binlog를 켜두고 있어야 한다.)

[그림으로 이해]
binlog 복제
쓰기 ----> Primay(원본) ----------------> replica(복제본) <<<<< Debezium이 여기에 붙음

3. Change Event 뜯어보기

3-1. 토픽 네이밍 규칙

커넥터는 한 테이블의 모든 변경 이벤트를 그 테이블 전용 Kafka 토픽 하나에 쓴다. 기본 네이밍 규칙은 아래와 같다.

topicPrefix.databaseName.tableName

ex) 예를 들어 topic.prefix가 fulfillment, 데이터베이스가 inventory이고, 테이블이 orders, customers, products라면?

fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.products

=> 원하는 이름이 아니라면 Topic Routing SMT로 커스터마이징 가능

3-2. 이벤트의 기본 골격 : key와 value

모든 change event는 key와 value를 가지며, 각각은 (JSON 컨버터를 모든 컴포넌트를 생성하도록 설정한 경우) schema와 payload로 구성된다. schema는 데이터의 구조, 타입을 기술하고 payload는 실제 값을 담는다. 이벤트가 스스로 자신의 구조를 설명하므로 컨슈머가 변화에 견딜 수 있다.

event key의 payload는 변경된 행의 기본 키(PK) 구조를 반영한다. PK가 없으면 unique key를, message.key.columns로 재정의했다면 그 키를 따른다.
event value의 payload는 변경된 행 자체의 구조를 반영하며, 중첩 스키마를 갖는 게 보통이다.

필드	의미
op	작업 종류. c(create), u(update), d(delete), r(read=스냅샷)
before	변경 전 행 상태(insert면 null)
after	변경 후 행 상태(delete면 null)
source	이벤트 출처 메타데이터(커넥터 버전, db/table, binlog 파일 위치, ts 등)
ts_ms	커넥터가 이벤트를 처리한 시각(ms)

3-3. op별 payload 차이

op값에 따라 payload가 어떻게 달라지는지 알아보자.

1. create(op:"c") - before는 null, after는 새 행

{
  "before": null,
  "after": {
    "id": 1004,
    "first_name": "Anne",
    "last_name": "Kretchmar",
    "email": "annek@noanswer.org"
  },
  "source": { "...": "..." },
  "op": "c",
  "ts_ms": 1465581029523
}

2. update(op:"u") - before는 변경 전, after는 변경 후

{
  "before": {
    "id": 1004,
    "first_name": "Anne",
    "last_name": "Kretchmar",
    "email": "annek@noanswer.org"
  },
  "after": {
    "id": 1004,
    "first_name": "Anne Marie",
    "last_name": "Kretchmar",
    "email": "annek@noanswer.org"
  },
  "source": { "...": "..." },
  "op": "u",
  "ts_ms": 1465581029523
}

3. delete(op:"d") - before에는 삭제 직전 행, after은 null

{
  "before": {
    "id": 1004,
    "first_name": "Anne Marie",
    "last_name": "Kretchmar",
    "email": "annek@noanswer.org"
  },
  "after": null,
  "source": { "...": "..." },
  "op": "d",
  "ts_ms": 1465581029523
}

=> before가 의미 있는 값을 가지려면 MySQL에는 binlog_row_image=FULL 이어야 한다.

3-4. tombstone 이벤트

delete가 발생하면 커넥터는 보통 두 개의 메시지를 보낸다. 하나는 op:"d" 이벤트이고, 그 직후 같은 키에 value가 null인 메시지(tombstone)를 하나 더 보낸다.

이건 Kafka의 log compaction 때문이다. compaction은 같은 키의 마지막 메시지만 남기는데, value가 null인 tombstone은 이 키를 완전히 제거해도 된다는 표시 역할을 한다. 즉, 삭제된 행에 대한 키가 토픽에 영구히 쌓이지 않게 해준다.

=> 이 동작이 불필요하면 tombstones.on.delete=false 로 끄면 된다.

- tombstone : 특정 키(Key)에 대한 데이터가 삭제되었음을 알리기 위해 보내는 빈 값(Null)의 메시지
- log compaction : 아파치 카프카(Apache Kafka) 등에서 동일한 키(Key)를 가진 데이터 중 가장 최신 값만 남기고 과거의 데이터는 삭제하여 저장 공간을 효율적으로 관리하고 상태(State)를 유지하는 데이터 삭제 정책

3-5. 데이터 유형 매핑

after, before에 담긴 컬럼 값이 어떤 형태의 JSON으로 나오는지는 MySQL 컬럼 타입에 따라 달라진다. 컨슈머가 값을 역직렬화할 때 직접 부딪히는 부분이라 중요한 부분이다.

기본 타입(numeric, string 등)은 대체로 직관적으로 매핑된다. (INT -> 정수, BIGINT -> 정수(클 경우 long), VARCHAR/TEXT -> 문자열, FLOAT/DOUBLE -> 실수) 주의해야할 것은 아래 세 타입이다.

1. 시간 타입(temporal)

Debezium은 MySQL의 시간 타입을 사람이 읽는 문자열이 아니라 epoch 기준 정수로 보내는 경우가 많고, 정밀도(time.precision.mode)에 따라 표현이 달라진다.

MySQL 타입	기본 표현(논리 타입)	값의 의미
DATE	io.debezium.time.Date	epoch 이후 일수(정수)
DATETIME	io.debezium.time.Timestamp	epoch 이후 밀리초 (정밀도↑면 Micro/Nano)
TIMESTAMP	io.debezium.time.ZonedTimestamp	UTC 기준 ISO-8601 문자열(예: 2025-01-01T09:00:00Z)
TIME	io.debezium.time.MicroTime	자정 이후 경과 시간(정수)

⚠️여기서 핵심은 DATETIME 과 TIMESTAMP 이다.

- DATETIME은 타임존 정보가 없는 벽시계 값이라 Debezium이 그대로 epoch 밀리초로 변환
- TIMESTAMP는 UTC로 정규화된 문자열
- 컨슈머에서 DATETIME 값을 받아 16172531200000 같은 숫자가 찍히면 epoch 밀리초다.
- 사람이 읽는 문자열로 받고 싶으면 소스 설정의 time.precision.mode나 TimezoneConverter SMT 또는 컨슈머 측 변환이 필요하다.

2. Decimal 타입

DECIMAL/NUMERIC은 정밀도 손실을 막기 위해 기본적으로 Java BigDecimal을 base64로 인코딩한 바이트로 나온다. JSON에서 이상한 문자열로 보일 수 있는데, 숫자 그대로 받고 싶으면 decimal.handling.mode를 double(실수, 정밀도 손실 가능) 또는 string(문자열)으로 바꾼다.

3. Boolean

MySQL에는 진짜 boolean이 없고 TINYINT(1)로 표현한다. Debezium은 기본적으로 이를 정수로 보내며, 진짜 boolean으로 받고 싶으면 TINYINT(1) to Boolean 커스텀 컨버터를 쓴다.

Tip!!!

소스 커넥터에 column.propagate.source.type/datatype.propagate.source.type 을 켜면, 이벤트 스키마에 원본 컬럼의 타입, 길이, 스케일 메타데이터가 함께 실린다. 이는 특히 JDBC Sink 로 같은 종류 DB에 다시 쓸 때 Sink가 컬럼 타입을 더 정확히 재현하도록 돕는다.

4. MySQL 사전 설정

커넥터를 띄우기 전에 MySQL 쪽 사전 준비가 필요하다.

4-1. 전용 유저와 권한

커넥터가 쓸 전용 계정을 만들고 필요한 권한을 부여한다.

SELECT : 스냅샷 시 테이블 데이터 읽기
RELOAD : flush로 global read lock 등을 수행
SHOW DATABASES : 데이터베이스 목록 조회
REPLICATION SLAVE : binlog를 읽기 위한 핵심 권한
REPLICATION CLIENT : binlog 위치/상태 조회

4-2. binlog 활성화

MySQL 서버 설정에 다음을 적용한다.

[mysqld]
server-id              = 223344        # 클러스터 내 유일한 값
log_bin                = mysql-bin      # binlog 파일 prefix (binlog 활성화)
binlog_format          = ROW           # row 기반이어야 변경 데이터를 캡처
binlog_row_image       = FULL          # before/after에 전체 컬럼을 기록
binlog_expire_logs_seconds = 864000    # binlog 보존 기간(예: 10일)

ROW 포맷이어야 행 단위 변경을 캡쳐하고, FULL이어야 update/delete 이벤트의 before에 모든 컬럼이 담긴다.
보존 기간이 너무 짧으면 커넥터가 잠시 멈춘 사이 binlog가 정리되어 재스냅샷이 필요해질 수 있으니, 운영 환경의 다운타임을 감안해 넉넉히 잡는다.

4-3. GTID 활성화(선택 사항)

GTID(Global Transaction Identifier)를 켜면 replica 간 페일오버나 read-only incrememtal snapshot 같은 기능이 수월해진다.

[mysqld]
gtid_mode                 = ON
enforce_gtid_consistency  = ON

4-4. 세션 타임아웃 주의

대용량 초기 스냅샷은 시간이 오래 걸릴 수 잇다. 이때 interactive_timeout, wait_timeout이 너무 짧으면 스냅샷 도중 연결이 끊길 수 있으니 충분히 키워둬야 한다.

5. Connector 배포 & 설정

Debezium 커넥터는 Kafka Connect 워커 위에서 동작한다. 플러그인(JAR)을 워커의 plugin path에 두고, 설정 JSON을 Connect REST API로 등록하면 된다.

5-1. 설정 예시

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",

    "database.hostname": "mysql.internal",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",

    "topic.prefix": "fulfillment",
    "table.include.list": "inventory.orders,inventory.customers",

    "snapshot.mode": "initial",

    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-history.inventory"
  }
}

connector.class — MySQL 커넥터 구현 클래스. 고정값.
tasks.max — MySQL 커넥터는 binlog를 단일 스트림으로 읽으므로 항상 1이다.
database.* — 접속 정보. database.server.id는 이 커넥터가 MySQL 복제 클라이언트로서 가질 유일한 식별자다(서버의 server-id와도 겹치면 안 됨).
topic.prefix — 이 커넥터가 만드는 모든 토픽의 네임스페이스이자 논리 이름. 커넥터마다 유일해야 한다.
table.include.list — 캡처할 테이블을 db.table 형식으로 지정(미지정 시 전체).
schema.history.internal.kafka.* — 내부 schema history 토픽이 저장될 Kafka와 토픽명.

5-2. 등록 방법

설정 JSON을 Connect REST API로 POST한다.

curl -i -X POST \
  -H "Accept:application/json" -H "Content-Type:application/json" \
  http://connect:8083/connectors/ \
  -d @inventory-connector.json

<상태 확인>

curl -s http://connect:8083/connectors/inventory-connector/status | jq

5-3. 동작 확인

등록되면 커넥터는 (기본 initial 모드 기준) 스냅샷을 수행하고 토픽을 만든 뒤 스트리밍으로 전환한다. 생성된 토픽을 구독해 이벤트가 흐르는지 확인한다.

kafka-console-consumer --bootstrap-server kafka:9092 \
  --topic fulfillment.inventory.customers --from-beginning

MySQL에서 INSERT/UPDATE/DELETE를 실행해 보면, op: c/u/d 이벤트가 토픽으로 들어오는 것을 볼 수 있다.

6. 운영 고도화

6-1. 운영 시 주의사항

재스냅샷 함정: 커넥터가 오래 멈춰 있는 동안 binlog가 삭제되면, 재시작 시 위치를 못 찾아 실패한다. binlog 보존 기간을 넉넉히 두고, 필요하면 snapshot.mode=when_needed로 자동 복구를 허용한다.
schema history 토픽 보호: 파티션 1개 유지, 함부로 삭제 금지. 손상되면 recovery 모드로 복구한다.
모니터링: 커넥터는 JMX로 snapshot/streaming/schema-history 메트릭을 노출한다. 스트리밍 지연(lag), 처리 이벤트 수 등을 대시보드로 관찰하면 장애를 빨리 잡을 수 있다.
tombstone & compaction: 다운스트림이 log compaction 토픽이라면 tombstone 동작을 이해하고 컨슈머가 null value를 올바르게 처리하도록 해야 한다.

6-2. 신호(Signal)로 제어하기

실제 운영에서는 무중단으로 동작을 바꿔야 하는 순간이 온다. 새 테이블 추가, 특정 데이터 재캡처, 손상 구간 건너뛰기 등. Debezium은 이를 위해 신호(signal) 메커니즘을 제공한다. 신호는 보통 소스 DB에 만들어 둔 signaling 테이블(예: debezium_signal)에 INSERT를 넣거나, 별도 Kafka signaling 토픽에 메시지를 보내 전달한다. 커넥터는 이 신호를 감지해 요청된 작업을 수행한다.

1. Incremental Snapshot(증분 스냅샷)

초기 스냅샷은 전체를 한 번에 뜨느라 대용량 테이블에서 오래 걸리고, 그동안 스트리밍이 밀린다. 증분 스냅샷은 이 문제를 푼다. 테이블을 기본 키 기준으로 정렬해 청크(chunk, 기본 1024행) 단위로 나눠, 스트리밍과 병행하며 조금씩 캡처한다. 각 행은 op: "r"(read) 이벤트로 나온다.

핵심 장점:

스트리밍을 막지 않는다. 스냅샷이 도는 중에도 실시간 변경 이벤트가 계속 흐른다.
중단되어도 이어서 재개한다. 처음부터 다시 뜨지 않고 멈춘 지점부터 계속한다.
언제든 on-demand로 실행할 수 있다. 운영 중 table.include.list에 테이블을 추가한 뒤 그 테이블만 따로 채우는 데 적합하다.

활용 예시) 운영 중 새 테이블을 캡처 대상에 추가했을 때, 그 테이블의 기존 데이터를 스트리밍을 멈추지 않고 채워 넣는 경우. signaling 테이블에 다음과 같이 신호를 넣는다.

INSERT INTO db1.debezium_signal (id, type, data)
VALUES (
  'ad-hoc-1',
  'execute-snapshot',
  '{"data-collections": ["db1.orders", "db1.customers"],
    "type": "incremental"}'
);

특정 조건의 행만 다시 캡처하고 싶으면 additional-conditions로 WHERE 필터를 걸 수 있다.

INSERT INTO db1.debezium_signal (id, type, data)
VALUES (
  'ad-hoc-2',
  'execute-snapshot',
  '{"data-collections": ["db1.orders"],
    "type": "incremental",
    "additional-conditions": [{"data-collection": "db1.orders", "filter": "region=\'KR\'"}]}'
);

증분 스냅샷이 도는 동안 같은 행에 대해 스냅샷의 read 이벤트와 실시간 update/delete 이벤트가 충돌할 수 있다. Debezium은 snapshot window라는 구간에서 기본 키를 비교해 중복을 제거(de-duplication)하고, 실시간 이벤트가 항상 스냅샷 값을 이기도록(logically supersede) 보장한다. 덕분에 병행 처리에도 최종 상태가 어긋나지 않는다.

반대로 잘못 시작한 증분 스냅샷을 멈추려면 stop-snapshot 신호를 보낸다. 또한 한 번에 끝내는 blocking snapshot(스트리밍을 잠시 멈추고 초기 스냅샷처럼 수행)도 신호로 트리거할 수 있는데, 대용량 신규 테이블을 빠르게 채울 때 쓴다.

7.2 set-binlog-position: 스트림 위치 직접 옮기기

평소 커넥터는 자신이 어디까지 읽었는지를 offset으로 관리하며 알아서 이어간다. 하지만 드물게 읽기 시작 위치를 사람이 직접 지정해야 할 때가 있다. 이때 set-binlog-position 신호를 보낸다.

언제 쓰나:

재해 복구: 시스템 장애 후 마지막으로 정상이었던 위치부터 재개.
손상 데이터 스킵: DB나 토픽에서 손상 구간을 발견했을 때, 그 뒤의 정상 위치로 건너뜀.
DB 리프레시 후 과거 재생 방지: 백업에서 DB를 복원한 뒤 커넥터가 옛 위치부터 전체를 재생하려는 것을 막고, 최근 지점부터 시작.
부분 복제 / 테스트: 특정 지점부터의 변경만 처리.

지정 방식은 GTID 사용 여부에 따라 둘 중 하나다. GTID가 꺼져 있으면 binlog 파일명과 위치로 지정한다.

INSERT INTO debezium_signal (id, type, data) VALUES (
  'set-position-001',
  'set-binlog-position',
  '{"binlog_filename": "mysql-bin.000003", "binlog_position": 1234}'
);

GTID가 켜져 있으면 GTID set으로 지정한다(둘을 동시에 쓸 수 없다).

INSERT INTO debezium_signal (id, type, data) VALUES (
  'set-gtid-001',
  'set-binlog-position',
  '{"gtid_set": "3E11FA47-71CA-11E1-9E33-C80AA9429562:1-100"}'
);

신호를 보낸 뒤 커넥터를 재시작하면 새 위치부터 스트리밍한다.

⚠️ 양날의 검이다. 지정한 위치가 현재보다 앞(미래) 이면 그 사이의 변경 이벤트를 건너뛴다(유실). 반대로 현재보다 뒤(과거)면 이미 보낸 이벤트를 다시 보내 중복이 발생한다. 그래서 이 신호는 위 같은 예외 상황에서, 결과를 정확히 이해한 상태로만 써야 한다.

7. 마치며

Debezium MySQL Source Connector에 대해 알아보았다. 처음에는 막막했는데 이렇게 정리하고 보니 구조가 잡히는 느낌이다. 전체적인 동작 흐름(binlog → change event → Kafka 토픽으로 변경을 흘려보내는 원리)을 이해 했으니 이제 실전 프로젝트에 적용해보고자 한다. 추가적으로 Sink Connector부분도 공부해서 정리해보겠다.

[참고]

https://debezium.io/documentation/reference/3.6/connectors/mysql.html

Debezium connector for MySQL :: Debezium Documentation

Records the number of change events that the connector identifies as erroneous during a snapshot operation. The connector increments this metric each time that it encounters an event that it cannot process during an initial, incremental, or ad hoc snapshot

debezium.io

저작자표시 비영리 (새창열림)

This is my space.