debezium setting

📡 Debezium + MySQL 설정 가이드 (Standalone 기준)

💡 실습 환경: 단일 서버에서 Kafka Connect Standalone 모드로 Debezium MySQL Connector 실행

1️⃣ MySQL 서버 설정

Debezium이 MySQL의 변경 데이터를 읽으려면 다음 옵션이 필요합니다.
Windows: my.ini / Linux: my.cnf

# Binlog 활성화
log_bin=mysql-bin

# GTID 활성화
gtid_mode=ON
enforce_gtid_consistency=ON

• Binlog: 변경 데이터 추적용
• GTID (Global Transaction ID): MySQL의 각 트랜잭션에 부여되는 전역 고유 ID
• enforce_gtid_consistency: GTID 모드에서 모든 트랜잭션이 GTID를 지원하도록 강제하는 옵션
  🎈 Debezium은 Binlog(변경 로그)를 읽으면서 트랜잭션 단위로 정확하게 데이터 변경을 추적해야 합니다.
• 만약 GTID가 일관되지 않으면:

• 서버 재시작, 스냅샷 후 이어서 읽을 때 위치가 꼬일 수 있음
• Debezium이 어느 트랜잭션까지 읽었는지 정확히 판단할 수 없음 즉, GTID + enforce_gtid_consistency = 항상 추적 가능한 트랜잭션 로그 확보

• GTID 없으면 Debezium이 “어디까지 읽었는지”를 정확히 기억하기 어렵습니다.

2️⃣ MySQL 계정 및 권한 설정

Debezium이 Binlog와 데이터를 읽을 수 있도록 전용 계정을 생성하고 권한 부여:

-- 계정 생성 (이미 존재하면 생략 가능)
CREATE USER 'example_user'@'%' IDENTIFIED BY 'example_password';

-- 권한 부여
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'example_user'@'%';

-- 권한 확인
SHOW GRANTS FOR 'example_user'@'%';

권한 및 역할

권한	역할
SELECT	일반 테이블 조회 가능. 초기 스냅샷(snapshots) 시 테이블 데이터 읽기 필요
REPLICATION SLAVE	Binlog 읽기 권한. MySQL 복제(Master → Slave)용 권한과 동일. Debezium이 실시간 변경 추적 시 필요
REPLICATION CLIENT	복제 상태 확인용. Debezium이 Binlog 위치, GTID 등을 확인할 때 사용

---

왜 모두 필요한가?

1. Debezium이 초기 스냅샷을 가져올 때 → SELECT
2. 실시간 Binlog 구독할 때 → REPLICATION SLAVE
3. Binlog 위치, 서버 상태 확인 → REPLICATION CLIENT
   💡 순서 중요: MySQL 설정 → 서버 재시작 → 계정/권한 생성

🔹 MySQL 설정 확인용 쿼리

-- Binlog 활성화 여부
SHOW VARIABLES LIKE 'log_bin';

-- Binlog 포맷 (ROW, STATEMENT, MIXED)
SHOW VARIABLES LIKE 'binlog_format';  
| STATEMENT | SQL 문장 그대로를 Binlog에 기록 |
| ROW | 실제 변경된 행 데이터를 Binlog에 기록 |
| MIXED | 상황에 따라 STATEMENT 또는 ROW를 자동 선택 |
-- 서버 ID
SHOW VARIABLES LIKE 'server_id';

-- ROW 기반 Binlog에서 이미지(전체/부분) 설정
SHOW VARIABLES LIKE 'binlog_row_image';

-- Binlog 파일 경로
SHOW VARIABLES LIKE 'log_bin_basename';

-- GTID 활성화 여부
SHOW VARIABLES LIKE 'gtid_mode';

-- GTID 일관성 강제 여부
SHOW VARIABLES LIKE 'enforce_gtid_consistency';

MySQL 변수 확인

