Kafka Connect로 CDC 파이프라인 세우기

참고자료

• Kafka Connect 공식문서
• Debezium Documentation
• Debezium - SQL Server Connector
• MySQL 8.4 Reference - Replication
• MySQL 8.4 Reference - The Binary Log
• MySQL 8.4 Reference - Replication with GTIDs
• 실습 템플릿 저장소

---

배경

DB의 변경을 다른 시스템에 전달해야 했다. 처음 떠오른 방법은 애플리케이션이 저장할 때 이벤트도 함께 발행하는 것이었다.

그런데 이 방식에는 앞서 FCM을 붙이면서 겪은 문제와 같은 구멍이 있다. 저장과 발행이 다른 시스템에 대한 두 개의 쓰기라서, 하나가 성공하고 하나가 실패할 수 있다.

CDC는 이 문제를 다른 자리에서 푼다. 애플리케이션이 아니라 DB가 이미 남기고 있는 변경 기록을 읽는다.

정리하면서 확인하고 싶었던 것들이다.

• CDC가 읽는다는 “변경 기록”이 정확히 무엇인가?
• 애플리케이션에서 이벤트를 발행하는 것과 무엇이 다른가?
• 커넥터가 죽었다 살아나면 어디서부터 다시 읽는가?
• 도입하면 무엇을 감수해야 하는가?

---

구축할 환경은 각기 다른 온프레미스 서버 인스턴스에 Kafka를 KRaft 모드로 띄우고 그 위에 CDC를 올리는 것이다.

따라서 각 노드에는 Controller, Broker, Connect 총 3개의 프로세스가 동작한다. KRaft 모드 자체는 따로 정리한 글에 있다.

---

Step 1. MSSQL 접속 및 데이터베이스, 스키마, 테이블 생성

1. SOURCE, EXTERNAL 데이터베이스를 생성
2. SOURCE, SOURCE 데이터베이스 내에 dbo 스키마를 사용
3. SOURCE, SOURCE 데이터베이스 내에 dbo 스키마 내에 MEMBER_BASE 테이블 생성

-- SOURCE 라는 데이터베이스를 사용
USE [EXTERNAL];

create table dbo.MEMBER_BASE(
    member_id bigint primary key,
    nickname nvarchar(50)
)

USE SOURCE;

create table dbo.MEMBER_BASE(
    member_id bigint primary key,
    nickname nvarchar(50)
)

EXEC sys.sp_cdc_enable_db;

ALTER DATABASE SOURCE SET CHANGE_TRACKING = ON(CHANGE_RETENTION = 3 DAYS, AUTO_CLEANUP = ON)

EXEC sys.sp_cdc_enable_table
      @source_schema = 'dbo',
      @source_name = 'MEMBER_BASE',
      @role_name = 'sa';

---

Step 2. Source, Sink Connector 등록

HTTP 요청으로 http://localhost:8083/connectors 혹은 http://localhost:8084/connectors 혹은 http://localhost:8085/connectors 으로 DB에 관한 커넥트를 등록한다. 이 때 메서드는 POST 요청으로 보내야하며, Requset Body는 아래와 같다.

1. 소스 커넥터 등록 [POST] http://localhost:8083/connectors

Confluent 공식문서 - JDBC Source Configure

