MSK Connect + Debezium 이용하여 CDC 파이프라인 구축

현재 새롭게 구축하는 프로젝트에서 특정 DB 의 변경 이벤트를 실시간으로 CDC(캡처)해서 타 서비스 (Neo4j, worker) 등에 전달하는 CDC 파이프라인이 필요하다.

* CDC 도입 배경 : 기존의 배치 방식은 주기적인 쿼리 실행으로 DB 부하 및 실시간성이 떨어짐. CDC 는 DB 의 Binlog를 직접 읽어서 변경분만 캡처해 실시간성, 부하 최소화 등 장점이 뚜렷함

 

일반적으로 Debezium을 쓰려면 Kafka Connect 서버를 별도로 띄워야 한다.

하지만 이미 EKS 클러스터에 서비스들이 올라가 있는 상황에서 Kafka Connect 전용 인프라를 추가로 관리하는 건 부담이 된다 판단.

 

해결책으로 AWS MSK Connect — Kafka Connect를 완전 관리형으로 제공하는 서비스를 이용하려고 한다.

 

아키텍처는 아래와 같다.

                           db (MySQL)
                              ↓ binlog
MSK Connect (Debezium MySQL Connector)  ←  AWS 완전 관리
                              ↓ CDC 이벤트
      MSK Provisioned Cluster (dev-msk-cluster)
                              ↓
		  ┌───────────────────┬──────────────────┐
        worker Pod        Neo4j-worker Pod   DLT 토픽
       (Kafka Consumer)    (그래프 갱신)       (에러 이벤트)

 

1. MSK Provisioned Cluster

- 서버리스를 사용하지 않고 프로비저닝을 사용한 이유는 모니터링(PER_TOPIC_PER_BROKER) 가 필요했기 때문.

  ㄴ  DLT 발생 시 모니터링을 정밀히 제어하기 위함

- 인증방식은 pod Identity 를 사용. EKS pod -> MSK 연결시 AWS SDK가 자동으로 IAM 서명 처리 (보안성, 편의성)

# MSK Provisioned Cluster
resource "aws_msk_cluster" "riman_one_dev" {
...

  broker_node_group_info {
    instance_type = "kafka.t3.small"

  # IAM 인증 (Pod Identity 연동)
  client_authentication {
    sasl {
      iam = true
    }
  }

  # topic별 메트릭 수집 (MessagesInPerSec, BytesInPerSec 등 → Grafana DLT 알람)
  enhanced_monitoring = "PER_TOPIC_PER_BROKER"

  # CloudWatch 모니터링
  logging_info {
    broker_logs {
      cloudwatch_logs {
        enabled   = true
        log_group = aws_cloudwatch_log_group.msk.name
      }
    }
  }

}

 

2. Debezium 설치

- 서버 없이 S3에 플러그인을 넣어 사용

- MSK Connect는 플러그인 ZIP을 S3에 업로드하여, Kafka Connect 서버를 직접 띄울 필요가 없다.

- ZIP 파일은 apply 전에 미리 업로드 해놓는다.

..S3 를 만들고,,

# MSK Connect Custom Plugin 등록
  resource "aws_mskconnect_custom_plugin" "debezium" {
    name         = "dev-debezium-plugin"
    content_type = "ZIP"
    location {
      s3 {
        bucket_arn = module.s3_debezium_plugins.s3_bucket_arn
        file_key   = "debezium-plugin.zip"
      }
    }
  }

 

3. Debezium 커넥터 핵심 설정

- 기존 커넥터가 있으면 server.id 는 겹치지 않게 진행

resource "aws_mskconnect_connector" "debezium_dev_mysql" {
    connector_configuration = {
      "connector.class"   = "io.debezium.connector.mysql.MySqlConnector"
      "database.hostname" = "development-domain-db.cluster-xxx.rds.amazonaws.com"

      # 기존 커넥터(server.id=10005)와 binlog 충돌 방지
      "database.server.id" = "20001"

      # 캡처 대상 테이블 4개
      "table.include.list" = join(",", [
        "member.test-1",
        "member.test-2",
        "member.test-3",
        "member.test-4",
      ])

      # 기존 데이터 스냅샷 제외 - 이후 변경분만 캡처
      "snapshot.mode"         = "no_data"
      "snapshot.locking.mode" = "none"  # 테이블 락 방지
    }
  }

 

4. 토픽 자동생성 

- 아마 요청한 팀에서 원하는 파티션 개수를 요청할것. ex) 일반 토픽은 2개(처리량 고려) , dlt는 1개(중복X, 순서보장, 에러추적)

  # CDC Origin 토픽: 파티션 2개, RF 2
  "topic.creation.default.partitions"         = "2"
  "topic.creation.default.replication.factor" = "2"
  "topic.creation.enable"                     = "true"

  # DLT: 파티션 1개, RF 2 (-dlt 패턴 자동 감지)
  "topic.creation.dlt.include"            = ".*-dlt$"
  "topic.creation.dlt.partitions"         = "1"
  "topic.creation.dlt.replication.factor" = "2"

 

이외에 Kafka , Debezium 에 대한 설정, 옵션 등 너무 다양하다.

핵심이 아니라 제외했지만, 메시지를 입맛에 맞게 파싱하거나(Header 등), 에러 처리 등 너무나 많다.

 

이외는 학습을 진행하면서 AI 도움도 받고 하면 더 안전하고 정교한 인프라 구축이 가능하다.