1. Debezium으로 MySQL CDC 구현하기 (Docker Compose)

Kafka와 Debezium을 사용해 MySQL의 변경 사항을 캡처하는 예제예요.

version: '3'
services:
  # MySQL 데이터베이스
  mysql:
    image: mysql:8.0
    ports:
      - 3306:3306
    environment:
      - MYSQL_ROOT_PASSWORD=debezium
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
      - MYSQL_DATABASE=inventory
    command: --server-id=1 --log-bin=mysql-bin --binlog-format=ROW

  # Zookeeper (Kafka 클러스터 관리)
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    ports:
      - 2181:2181
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181

  # Kafka 브로커
  kafka:
    image: confluentinc/cp-kafka:7.3.0
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1

  # Kafka Connect (Debezium 포함)
  connect:
    image: debezium/connect:2.3
    ports:
      - 8083:8083
    depends_on:
      - kafka
      - mysql
    environment:
      - BOOTSTRAP_SERVERS=kafka:9092
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=connect_configs
      - OFFSET_STORAGE_TOPIC=connect_offsets
      - STATUS_STORAGE_TOPIC=connect_statuses

위 Docker Compose 설정으로 환경을 구성한 후, Debezium 커넥터를 등록하는 API 호출:

curl -X POST \
  http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "inventory-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",
        "database.server.name": "dbserver1",
        "database.include.list": "inventory",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "include.schema.changes": "true"
    }
}'

이렇게 설정하면 MySQL의 모든 변경 사항이 Kafka 토픽으로 스트리밍됩니다. 토픽 이름은 dbserver1.inventory.{table_name} 형식으로 생성돼요.

2. 트리거 기반 CDC 구현 (PostgreSQL)

PostgreSQL에서 트리거를 사용해 간단한 CDC를 구현하는 예제예요.

-- CDC 메타데이터 테이블 생성
CREATE TABLE cdc_events (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100) NOT NULL,
    operation CHAR(1) NOT NULL, -- I: Insert, U: Update, D: Delete
    record_id BIGINT NOT NULL,
    old_data JSONB,
    new_data JSONB,
    changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 트리거 함수 생성
CREATE OR REPLACE FUNCTION process_cdc()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO cdc_events(table_name, operation, record_id, new_data)
        VALUES (TG_TABLE_NAME, 'I', NEW.id, row_to_json(NEW));
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO cdc_events(table_name, operation, record_id, old_data, new_data)
        VALUES (TG_TABLE_NAME, 'U', NEW.id, row_to_json(OLD), row_to_json(NEW));
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO cdc_events(table_name, operation, record_id, old_data)
        VALUES (TG_TABLE_NAME, 'D', OLD.id, row_to_json(OLD));
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- 사용자 테이블에 트리거 적용 예시
CREATE TRIGGER users_cdc_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW EXECUTE FUNCTION process_cdc();

이 방식은 구현이 간단하지만, 테이블마다 트리거를 설정해야 하고 DB 성능에 영향을 줄 수 있어요. 소규모 시스템에 적합한 방식이죠.

3. Spring Boot로 구현하는 쿼리 기반 CDC

Spring Boot 애플리케이션에서 주기적으로 DB를 폴링하여 변경 사항을 감지하는 예제예요.

@Service
public class CdcPollingService {

    private final JdbcTemplate jdbcTemplate;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    
    // 마지막 폴링 시간 저장
    private Timestamp lastPollingTime = new Timestamp(System.currentTimeMillis());
    
    @Autowired
    public CdcPollingService(JdbcTemplate jdbcTemplate, 
                            KafkaTemplate<String, String> kafkaTemplate,
                            ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }
    
    @Scheduled(fixedRate = 5000) // 5초마다 실행
    public void pollChanges() {
        Timestamp currentTime = new Timestamp(System.currentTimeMillis());
        
        // 변경된 사용자 데이터 조회 (last_updated 컬럼 필요)
        String sql = "SELECT * FROM users WHERE last_updated BETWEEN ? AND ?";
        List<Map<String, Object>> changedUsers = jdbcTemplate.queryForList(
            sql, lastPollingTime, currentTime);
        
        // 변경 데이터를 Kafka로 전송
        for (Map<String, Object> user : changedUsers) {
            try {
                String userJson = objectMapper.writeValueAsString(user);
                kafkaTemplate.send("user-changes", userJson);
            } catch (JsonProcessingException e) {
                // 에러 처리
            }
        }
        
        // 마지막 폴링 시간 업데이트
        lastPollingTime = currentTime;
    }
}

이 방식은 구현이 쉽지만 실시간성이 떨어지고, 삭제된 레코드는 감지하기 어려워요. 또한 테이블에 last_updated 같은 타임스탬프 컬럼이 필요해요.

4. AWS DMS를 사용한 CDC 구성 (AWS CLI)

AWS CLI를 사용해 DMS 복제 인스턴스와 태스크를 생성하는 예제예요.

# 복제 인스턴스 생성
aws dms create-replication-instance \
    --replication-instance-identifier my-cdc-instance \
    --replication-instance-class dms.t3.medium \
    --allocated-storage 50 \
    --vpc-security-group-ids sg-12345678 \
    --replication-subnet-group-id default-vpc-subnet-group

# 소스 엔드포인트 생성 (MySQL)
aws dms create-endpoint \
    --endpoint-identifier mysql-source \
    --endpoint-type source \
    --engine-name mysql \
    --username admin \
    --password mypassword \
    --server-name mysql-instance.xyz.us-east-1.rds.amazonaws.com \
    --port 3306 \
    --database-name mydb

# 대상 엔드포인트 생성 (Kinesis)
aws dms create-endpoint \
    --endpoint-identifier kinesis-target \
    --endpoint-type target \
    --engine-name kinesis \
    --kinesis-settings '{"StreamArn":"arn:aws:kinesis:us-east-1:123456789012:stream/cdc-stream","MessageFormat":"json"}'

# 복제 태스크 생성 (CDC 모드)
aws dms create-replication-task \
    --replication-task-identifier mysql-to-kinesis-cdc \
    --source-endpoint-arn arn:aws:dms:us-east-1:123456789012:endpoint:mysql-source \
    --target-endpoint-arn arn:aws:dms:us-east-1:123456789012:endpoint:kinesis-target \
    --replication-instance-arn arn:aws:dms:us-east-1:123456789012:rep:my-cdc-instance \
    --migration-type cdc \
    --table-mappings file://table-mappings.json \
    --replication-task-settings file://task-settings.json

AWS DMS는 관리형 서비스라 인프라 관리 부담이 적고, 다양한 소스/타겟을 지원해요. 클라우드 환경에서 CDC를 구현할 때 좋은 선택이 될 수 있어요.