CDC 심화

학습 목표

Debezium 커넥터 토폴로지(Connect worker·slot·topic·SMT)를 설명할 수 있다.

Schema registry와 스키마 진화(backward/forward compatibility)를 설계할 수 있다.

ordering·exactly-once 보장 범위와 한계를 DB·sink 관점에서 구분할 수 있다.

DLQ·재처리·slot lag 복구 런북을 작성할 수 있다.

snapshot·incremental 전환·DDL 장애를 운영할 수 있다.

문제 상황

• ALTER TABLE products ADD COLUMN — Debezium Avro deserialize 실패 · consumer 전부 stop
• connector 재시작 — offset earliest 실수 — 과거 6개월 replay · 중복 적재
• pg_replication_slots lag 80GB — disk 90% · vacuum blocked
• exactly-once 기대 — Kafka EOS 있는데 Elasticsearch 중복 문서
• poison row — 한 메시지 무한 retry — lag 폭증
• 스키마 v3 consumer · v1 topic 메시지 — 필드 누락 NPE

앞서 CDC — WAL·outbox·멱등 개요; Architecture — Kafka·EDA 조립. 이번 편은 Debezium 운영 심화이자 Database 시리즈 마무리다.

1. Debezium 토폴로지

Kafka Connect — connector 플러그인이 DB 스트림을 topic으로 변환.

PostgreSQL (wal_level=logical)
    → replication slot (pgoutput)
    → Debezium PostgreSQL connector (Connect worker)
    → schema history topic + data topics
    → Schema Registry (optional Avro)
    → consumers (search, cache, warehouse)

구성요소	역할
Source connector	WAL/binlog 읽기 · envelope 생성
Replication slot	LSN 커서 — 미소비 WAL 보존
Offset storage	Connect 내부 topic — 재시작 위치
Schema history	DDL 이력 — deserializer 동기화
SMT	Single Message Transform — 필터·라우팅·outbox router

// code/debezium-postgres-connector.example
{
  "name": "orders-pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",
    "publication.name": "dbz_orders_pub",
    "table.include.list": "public.orders,public.outbox",
    "tombstones.on.delete": "false",
    "snapshot.mode": "initial"
  }
}

-- publication (PostgreSQL 10+)
CREATE PUBLICATION dbz_orders_pub FOR TABLE orders, outbox;

SELECT slot_name, confirmed_flush_lsn, pg_size_pretty(
         pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)
       ) AS lag
FROM pg_replication_slots
WHERE slot_name = 'debezium_orders';

토픽
shop.public.orders	row change stream
schema-changelog.shop	DDL history
Heartbeat	idle DB slot advance (설정 시)

• 앞서 logical replication — 같은 slot 메커니즘
• Connect cluster — connector task 분산 — 단일 PG slot per connector

2. Schema registry와 스키마 진화

CDC payload — 테이블 스키마 결합 — DDL 시 깨짐 위험.

	JSON envelope	Avro + Registry
스키마	connector 내장	별도 ID 참조
진화	유연 · 검증 약함	compatibility 정책
크기	큼	compact
운영	단순	registry HA 필요

// code/schema-compatibility.example
// Registry: BACKWARD — new consumer reads old + new messages
// FORWARD   — old consumer reads new messages (additive only)
// FULL      — both directions

DDL	영향
ADD COLUMN nullable	보통 backward OK
DROP COLUMN	forward 주의 · breaking
RENAME	breaking — 새 topic·마이그레이션
TYPE change	connector 재스냅샷·수동 조정

ALTER ADD COLUMN  →  schema v2 registered  →  consumer with v2 reader OK (BACKWARD)
ALTER DROP COLUMN →  old messages still have field  →  plan deprecation window

운영
Schema history topic	보존 무기한 — 삭제 금지
Consumer upgrade	새 스키마 먼저 배포 (backward)
Breaking DDL	blue/green table · dual write 기간

• Architecture 이벤트 계약 — 서비스 API; 여기서는 DB DDL → Avro 진화

3. Ordering과 exactly-once

DB commit 순 ≈ LSN 순 — 단일 테이블 단일 partition topic 내.

보장	범위	현실
DB commit order	source 테이블 stream	LSN 단조
Kafka partition	key 동일 → 순서	hash 설계
At-least-once	Connect 기본	중복 가능
EOS	Kafka transaction	broker 내부 — sink 별개

Debezium → Kafka (at-least-once default)
Consumer → Elasticsearch (no EOS unless idempotent upsert by _id)

exactly-once 착각	실제
"Kafka EOS 켰다"	producer·streams 범위
"한 번만 처리"	sink 멱등 필수 (20편)
"전역 순서"	불가 — partition key

# code/idempotent-sink.example — DB primary key as document id
def index_product(change: dict):
    doc_id = change["after"]["id"]
    op = change["op"]
    if op == "d":
        es.delete(index="products", id=doc_id, ignore=[404])
    else:
        es.index(index="products", id=doc_id, document=change["after"])

패턴
Idempotent sink	PK = document id · UPSERT
processed_offsets	(topic, partition, offset) UNIQUE
Outbox relay	published_at after ack
Kafka EOS	Streams 내부 상태·changelog