{
    // 기본값: 없음 (필수 설정)
    // 다른 옵션: 사용자가 지정한 어떤 문자열이든 가능함
    "name": "mssql-cdc-member-source-connector",
    "config": {
        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 다른 source connector 클래스 (예: MySqlConnector, PostgresConnector 등)
        // 중요도: high
        "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",

        // 기본값: 1
        // 다른 옵션: 1보다 큰 정수 (병렬 처리를 위해 증가 가능)
        // 중요도: low
        "tasks.max": "1",

        // 기본값: false
        // 다른 옵션: true (SSL/TLS를 사용한 암호화된 연결)
        // 중요도: medium
        "database.encrypt": false,

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: SQL Server가 실행 중인 다른 호스트 이름이나 IP 주소
        // 중요도: high
        "database.hostname": "host.docker.internal",

        // 기본값: 1433 (SQL Server의 기본 포트)
        // 다른 옵션: SQL Server가 사용하는 다른 포트 번호
        // 중요도: high
        "database.port": "1433",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 데이터베이스에 접근 권한이 있는 다른 사용자 이름
        // 중요도: high
        "database.user": "SA",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 해당 사용자의 올바른 비밀번호
        // 중요도: high
        "database.password": "admin123$%",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 다른 데이터베이스 이름 또는 여러 데이터베이스 (쉼표로 구분)
        // 중요도: high
        "database.names": "SOURCE",

        // 기본값: 없음
        // 다른 옵션: 다른 스키마 이름 또는 여러 스키마 (쉼표로 구분)
        // 중요도: medium
        "schema.include.list": "dbo",

        // 기본값: 없음
        // 다른 옵션: 다른 테이블 이름 또는 여러 테이블 (쉼표로 구분)
        // 중요도: medium
        "table.include.list": "dbo.MEMBER_BASE",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: Kafka 브로커의 다른 주소와 포트
        // 중요도: high
        "database.history.kafka.bootstrap.servers": "broker1:19091,broker2:29092,broker3:39093",

        // 기본값: UTC
        // 다른 옵션: 다른 유효한 타임존 ID
        // 중요도: medium
        "db.timezone": "Asia/Seoul",

        // 기본값: 서버 이름
        // 다른 옵션: 사용자가 지정한 다른 접두사
        // 중요도: high
        "topic.prefix": "member-source",

        // 기본값: 없음
        // 다른 옵션: 다른 transform 이름 또는 여러 transform (쉼표로 구분)
        // 중요도: low
        "transforms": "route",

        // 기본값: 없음
        // 다른 옵션: 다른 Kafka Connect transform 클래스
        // 중요도: low
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",

        // 기본값: 없음
        // 다른 옵션: 다른 정규식 패턴
        // 중요도: low
        "transforms.route.regex": "member-source.SOURCE.dbo.MEMBER_BASE",

        // 기본값: 없음
        // 다른 옵션: 다른 대체 문자열
        // 중요도: low
        "transforms.route.replacement": "source",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: Kafka 브로커의 다른 주소와 포트
        // 중요도: high
        "schema.history.internal.kafka.bootstrap.servers": "broker1:19091,broker2:29092,broker3:39093",

        // 기본값: ${database.server.name}.schema-changes.inventory
        // 다른 옵션: 사용자가 지정한 다른 토픽 이름
        // 중요도: medium
        "schema.history.internal.kafka.topic": "schema-history-cdc"

        // -----------------고려해볼만한 옵션------------------- //

        // 기본값: initial (중요도: high)
        // 다른 옵션: latest, timestamp
        // 설명: 커넥터가 시작될 때 데이터를 읽기 시작할 위치를 지정
        "snapshot.mode": "initial",

        // 기본값: 8192 (중요도: medium)
        // 다른 옵션: 1024에서 1048576 사이의 정수
        // 설명: 데이터베이스 로그에서 한 번에 읽어올 최대 이벤트 수
        "max.queue.size": "16384",

        // 기본값: 0 (중요도: medium)
        // 다른 옵션: 0보다 큰 정수 (밀리초 단위)
        // 설명: 폴링 간격. 0은 지속적인 폴링을 의미
        "poll.interval.ms": "1000",

        // 기본값: 10 (중요도: medium)
        // 다른 옵션: 1에서 100 사이의 정수
        // 설명: 커넥터가 데이터베이스에 연결을 시도할 최대 횟수
        "connect.max.retries": "10",

        // 기본값: 10000 (중요도: medium)
        // 다른 옵션: 1000에서 60000 사이의 정수 (밀리초 단위)
        // 설명: 연결 재시도 사이의 대기 시간
        "connect.backoff.max.ms": "10000",

        // 기본값: false (중요도: medium)
        // 다른 옵션: true
        // 설명: 데이터베이스 테이블의 컬럼 이름을 소문자로 변환할지 여부
        "column.lowercase": "false",

        // 기본값: 8192 (중요도: medium)
        // 다른 옵션: 1024에서 1048576 사이의 정수
        // 설명: 트랜잭션 로그에서 한 번에 처리할 최대 배치 크기
        "max.batch.size": "8192",

        // 기본값: 0 (중요도: medium)
        // 다른 옵션: 0 이상의 정수 (밀리초 단위)
        // 설명: 각 배치 처리 사이의 지연 시간. 0은 지연 없음을 의미
        "poll.interval.ms": "1000",

        // 기본값: false (중요도: high)
        // 다른 옵션: true
        // 설명: 데이터베이스의 시스템 테이블에서 변경 사항을 캡처할지 여부
        "include.schema.changes": "true",

        // 기본값: 0 (중요도: medium)
        // 다른 옵션: 0 이상의 정수
        // 설명: 스키마 변경 이벤트의 최대 크기 (바이트 단위). 0은 제한 없음을 의미
        "max.queue.size.in.bytes": "104857600",

        // 기본값: adaptive_time_microseconds (중요도: medium)
        // 다른 옵션: adaptive, adaptive_time, commit_time
        // 설명: 변경 이벤트의 순서를 결정하는 방법
        "event.processing.failure.handling.mode": "warn",

        // 기본값: fail (중요도: high)
        // 다른 옵션: warn, skip, stop
        // 설명: 변경 이벤트 처리 중 오류 발생 시 처리 방법
        "event.processing.failure.handling.mode": "warn"
    }
}

