Outbox 패턴과 CDC로 구현하는 마이크로서비스 데이터 동기화: Debezium 실전 가이드

// 위험한 Dual Write 패턴
@Transactional
public Order createOrder(OrderRequest request) {
    Order order = orderRepository.save(new Order(request));  // Step 1: DB 저장

    // Step 2: 메시지 브로커에 이벤트 발행
    kafkaTemplate.send("order-events", new OrderCreatedEvent(order));

    return order;
}

[서비스 코드]
    │
    ├── 비즈니스 데이터 INSERT ─────┐
    │                                │  같은 DB 트랜잭션
    └── Outbox 테이블 INSERT ───────┘
                                      │
                                      ▼
                              [Outbox Relay]
                              (CDC or Polling)
                                      │
                                      ▼
                              [메시지 브로커]
                              (Kafka, RabbitMQ)
                                      │
                                      ▼
                              [다른 마이크로서비스]

-- PostgreSQL Outbox 테이블 DDL
CREATE TABLE outbox_events (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type  VARCHAR(255) NOT NULL,   -- 이벤트 소속 도메인 (예: 'Order')
    aggregate_id    VARCHAR(255) NOT NULL,   -- 도메인 엔티티 ID (예: 주문 ID)
    event_type      VARCHAR(255) NOT NULL,   -- 이벤트 타입 (예: 'OrderCreated')
    payload         JSONB NOT NULL,          -- 이벤트 본문
    metadata        JSONB DEFAULT '{}',      -- 추가 메타데이터
    created_at      TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMP WITH TIME ZONE,-- Polling 방식용: 발행 완료 시각
    retry_count     INT DEFAULT 0,           -- Polling 방식용: 재시도 횟수
    status          VARCHAR(20) DEFAULT 'PENDING'  -- PENDING, PUBLISHED, FAILED
);

-- Polling 방식에서 미발행 이벤트 조회를 위한 인덱스
CREATE INDEX idx_outbox_status_created ON outbox_events(status, created_at)
    WHERE status = 'PENDING';

-- CDC 방식에서는 status/published_at 컬럼이 불필요하며,
-- 이벤트 발행 후 행을 삭제하는 방식을 사용할 수 있다

-- MySQL에서 트리거를 이용한 자동 Outbox 기록
DELIMITER //

CREATE TRIGGER after_order_insert
AFTER INSERT ON orders
FOR EACH ROW
BEGIN
    INSERT INTO outbox_events (
        id, aggregate_type, aggregate_id, event_type, payload, created_at
    ) VALUES (
        UUID(),
        'Order',
        NEW.order_id,
        'OrderCreated',
        JSON_OBJECT(
            'orderId', NEW.order_id,
            'userId', NEW.user_id,
            'totalAmount', NEW.total_amount,
            'status', NEW.status,
            'createdAt', NEW.created_at
        ),
        NOW()
    );
END //

CREATE TRIGGER after_order_update
AFTER UPDATE ON orders
FOR EACH ROW
BEGIN
    IF OLD.status != NEW.status THEN
        INSERT INTO outbox_events (
            id, aggregate_type, aggregate_id, event_type, payload, created_at
        ) VALUES (
            UUID(),
            'Order',
            NEW.order_id,
            CONCAT('OrderStatus', NEW.status),
            JSON_OBJECT(
                'orderId', NEW.order_id,
                'previousStatus', OLD.status,
                'newStatus', NEW.status,
                'updatedAt', NEW.updated_at
            ),
            NOW()
        );
    END IF;
END //

DELIMITER ;

# docker-compose.yml - Debezium + Kafka 전체 스택
version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: orderdb
      POSTGRES_USER: appuser
      POSTGRES_PASSWORD: secret
    command:
      - 'postgres'
      - '-c'
      - 'wal_level=logical' # CDC를 위한 필수 설정
      - '-c'
      - 'max_replication_slots=4'
      - '-c'
      - 'max_wal_senders=4'
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    depends_on:
      - zookeeper
    ports:
      - '9092:9092'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: debezium/connect:2.7
    depends_on:
      - kafka
      - postgres
    ports:
      - '8083:8083'
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: debezium-connect-group
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - '8080:8080'
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: debezium
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://connect:8083

