DB 커밋은 됐는데 Kafka 이벤트가 사라졌다 — PostgreSQL + Kafka로 Dual-Write 문제를 Transactional Outbox 패턴으로 해결하기

┌──────────────────────────────────────────────────┐
│                   애플리케이션                    │
│           createOrder(), createPayment()          │
└──────────────────────┬───────────────────────────┘
                       │ 단일 ACID 트랜잭션
                       ▼
┌──────────────────────────────────────────────────┐
│                  PostgreSQL                       │
│   ┌─────────────┐        ┌────────────────────┐  │
│   │   orders    │   +    │   outbox_events     │  │
│   └─────────────┘        └──────────┬─────────┘  │
└─────────────────────────────────────┼────────────┘
                                      │
              ┌───────────────────────┤
              │                       │
              ▼                       ▼
  ┌───────────────────┐   ┌─────────────────────────┐
  │  Polling Publisher │   │  Debezium CDC            │
  │  (500ms 주기 폴링) │   │  (WAL 실시간 스트리밍)   │
  └─────────┬─────────┘   └────────────┬────────────┘
             └──────────────┬──────────┘
                            ▼
                ┌───────────────────────┐
                │         Kafka          │
                │  order.events          │
                │  payment.events        │
                └───────────┬───────────┘
                            │
              ┌─────────────┴────────────┐
              ▼                          ▼
   ┌──────────────────┐     ┌───────────────────────┐
   │    재고 서비스    │     │      배송 서비스        │
   └──────────────────┘     └───────────────────────┘

CREATE TABLE outbox_events (
  id             UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_type VARCHAR(255) NOT NULL,  -- 비즈니스 단위 타입: "Order", "Payment"
  aggregate_id   VARCHAR(255) NOT NULL,  -- 해당 단위의 고유 ID (Kafka 파티션 키로 사용)
  event_type     VARCHAR(255) NOT NULL,  -- 이벤트 종류: "OrderCreated", "PaymentCompleted"
  payload        JSONB        NOT NULL,
  created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
  published_at   TIMESTAMPTZ,            -- NULL = 미발행, NOT NULL = 발행 완료
  retry_count    INT          NOT NULL DEFAULT 0,
  failed_at      TIMESTAMPTZ            -- 최근 발행 실패 시각
);
 
-- 미발행 이벤트 조회 성능을 위한 부분 인덱스
CREATE INDEX idx_outbox_unpublished
  ON outbox_events (created_at ASC)
  WHERE published_at IS NULL;

// TypeScript + pg 라이브러리
import { Pool } from 'pg';
 
interface CreateOrderDto {
  userId: string;
  totalAmount: number;
}
 
async function createOrder(pool: Pool, orderData: CreateOrderDto): Promise<string> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
 
    const orderResult = await client.query(
      `INSERT INTO orders (user_id, total_amount, status)
       VALUES ($1, $2, 'PENDING')
       RETURNING id`,
      [orderData.userId, orderData.totalAmount]
    );
    const orderId = orderResult.rows[0].id;
 
    // 같은 트랜잭션 안에서 outbox에 이벤트 삽입
    await client.query(
      `INSERT INTO outbox_events
         (aggregate_type, aggregate_id, event_type, payload)
       VALUES ($1, $2, $3, $4)`,
      [
        'Order',
        orderId,
        'OrderCreated',
        JSON.stringify({
          orderId,
          userId: orderData.userId,
          totalAmount: orderData.totalAmount,
        }),
      ]
    );
 
    await client.query('COMMIT');
    return orderId;
  } catch (err) {
    await client.query('ROLLBACK');
    console.error('[createOrder] 주문 생성 실패, 롤백:', err);
    throw err;
  } finally {
    client.release();
  }
}

import { Pool } from 'pg';
import { Producer } from 'kafkajs';
 
async function runPollingPublisher(pool: Pool, producer: Producer): Promise<void> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
 
    const result = await client.query(`
      SELECT id, aggregate_type, aggregate_id, event_type, payload
      FROM outbox_events
      WHERE published_at IS NULL
      ORDER BY created_at ASC
      LIMIT 100
      FOR UPDATE SKIP LOCKED
    `);
 
    if (result.rows.length === 0) {
      await client.query('ROLLBACK');
      return;
    }
 
    // kafkajs sendBatch API
    await producer.sendBatch({
      topicMessages: result.rows.map(row => ({
        topic: `${row.aggregate_type.toLowerCase()}.events`,
        messages: [
          {
            key: row.aggregate_id,  // 파티션 키 — 같은 주문의 이벤트는 항상 같은 파티션
            value: JSON.stringify(row.payload),
          },
        ],
      })),
    });
 
    const ids = result.rows.map(r => r.id);
    await client.query(
      `UPDATE outbox_events
       SET published_at = NOW()
       WHERE id = ANY($1::uuid[])`,
      [ids]
    );
 
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    // published_at이 갱신되지 않으므로 다음 폴링에서 자동 재시도
    console.error('[PollingPublisher] 발행 실패, 다음 주기에 재시도:', err);
  } finally {
    client.release();
  }
}
 
// setInterval 대신 재귀 setTimeout — 이전 실행이 끝나야 다음이 시작됨
async function schedulePolling(pool: Pool, producer: Producer): Promise<void> {
  await runPollingPublisher(pool, producer).catch(err => {
    console.error('[PollingPublisher] 처리되지 않은 에러:', err);
  });
  setTimeout(() => schedulePolling(pool, producer), 500);
}
 
schedulePolling(pool, producer);

-- postgresql.conf 설정 (또는 ALTER SYSTEM으로 온라인 변경)
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 5;
ALTER SYSTEM SET max_wal_senders = 5;
-- 설정 후 PostgreSQL 재시작 필요

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "mydb",
    "topic.prefix": "mydb",
    "table.include.list": "public.outbox_events",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
 
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.type": "event_type",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "${routedByValue}.events",
 
    "tombstones.on.delete": "false"
  }
}

[PostgreSQL WAL]
  └─ logical replication slot (pgoutput)
       └─ Debezium PostgreSQL Connector (Kafka Connect)
            └─ Outbox Event Router SMT
                 ├─ order.events
                 ├─ payment.events
                 └─ delivery.events

-- 7일 지난 발행 완료 이벤트 정리 (pg_cron 또는 외부 cron job)
DELETE FROM outbox_events
WHERE published_at IS NOT NULL
  AND published_at < NOW() - INTERVAL '7 days';

-- WAL 보존량 확인
SELECT
  slot_name,
  active,
  pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
  ) AS wal_behind
FROM pg_replication_slots;