2. 싱크 커넥터 등록 [POST] http://localhost:18083/connectors

Confluent 공식문서 - JDBC Sink Configure

Debezium 공식문서 - SQL Server Connectors

{
    // 기본값: 없음 (필수 설정)
    // 다른 옵션: 사용자가 지정한 어떤 문자열이든 가능함
    "name": "mssql-cdc-member-sink-connector",
    "config": {
        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 다른 sink connector 클래스 (예: ElasticsearchSinkConnector, S3SinkConnector 등)
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",

        // 기본값: 없음 (필수 설정, 중요도: high)
        // 다른 옵션: 다른 데이터베이스 URL (예: MySQL, PostgreSQL 등의 JDBC URL)
        "connection.url": "jdbc:sqlserver://host.docker.internal:1433;databaseName=EXTERNAL",

        // 기본값: null (필수 설정, 중요도: high)
        // 다른 옵션: 데이터베이스에 접근 권한이 있는 다른 사용자 이름
        "connection.user": "sa",

        // 기본값: null (필수 설정, 중요도: high)
        // 다른 옵션: 해당 사용자의 올바른 비밀번호
        "connection.password": "admin123$%",

        // 기본값: false (중요도: medium)
        // 다른 옵션: true (테이블이 존재하지 않을 경우 자동으로 생성)
        "auto.create": "false",

        // 기본값: false (중요도: medium)
        // 다른 옵션: true (스키마 변경 시 테이블 구조를 자동으로 변경)
        "auto.evolve": "false",

        // 기본값: false (중요도: medium)
        // 다른 옵션: true (현재 설정된 값, 삭제 작업 활성화)
        // 주의: pk.mode가 record_key로 설정되어야 함
        "delete.enabled": "true",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 1보다 큰 정수 (병렬 처리를 위해 증가 가능)
        "tasks.max": "1",

        // 기본값: 없음 (필수 설정)
        // 다른 옵션: 다른 토픽 이름 또는 여러 토픽 (쉼표로 구분)
        "topics": "member-source.SOURCE.dbo.MEMBER_BASE",

        // 기본값: ${topic} (중요도: medium)
        // 다른 옵션: 다른 테이블 이름 형식 (예: "${topic}" 사용 가능)
        "table.name.format": "EXTERNAL.dbo.MEMBER_BASE",

        // 기본값: none (중요도: high)
        // 다른 옵션: kafka, record_key, record_value (현재 설정)
        "pk.mode": "record_value",

        // 기본값: insert (중요도: high)
        // 다른 옵션: upsert (현재 설정), update
        "insert.mode": "upsert",

        // 기본값: 없음
        // 다른 옵션: 다른 transform 이름 또는 여러 transform (쉼표로 구분)
        "transforms": "unwrap",

        // 기본값: 없음
        // 다른 옵션: 다른 Kafka Connect transform 클래스
        "transforms.unwrap.type": "org.apache.kafka.connect.transforms.ExtractField$Value",

        // 기본값: 없음
        // 다른 옵션: 추출하고자 하는 다른 필드 이름
        "transforms.unwrap.field": "after",

        // 소스 데이터의 CREATED_AT과 UPDATED_AT 값이 Unix timestamp 형식(밀리초)일 경우 (ex) 1644421787000 = 2022-02-09 15:16:27 UTC)
        // SQL Server의 datetime 타입은 이러한 큰 숫자를 직접 받아들일 수 없어 삽입 전에 별도로 변환해줘야한다.
        "transforms.convertInsDate.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.convertInsDate.target.type": "Timestamp",
        "transforms.convertInsDate.field": "CREATED_AT",
        "transforms.convertInsDate.format": "yyyy-MM-dd HH:mm:ss.SSS",
        "transforms.convertUptDate.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.convertUptDate.target.type": "Timestamp",
        "transforms.convertUptDate.field": "UPDATED_AT",
        "transforms.convertUptDate.format": "yyyy-MM-dd HH:mm:ss.SSS",

        // 기본값: initial (중요도: high)
        // 다른 옵션: schema_only, initial, never
        // 설명: 커넥터 시작 시 스냅샷 모드를 지정. 'schema_only'는 스키마만 캡처하고 데이터는 캡처하지 않음
        "snapshot.mode": "schema_only",

        // 기본값: repeatable_read (중요도: medium)
        // 다른 옵션: read_committed, read_uncommitted, exclusive
        // 설명: 스냅샷 생성 시 사용할 트랜잭션 격리 수준. 스냅샷이 생성되는 동안 다른 트랜잭션에 의한 데이터 변경을 볼 수 없어 스냅샷 전체에 걸쳐 일관된 데이터 뷰를 보장한다.
        "snapshot.isolation.mode": "repeatable_read",

        // -----------------고려해볼만한 옵션------------------- //

        // 기본값: 3000 (중요도: medium)
        // 다른 옵션: 1에서 수만까지의 정수
        // 설명: 단일 배치 작업에서 처리할 최대 레코드 수
        "batch.size": "3000",

        // 기본값: 0 (중요도: medium)
        // 다른 옵션: 0보다 큰 정수
        // 설명: 작업 실패 시 재시도할 최대 횟수
        "max.retries": "3",

        // 기본값: 0 (중요도: medium)
        // 다른 옵션: 0보다 큰 정수 (밀리초 단위)
        // 설명: 재시도 사이의 대기 시간
        "retry.backoff.ms": "1000",

        // 기본값: 300000 (5분, 중요도: medium)
        // 다른 옵션: 0보다 큰 정수 (밀리초 단위)
        // 설명: 커넥터가 새 메시지를 폴링하는 최대 간격
        "max.poll.interval.ms": "300000",

        // 기본값: 500 (중요도: medium)
        // 다른 옵션: 1에서 수천까지의 정수
        // 설명: 단일 폴링 작업에서 가져올 최대 레코드 수
        "max.poll.records": "500"
    }
}

