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';