volumes:
  postgres_data:

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "appuser",
    "database.password": "secret",
    "database.dbname": "orderdb",
    "topic.prefix": "order-service",
    "schema.include.list": "public",
    "table.include.list": "public.outbox_events",

    "tombstones.on.delete": "false",
    "slot.name": "order_outbox_slot",
    "plugin.name": "pgoutput",

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

    "transforms.outbox.table.expand.json.payload": "true",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    "heartbeat.interval.ms": "10000",
    "snapshot.mode": "initial"
  }
}

# 커넥터 등록
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @order-outbox-connector.json

# 커넥터 상태 확인
curl -s http://localhost:8083/connectors/order-outbox-connector/status | jq .

# 커넥터 재시작
curl -X POST http://localhost:8083/connectors/order-outbox-connector/restart

# 토픽 목록 확인 (Kafka UI 또는 CLI)
docker exec -it kafka kafka-topics --bootstrap-server localhost:9092 --list

// OutboxEvent.java - Outbox 엔티티
@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(name = "aggregate_type", nullable = false)
    private String aggregateType;

    @Column(name = "aggregate_id", nullable = false)
    private String aggregateId;

    @Column(name = "event_type", nullable = false)
    private String eventType;

    @Column(name = "payload", columnDefinition = "jsonb", nullable = false)
    private String payload;

    @Column(name = "created_at", nullable = false)
    private Instant createdAt;

    // 기본 생성자, getter, setter 생략

    public static OutboxEvent create(String aggregateType, String aggregateId,
                                     String eventType, Object payload) {
        OutboxEvent event = new OutboxEvent();
        event.aggregateType = aggregateType;
        event.aggregateId = aggregateId;
        event.eventType = eventType;
        event.payload = toJson(payload);
        event.createdAt = Instant.now();
        return event;
    }

    private static String toJson(Object obj) {
        try {
            return new ObjectMapper()
                .registerModule(new JavaTimeModule())
                .writeValueAsString(obj);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("JSON 직렬화 실패", e);
        }
    }
}

// OrderService.java - Outbox 패턴 적용 주문 서비스
@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;

    @Transactional  // 하나의 트랜잭션으로 비즈니스 데이터와 이벤트를 함께 저장
    public Order createOrder(OrderRequest request) {
        // 1. 비즈니스 로직 실행
        Order order = Order.builder()
            .userId(request.getUserId())
            .items(request.getItems())
            .totalAmount(calculateTotal(request.getItems()))
            .status(OrderStatus.CREATED)
            .build();

        Order savedOrder = orderRepository.save(order);

        // 2. Outbox 테이블에 이벤트 기록 (같은 트랜잭션)
        OrderCreatedPayload eventPayload = OrderCreatedPayload.builder()
            .orderId(savedOrder.getId().toString())
            .userId(savedOrder.getUserId())
            .totalAmount(savedOrder.getTotalAmount())
            .items(savedOrder.getItems())
            .createdAt(savedOrder.getCreatedAt())
            .build();

        OutboxEvent outboxEvent = OutboxEvent.create(
            "Order",                              // aggregate_type -> Kafka 토픽 결정
            savedOrder.getId().toString(),         // aggregate_id -> Kafka 파티션 키
            "OrderCreated",                        // event_type
            eventPayload                           // payload
        );

        outboxRepository.save(outboxEvent);

        return savedOrder;
    }

    @Transactional
    public Order cancelOrder(UUID orderId) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));

        if (order.getStatus() != OrderStatus.CREATED) {
            throw new IllegalStateException(
                "CREATED 상태의 주문만 취소할 수 있습니다. 현재 상태: " + order.getStatus()
            );
        }

        order.setStatus(OrderStatus.CANCELLED);
        Order savedOrder = orderRepository.save(order);

        OutboxEvent outboxEvent = OutboxEvent.create(
            "Order",
            savedOrder.getId().toString(),
            "OrderCancelled",
            Map.of(
                "orderId", savedOrder.getId().toString(),
                "reason", "사용자 요청",
                "cancelledAt", Instant.now().toString()
            )
        );

        outboxRepository.save(outboxEvent);

        return savedOrder;
    }
}

import json
import uuid
from datetime import datetime
from kafka import KafkaConsumer
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