---

Step 3. 테스트

이제 DB의 데이터를 삭제하고 생성하고 변경하는 등 쓰기요청을 수행한 후에 http:://localhost:8080 카프카 UI로 들어가서 CDC내용에 관한 토픽 메세지를 확인한다.

USE SOURCE;

insert into dbo.MEMBER_BASE(member_id, nickname) values (1, 'testNickname1');
insert into dbo.MEMBER_BASE(member_id, nickname) values (2, 'testNickname2');
insert into dbo.MEMBER_BASE(member_id, nickname) values (3, 'testNickname3');
insert into dbo.MEMBER_BASE(member_id, nickname) values (4, 'testNickname4');
insert into dbo.MEMBER_BASE(member_id, nickname) values (5, 'testNickname5');
insert into dbo.MEMBER_BASE(member_id, nickname) values (6, 'testNickname6');
insert into dbo.MEMBER_BASE(member_id, nickname) values (7, 'testNickname7');
insert into dbo.MEMBER_BASE(member_id, nickname) values (8, 'testNickname8');
insert into dbo.MEMBER_BASE(member_id, nickname) values (9, 'testNickname9');
insert into dbo.MEMBER_BASE(member_id, nickname) values (10, 'testNickname10');

INSERT를 실행하면 커넥터가 트랜잭션 로그의 변경을 읽어 토픽에 메시지를 만든다. 토픽 이름은 커넥터 설정의 접두어를 따라 <접두어>.<스키마>.<테이블> 형태로 자동 생성된다.

콘솔에서 직접 확인할 수 있다.

# 만들어진 토픽 목록
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 해당 토픽의 메시지를 처음부터 읽는다
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic mssql.dbo.MEMBER_BASE --from-beginning

커넥터 상태도 REST API로 확인한다.

curl -s localhost:8083/connectors/mssql-source-connector/status | jq

state가 RUNNING이 아니면 trace 필드에 실패 이유가 담겨 있다. 커넥터가 죽어도 카프카는 정상으로 보이므로, 이 상태를 따로 감시하지 않으면 데이터가 안 들어오는 것을 한참 뒤에 알게 된다.

메시지를 열어보면 아래와 같은 구조가 나온다.

"payload": {
  "before": null,
  "after": {
    "member_id": 1,
    "nickname": "testNickname1"
  },
}

위 항목으로 데이터가 어떻게 변경되었는지(생성, 변경, 삭제)를 파악할 수 있다.

생성

"payload": {
  "before": null,
  "after": {
    "member_id": 1,
    "nickname": "testNickname1"
  },
}

수정

