Hong로그인
# MySQL# Kafka

MySQL CDC는 UPDATE 한 줄을 어떻게 Kafka로 보낼까?

@Hong2026-09-29

이 글에 담긴 코드와 결과는 직접 실행해 측정한 값을 바탕으로 작성했습니다. 실측은 Hong과 팀원들이 진행했고, 그 결과를 바탕으로 AI가 글을 쓰고 Hong이 검토합니다.

실습 코드 내려받기

질문
MySQL에서 UPDATE 한 줄이 CDC로 어떻게 Kafka 이벤트가 될까?
답
Debezium이 Binlog를 읽어 UPDATE 한 줄을 애플리케이션 코드 없이 변경 이벤트 하나로 만들었고, 이벤트에는 바뀌기 전 등급 1과 바뀐 뒤 등급 2가 함께 담겼다.
조건
MySQL 8.4, Kafka 3.9.1(apache/kafka:3.9.1), Debezium 3.1.3.Final, Docker Compose, 테이블 shop.users 하나

환경

MySQL
mysql:8.4 이미지
Kafka
3.9.1(apache/kafka:3.9.1)
Debezium
3.1.3.Final(quay.io/debezium/connect:3.1)
실행
Docker Compose로 세 컨테이너를 함께 띄움
데이터
테이블 shop.users 하나
목차
  1. CDC는 Binlog를 읽습니다
  2. 실습 준비
  3. 실습 1: 커넥터 등록
  4. 실습 2: UPDATE 한 줄이 이벤트가 되기까지
  5. 정리

이번 시간에는 MySQL에서 UPDATE 한 줄이 CDC를 거쳐 Kafka 이벤트가 되는 과정을 직접 따라가 보며 배워가는 시간입니다.

예를 들어서 회원 등급이 1에서 2로 바뀌었다고 해볼게요. 검색 인덱스는 새 등급으로 다시 색인해야 하고, 분석 시스템은 변경 이력을 쌓아야 하고, 캐시는 오래된 값을 지워야 하죠.

이럴 때 쓰는 게 CDC(Change Data Capture, 변경 데이터 캡처)예요. DB의 변경 기록을 읽어서 Kafka로 보내는 방식이에요.

CDC는 Binlog를 읽습니다

MySQL은 데이터를 바꿀 때마다 그 내용을 Binlog(바이너리 로그)에 순서대로 기록해요. Binlog 이벤트와 형식이 궁금하면 Binlog를 다룬 글을 참고하세요.

CDC 커넥터는 마치 MySQL 복제 서버인 것처럼 접속해서 이 Binlog를 받아 읽어요. 그리고 변경 하나하나를 이벤트로 바꿔 Kafka에 보냅니다.

MySQLUPDATE 실행BinlogROW 형식으로 기록Debezium Connector변경을 이벤트로 변환Kafka Topicshop.shop.users기록읽기발행검색분석캐시
DB는 그대로 두고, 변경 기록을 읽어 Kafka로 흘려보냅니다.

특히 ROW 형식의 Binlog는 SQL 문장이 아니라 바뀐 행을 기록하고, UPDATE라면 바뀌기 전 값과 바뀐 뒤 값을 함께 남겨요. 그래서 CDC는 삭제와 이전 값까지 그대로 전달할 수 있습니다.

실습 준비

이 글의 명령은 모두 실습 코드의 mysql-cdc-kafka/ 디렉터리에서 실행합니다.

  • compose.yaml: MySQL, Kafka, Debezium Kafka Connect 정의
  • setup.sql: 회원 테이블과 Debezium 계정 생성
  • connector.json: Debezium MySQL 커넥터 설정

Debezium은 가장 널리 쓰이는 오픈소스 CDC 도구이고, Kafka Connect 위에서 커넥터로 동작해요. compose.yaml은 mysql(mysql:8.4), kafka(apache/kafka:3.9.1), connect(quay.io/debezium/connect:3.1) 세 서비스를 정의하고 있어요.

MySQL에 넣을 setup.sql은 데이터베이스 shop에 회원 테이블 users(id, name, level, status)를 만들고 첫 번째 회원 Hong을 넣어요.

그리고 Debezium이 쓸 계정 debezium(비밀번호 dbz)을 만들어서 다음 권한을 줍니다.

  • SELECT: 처음 연결할 때 테이블의 현재 데이터를 읽어요.
  • RELOAD: 처음 읽는 동안 일관된 시점을 잡아요.
  • SHOW DATABASES: 어떤 데이터베이스가 있는지 확인해요.
  • REPLICATION SLAVE, REPLICATION CLIENT: 복제 서버처럼 Binlog를 받아 읽어요.

이제 컨테이너를 띄우고 SQL을 실행해 볼게요. 비밀번호 경고가 출력되지 않도록 MYSQL_PWD 환경 변수로 넘기는데, 이 방식은 실습에서만 써 주세요.

docker compose up -d
docker compose exec -T -e MYSQL_PWD=root mysql mysql -uroot < setup.sql
docker compose exec -e MYSQL_PWD=root mysql mysql -uroot -e "SHOW VARIABLES WHERE Variable_name IN ('log_bin', 'binlog_format', 'binlog_row_image');"