• Cross-table 순서 — 보장 없음 — saga·도메인 설계
• Observability — consumer lag · duplicate rate 메트릭

4. DLQ와 재처리

Poison message — deserialize 실패 · 비즈니스 예외 — 무한 retry 방지.

단계
Retry	transient 오류 — exponential backoff
DLQ	N회 실패 → dead-letter topic
Alert	DLQ rate · lag spike
Fix	스키마·코드·데이터 수정
Replay	DLQ → 원 topic · 시간 범위 replay

// code/connect-error-tolerance.example
{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "dlq.shop.orders",
  "errors.deadletterqueue.context.headers.enable": "true",
  "errors.retry.timeout": "60000",
  "errors.retry.delay.max.ms": "10000"
}

# code/dlq-replay.example
# 1. Fix consumer / schema
# 2. Replay DLQ messages to target topic with same key (preserve order)
for msg in dlq_consumer:
    producer.send("shop.public.orders", key=msg.key, value=msg.value)

Connect 장애	대응
Connector FAILED	로그 · slot 활성 여부
Deserialization	schema history · registry 버전
Slot inactive	재시작 — offset 이어 읽기
Offset reset 실수	snapshot 재실행 — sink 멱등 필수

• 앞서 멱등 consumer — replay 전제
• 수동 skip — offset commit — 감사 로그 남기기

5. Snapshot과 incremental

snapshot.mode
initial	첫 기동 full read + stream
never	stream 만 — 기존 데이터 없음
when_needed	offset 없을 때 만
no_data	스키마 만

Phase 1: consistent snapshot (SELECT ... EXPORT_SNAPSHOT)
Phase 2: stream from LSN captured at snapshot start
         → no gap if connector coordinates correctly

주의
대형 테이블	snapshot 부하 — off-peak · signal table
Replica from	읽기 부하 분리 (설정 시)
Resume	interrupt 후 이어 읽기

6. 운영 런북

시나리오	조치
Slot lag 증가	consumer lag · connector 상태 · down sink
Disk 압박	lag 해소 전 slot 삭제 금지 — WAL 폭주
DDL 배포	registry compatibility 확인 · consumer 순서
Connector 업그레이드	staging 검증 · offset 백업
재동기화	새 connector + snapshot · sink truncate·멱등

-- code/slot-monitor.sql
SELECT slot_name,
       active,
       confirmed_flush_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag_bytes
FROM pg_replication_slots;

-- emergency: only after stopping connector and accepting data re-sync
-- SELECT pg_drop_replication_slot('debezium_orders');

메트릭	임계
replication_slot_lag	디스크 대비
debezium_metrics MilliSecondsBehindSource	staleness
dlq_topic_rate	> 0 지속
schema incompatible count	0 목표

• 앞서 vacuum — slot lag → WAL 쌓임 → disk full
• ELK·파이프라인 — 9_ELKElasticCloud 심화

7. Outbox router (심화)

Debezium Outbox Event Router SMT — outbox 테이블 → 도메인 topic.

outbox row (aggregate_type=Order, event_type=OrderPlaced)
    → SMT routes to topic OrderEvents
    → payload from outbox.payload column

	Raw CDC on outbox	Outbox Router
Topic	shop.public.outbox	OrderEvents
Payload	full row	JSON column
도메인	consumer 변환	router 변환

• 20편 outbox + 이번 SMT = 운영 표준 조합

8. 시리즈 마무리

Database 24편 — 저장 → 쿼리 → 트랜잭션 → 확장 → 실무 → 운영 축 완료.

Part	핵심
0~1	관계형·SQL·정규화·쿼리 처리
2	디스크·인덱스·실행 계획
3	ACID·동시성·WAL·MVCC
4	복제·파티션·분산·NoSQL
5	Pool·캐시·CDC 개요
6	슬로우 쿼리·vacuum·파티션 ops·CDC 심화

Application  →  Connection pool  →  PostgreSQL
                    ↓                      ↓ WAL
                 Cache aside          Debezium → projections

연계 시리즈
Observability	pg_stat·slot lag·consumer lag
Architecture	MSA·EDA·CQRS 조립
ELK	CDC → ingest 파이프라인

9. 정리

• Debezium — slot·publication·Connect offset·schema history
• Schema registry — DDL 진화 — compatibility 정책
• Exactly-once — sink 멱등까지 포함해야 의미 있음
• DLQ·replay — poison 격리 · 멱등 전제
• Slot lag — DB disk 위험 — vacuum·consumer 연동 모니터
• 시리즈 완결 — 엔진 내부 이해 + 운영 루틴

다음에 다룰 것

• 9_ELKElasticCloud — CDC·로그 ingest 파이프라인 심화
• Observability — DB·Kafka 메트릭 대시보드

---

해당 내용은 Database System Concepts, 7/E (Avraham Silberschatz, Henry F. Korth, S. Sudarshan), Designing Data-Intensive Applications (Martin Kleppmann), Debezium · Confluent Schema Registry 공식 문서 의 내용을 기반으로 합니다.