"payload": {
  "before": {
    "member_id": 1,
    "nickname": "testNickname1"
  },
  "after": {
    "member_id": 1,
    "nickname": "testModifiedNickname1"
},

삭제

"payload": {
    "before": {
        "member_id": 2,
        "nickname": "testNickname2"
    },
    "after": null,
}

---

이론으로

로그 기반 CDC 5가지 장점

Debezium을 사용하는 Source CDC는 로그 기반으로 동작하게된다.

즉, DBMS의 binlog를 활용하여 변경된 데이터를 감지하게 되는 것인데 이 방식의 구체적인 내용과 장점을 살펴본다.

Binary Log

MySQL 공식문서 - BinaryLog

Binary Log는 데이터베이스의 변경 사항을 기록하는 기능이다. 주로 소스 서버의 변경 사항을 레플리카 서버로 전송하는 데이터복제나 백업 이후의 변경 사항을 재실행하여 최신 상태로 복구할 때 활용된다.

바이너리 로그는 예상치 못한 중단이 일어날 때를 대비하여 완전히 완료된 트랜잭션만 기록하고 다시 읽는 방식으로 동작한다.

즉, 트랜잭션이 끝나지 않은 내용은 이후 BinaryLog에 기록되지 않아 DB에 적용되지 않는다.

참고로, Debezium을 MySQL 사용하기 위해 Binary Log를 활용한다는 내용이었지만 MSSQL, PostgreSQL등 각자만의 Binary Log의 역할을 하는 개념을 도입시키거나 DBMS단위에서 자체적인 CDC를 제공한다.

---

Global Transaction ID - GTID

MySQL 공식문서 - Replication

MySQL에서는 Binary Log방식이 아닌 Transaction ID를 통해 복제를 수행하는 GTID방식도 존재한다.

---

정리하며

처음 던진 질문들에 대한 답이다.

변경 기록이 무엇인가. DB가 복제를 위해 이미 남기고 있는 로그다. MySQL이면 바이너리 로그, SQL Server면 CDC 테이블이다. 원래 목적은 복제본을 만드는 것이고, CDC 도구는 자기를 복제본인 척하고 그 로그를 읽는다.

그래서 DB 입장에서는 특별한 부담이 아니다. 어차피 남기고 있던 것을 하나 더 읽어갈 뿐이다. 애플리케이션 테이블에 트리거를 걸거나 주기적으로 조회하는 방식과 여기서 갈린다.

애플리케이션 발행과 무엇이 다른가. 원자성이다. 애플리케이션이 저장하고 나서 이벤트를 발행하면 그 사이에 죽을 수 있고, 그러면 데이터는 바뀌었는데 아무도 모른다. CDC는 커밋된 것만 로그에 남으므로 그 틈이 없다.

대신 애플리케이션의 의도가 사라진다. 로그에 남는 것은 “주문 테이블의 이 행이 이렇게 바뀌었다”이지 “주문이 취소됐다”가 아니다. 받는 쪽이 컬럼 변화를 보고 의미를 되짚어야 한다.

의미가 중요하면 아웃박스 패턴이 낫다. 이벤트를 같은 DB의 테이블에 함께 저장하면 원자성도 지키고 의미도 담을 수 있다. 그리고 그 테이블을 CDC로 읽으면 된다.

커넥터가 죽으면 어디서부터 다시 읽는가. 오프셋을 Kafka 토픽에 저장해두고 거기서부터 이어간다. 그래서 재시작해도 처음부터 다시 읽지 않는다.

여기에 조건이 있다. DB가 그 시점의 로그를 아직 갖고 있어야 한다. 바이너리 로그의 보관 기간이 지나면 그 구간이 사라지고, 그러면 전체를 다시 읽는 스냅샷부터 시작해야 한다. 커넥터가 오래 멈춰 있으면 이 상황이 온다.

감수할 것. 세 가지가 눈에 띄었다.

중복이 올 수 있다. 오프셋을 저장하기 직전에 죽으면 같은 변경이 두 번 발행된다. 받는 쪽이 여러 번 처리해도 결과가 같도록 만들어야 한다.

스키마 변경이 그대로 전파된다. 컬럼을 추가하거나 이름을 바꾸면 메시지 형태가 바뀐다. 스키마 레지스트리를 두거나 호환되는 방식으로만 바꾸는 규칙이 필요하다.

운영 대상이 늘어난다. Kafka Connect 클러스터가 하나 더 생기고, 커넥터마다 상태를 봐야 한다. 커넥터가 조용히 멈춰 있으면 데이터가 안 오는데 아무도 에러를 안 본다. 지연 시간을 지표로 뽑아 감시하는 것이 필수였다.

구축하고 나서 남은 감각은 CDC가 “이벤트를 만드는 도구”가 아니라 “이미 있는 것을 옮기는 도구” 라는 것이었다. 그래서 안전하지만, 그만큼 의미를 담을 자리가 없다. 무엇을 전달하고 싶은가에 따라 이 방식이 맞을 수도 아닐 수도 있었다.