+------------------+-------+
| Variable_name    | Value |
+------------------+-------+
| binlog_format    | ROW   |
| binlog_row_image | FULL  |
| log_bin          | ON    |
+------------------+-------+

출력을 보면 MySQL 8.4는 기본 설정만으로 CDC 조건을 다 갖추고 있어요.

  • log_bin: Binlog가 켜져 있어야 해요.
  • binlog_format: 바뀐 행을 기록하는 ROW여야 해요.
  • binlog_row_image: 바뀐 컬럼뿐 아니라 행 전체를 남기는 FULL이어야 해요.

실습 1: 커넥터 등록

Kafka Connect에 등록할 커넥터 설정은 connector.json이에요.

{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "topic.prefix": "shop",
    "database.include.list": "shop",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.shop"
  }
}
  • database.hostname, database.port, database.user, database.password: 접속할 MySQL과 앞에서 만든 Debezium 계정이에요.
  • database.server.id: 복제 서버로 접속할 때 쓸 ID예요. MySQL 서버들과 겹치지 않는 값이어야 해요.
  • topic.prefix: Kafka 토픽 이름의 앞부분이에요. 토픽은 접두사.데이터베이스.테이블 형태로 만들어집니다.
  • database.include.list: 변경을 가져올 데이터베이스예요.
  • schema.history.internal.*: 테이블 구조의 변경 이력을 저장할 Kafka 토픽이에요.

Kafka Connect가 뜨는 데는 몇십 초가 걸려요. 준비되면 커넥터를 등록하고 상태를 확인해 볼게요.

curl -s -X POST -H "Content-Type: application/json" --data @connector.json localhost:8083/connectors > /dev/null
curl -s localhost:8083/connectors/shop-connector/status | jq

{
  "name": "shop-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "172.25.0.4:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "172.25.0.4:8083"
    }
  ],
  "type": "source"
}

커넥터와 태스크가 모두 RUNNING이면, users 테이블의 변경은 토픽 shop.shop.users로 들어가요.

실습 2: UPDATE 한 줄이 이벤트가 되기까지

이제 회원 등급을 바꾸고, 토픽을 처음부터 읽어 볼게요.

docker compose exec -e MYSQL_PWD=root mysql mysql -uroot -e "UPDATE shop.users SET level = 2 WHERE id = 1;"
docker compose exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 \
  --topic shop.shop.users --from-beginning --max-messages 2 | jq -c '{op, before, after, file: .source.file, pos: .source.pos}'

{"op":"r","before":null,"after":{"id":1,"name":"Hong","level":1,"status":"active"},"file":"binlog.000002","pos":1504}
{"op":"u","before":{"id":1,"name":"Hong","level":1,"status":"active"},"after":{"id":1,"name":"Hong","level":2,"status":"active"},"file":"binlog.000002","pos":1726}
Processed a total of 2 messages
  • op: 변경 종류예요. r은 스냅숏으로 읽은 행, u는 UPDATE로 바뀐 행이에요.
  • before, after: 바뀌기 전과 뒤의 행이에요. 스냅숏은 이전 값이 없으니 before가 null이죠.
  • file, pos: 이 이벤트가 Binlog의 어느 위치에 해당하는지 알려 줘요(이벤트의 source.file, source.pos).

첫 줄은 커넥터가 처음 연결할 때 테이블의 현재 데이터를 읽은 스냅숏이에요. UPDATE를 하기 전부터 토픽에 들어 있던 이벤트입니다.

두 번째 줄이 방금 실행한 UPDATE예요. op는 u이고, before에는 등급 1, after에는 등급 2가 들어 있죠.

여기서 눈여겨볼 건 이 이벤트를 만드는 데 애플리케이션 코드가 전혀 필요하지 않았다는 점이에요.

정리

범위: MySQL 8.4, Kafka 3.9.1(apache/kafka:3.9.1), Debezium 3.1.3.Final(quay.io/debezium/connect:3.1), 테이블 shop.users 하나.

  • MySQL 8.4는 기본 설정에서 log_bin이 ON, binlog_format이 ROW, binlog_row_image가 FULL이었습니다. CDC를 붙이기 전에 이 세 값을 확인해 주세요.
  • UPDATE 한 줄은 애플리케이션 코드 없이 op가 u인 이벤트 하나가 됐고, before에 등급 1, after에 등급 2가 담겼습니다.
  • 처음 연결하면 기존 행이 op가 r인 스냅숏 이벤트로 먼저 나왔습니다. 테이블이 크면 운영 DB에 처음 연결하기 전에 스냅숏 방식과 시간을 정해 두세요.
  • 이 글에 싣지 않은 추가 측정에서는, 커넥터가 멈춘 사이 Binlog를 지우자 커넥터가 Error code: 1236으로 멈췄고 스냅숏부터 다시 받아야 했습니다. Binlog 보관 기간(MySQL 8.4 기본 30일)은 커넥터가 멈춰 있을 수 있는 가장 긴 시간보다 넉넉하게 잡아 주세요.
내 링크 만들기