# 멱등성 보장을 위한 처리 이력 테이블
# CREATE TABLE processed_events (
#     event_id UUID PRIMARY KEY,
#     processed_at TIMESTAMP NOT NULL DEFAULT NOW()
# );

engine = create_engine('postgresql://user:pass@localhost:5432/inventorydb')

consumer = KafkaConsumer(
    'events.Order',
    bootstrap_servers=['localhost:9092'],
    group_id='inventory-service',
    auto_offset_reset='earliest',
    enable_auto_commit=False,          # 수동 커밋으로 정확한 제어
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    max_poll_records=100,
)

def process_order_event(event_data: dict, event_id: str) -> None:
    """멱등성을 보장하는 이벤트 처리"""
    with Session(engine) as session:
        with session.begin():
            # 1. 이미 처리된 이벤트인지 확인
            result = session.execute(
                text("SELECT 1 FROM processed_events WHERE event_id = :id"),
                {"id": event_id}
            )
            if result.fetchone():
                print(f"이미 처리된 이벤트 건너뜀: {event_id}")
                return

            # 2. 비즈니스 로직 실행
            event_type = event_data.get('eventType', '')

            if event_type == 'OrderCreated':
                items = event_data.get('items', [])
                for item in items:
                    session.execute(
                        text("""
                            UPDATE inventory
                            SET reserved_quantity = reserved_quantity + :qty
                            WHERE product_id = :product_id
                              AND available_quantity >= :qty
                        """),
                        {"qty": item['quantity'], "product_id": item['productId']}
                    )

            elif event_type == 'OrderCancelled':
                order_id = event_data.get('orderId')
                session.execute(
                    text("""
                        UPDATE inventory i
                        SET reserved_quantity = reserved_quantity - oi.quantity
                        FROM order_items oi
                        WHERE oi.order_id = :order_id
                          AND oi.product_id = i.product_id
                    """),
                    {"order_id": order_id}
                )

            # 3. 처리 이력 기록 (같은 트랜잭션)
            session.execute(
                text("""
                    INSERT INTO processed_events (event_id, processed_at)
                    VALUES (:id, :now)
                """),
                {"id": event_id, "now": datetime.utcnow()}
            )

    print(f"이벤트 처리 완료: {event_id} ({event_type})")


# 메인 컨슈머 루프
print("이벤트 컨슈머 시작...")
try:
    for message in consumer:
        try:
            event_data = message.value
            # Debezium Outbox Event Router가 설정한 헤더에서 event_id 추출
            headers = {k: v.decode('utf-8') for k, v in message.headers}
            event_id = headers.get('id', str(uuid.uuid4()))

            process_order_event(event_data, event_id)

            # 처리 성공 시에만 오프셋 커밋
            consumer.commit()

        except Exception as e:
            print(f"이벤트 처리 실패: {e}")
            # 실패 시 DLQ(Dead Letter Queue)로 전송하거나 재시도 로직 적용
            # 여기서는 단순히 로깅하고 다음 이벤트로 진행

except KeyboardInterrupt:
    print("컨슈머 종료")
finally:
    consumer.close()

# 1. 커넥터 상태 확인
curl -s http://localhost:8083/connectors/order-outbox-connector/status | jq .

# 2. 커넥터 태스크 재시작 시도
curl -X POST http://localhost:8083/connectors/order-outbox-connector/tasks/0/restart

# 3. 재시작이 안 되면 커넥터 삭제 후 재등록
curl -X DELETE http://localhost:8083/connectors/order-outbox-connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @order-outbox-connector.json

# 4. PostgreSQL WAL 슬롯 상태 확인
psql -c "SELECT * FROM pg_replication_slots;"

# 5. 필요 시 슬롯 재생성
psql -c "SELECT pg_drop_replication_slot('order_outbox_slot');"
# 커넥터 재등록 시 자동으로 슬롯이 생성됨

-- 일별 파티셔닝된 Outbox 테이블
CREATE TABLE outbox_events (
    id              UUID NOT NULL 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 WITH TIME ZONE NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- 일별 파티션 자동 생성 (pg_partman 활용)
SELECT partman.create_parent(
    'public.outbox_events',
    'created_at',
    'native',
    'daily'
);

-- 7일 이상 된 파티션 자동 삭제
UPDATE partman.part_config
SET retention = '7 days', retention_keep_table = false
WHERE parent_table = 'public.outbox_events';