8장: CDC(Change Data Capture)

{
  "before": {
    "id": 1001,
    "status": "pending",
    "amount": 50000
  },
  "after": {
    "id": 1001,
    "status": "completed",
    "amount": 50000
  },
  "source": {
    "connector": "postgresql",
    "db": "shop",
    "schema": "public",
    "table": "orders",
    "txId": 12345,
    "lsn": 234567890
  },
  "op": "u",
  "ts_ms": 1710648000000
}

-- postgresql.conf에서 logical replication 활성화
-- wal_level = logical
-- max_replication_slots = 4
-- max_wal_senders = 4
 
-- 리플리케이션 사용자 생성
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'secret';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
 
-- publication 생성 (캡처할 테이블 지정)
CREATE PUBLICATION dbz_publication FOR TABLE orders, payments;

{
  "name": "pg-orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "shop",
    "database.server.name": "shop-db",
    "schema.include.list": "public",
    "table.include.list": "public.orders,public.payments",
    "topic.prefix": "cdc",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication",
    "slot.name": "debezium_slot",
    "snapshot.mode": "initial",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

{
  "name": "mysql-orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "secret",
    "database.server.id": "1",
    "database.server.name": "shop-mysql",
    "database.include.list": "shop",
    "table.include.list": "shop.orders,shop.payments",
    "topic.prefix": "cdc",
    "include.schema.changes": "true",
    "snapshot.mode": "initial"
  }
}

-- PostgreSQL CDC 소스 테이블
CREATE TABLE orders_source (
    id INT,
    user_id VARCHAR,
    amount DECIMAL(10, 2),
    status VARCHAR,
    created_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = 'postgres',
    'port' = '5432',
    'username' = 'debezium',
    'password' = 'secret',
    'database-name' = 'shop',
    'schema-name' = 'public',
    'table-name' = 'orders',
    'slot.name' = 'flink_cdc_slot',
    'decoding.plugin.name' = 'pgoutput'
);
 
-- Elasticsearch 싱크 테이블
CREATE TABLE orders_search (
    id INT,
    user_id VARCHAR,
    amount DECIMAL(10, 2),
    status VARCHAR,
    created_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://elasticsearch:9200',
    'index' = 'orders'
);
 
-- CDC 데이터를 실시간으로 Elasticsearch에 동기화
INSERT INTO orders_search
SELECT * FROM orders_source;

1. 데이터베이스에 주문 저장 ← 성공
2. Kafka에 이벤트 발행 ← 실패 (네트워크 오류)
→ DB에는 주문이 있지만 이벤트는 발행되지 않음 (불일치)

-- 아웃박스 테이블 생성
CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

# 단일 트랜잭션으로 주문과 이벤트를 함께 저장
def place_order(conn, order):
    with conn.begin():
        # 1. 주문 저장
        conn.execute(
            "INSERT INTO orders (id, user_id, amount, status) "
            "VALUES (%s, %s, %s, %s)",
            (order.id, order.user_id, order.amount, "pending")
        )
        # 2. 아웃박스에 이벤트 기록 (같은 트랜잭션)
        conn.execute(
            "INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload) "
            "VALUES (%s, %s, %s, %s)",
            ("Order", order.id, "OrderPlaced",
             json.dumps({"orderId": order.id, "amount": order.amount}))
        )
    # 트랜잭션 커밋 → 주문과 이벤트가 원자적으로 저장됨

{
  "transforms": "outbox",
  "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
  "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType",
  "transforms.outbox.route.by.field": "aggregate_type",
  "transforms.outbox.route.topic.replacement": "events.${routedByValue}"
}