변수	의미
log_bin	Binlog 활성화 여부
binlog_format	Binlog 기록 방식 (Debezium은 ROW 권장)
server_id	MySQL 서버 고유 ID (Debezium connector에서 필요)
binlog_row_image	ROW 포맷에서 얼마나 많은 컬럼을 기록할지 (FULL 권장)
log_bin_basename	Binlog 파일 기본 경로
gtid_mode	GTID 사용 여부
enforce_gtid_consistency	GTID 일관성 강제 여부

3️⃣ Kafka Connect Standalone 설정

파일: config/connect-standalone.properties

plugin.path=C:/kafka_2.13-4.1.1/plugins

• Debezium Connector가 위치한 폴더를 지정
• Standalone 모드에서 Connector 자동 로드

4️⃣ Debezium Connector 설정 (Properties 파일)

파일: config/connect-mysql-source.properties

name=example_connector
connector.class=io.debezium.connector.mysql.MySqlConnector
connection.url=jdbc:mysql://localhost:3306/example_db?serverTimezone=Asia/Seoul&useUnicode=true&characterEncoding=UTF-8

# DB 연결
database.hostname=localhost
database.port=3306
database.user=example_user
database.password=password

# Server ID / Prefix
database.server.id=12345
database.server.name=example_server

# History Topic
database.history.kafka.bootstrap.servers=localhost:9092
database.history.kafka.topic=example_schema_history

# 감시할 테이블
database.include.list=example_db
table.include.list=example_table

# Kafka 토픽 prefix
topic.prefix=example_topic

# 옵션
include.schema.changes=true
database.serverTimezone=Asia/Seoul

💡 실제 단일 서버 실습용이므로 간단하게 설정

5️⃣ Kafka Connect 실행 (Standalone)

bin\windows\connect-standalone.bat config\connect-standalone.properties config\connect-mysql-source.properties

• Standalone 모드에서 properties 파일을 바로 읽어 실행
• 설치 단계에서 배치한 Debezium Connector와 MySQL JDBC를 자동 로드

6️⃣ Spring Boot Kafka 설정 (application.yml 예시)

kafka:
  bootstrap-servers: localhost:9092

  consumer:
    group-id: mineping-group
    auto-offset-reset: earliest
    key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
    value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
    properties:
      isolation.level: read_committed
      spring.json.trusted.packages: com.minePing.BackEnd.dto, com.minePing.BackEnd.event
      spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
      spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

  producer:
    key-serializer: org.apache.kafka.common.serialization.StringSerializer
    value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    properties:
      enable.idempotence: true

🔹 설명

1. Consumer 설정 group-id: Kafka 컨슈머 그룹 지정, 오프셋 공유를 통해 여러 인스턴스에서 메시지 처리 가능
   auto-offset-reset: earliest: 새로운 그룹이면 토픽 처음부터 읽음
   ErrorHandlingDeserializer + JsonDeserializer: JSON 메시지 역직렬화, 오류 발생 시 안전하게 처리
   spring.json.trusted.packages: JSON 역직렬화 가능한 패키지 제한, 보안용
   isolation.level: read_committed: Debezium에서 트랜잭션 단위로 커밋된 데이터만 읽기
2. Producer 설정 JSON 직렬화, 문자열 키 사용
   enable.idempotence: true: 중복 메시지 방지, 멱등성 적용(Exactly-Once Semantics)

💡 요약:

🔹 동작 흐름

MySQL → Debezium (Kafka Connect) → Kafka 토픽 → Spring Boot Consumer/Producer

⚠️ 포인트

• Standalone 모드: 단일 서버/단일 프로세스 실행
• plugin.path 필수 설정
• JSON 파일은 REST API용으로 Standalone 환경에서는 필요 없음
• Properties 파일 하나로 Connector 구성 및 실행 가능
• Spring Boot 애플리케이션에서는 Kafka consumer/producer 설정 필요

• JSON 직렬화/역직렬화, trusted packages, 오프셋 관리 등