Files

T

Antigravity Agent 93ec7e9056 [G1-Sync] Manual knowledge update

2026-05-09 21:08:02 +09:00

4.7 KiB

Raw Blame History

id, title, category, status, source_trust_level, verification_status, created_at, updated_at, tags, tech_stack, applied_in, aliases

title

Change Data Capture

DB 변경을 실시간 stream 으로. Postgres logical replication / MySQL binlog → Debezium → Kafka. Outbox 패턴의 외부 + 무관 시스템 동기화에 강력. 앱 변경 X.

📖 핵심 개념

WAL (Postgres) / binlog (MySQL): DB 가 commit 한 모든 변경을 시간 순으로 기록.
Logical replication: WAL 을 row-level 변경으로 디코드.
CDC tool: Debezium / wal2json / pgcapsule / Materialize.
Use case: 검색 인덱스 / cache 갱신 / 마이크로서비스 동기화 / 분석 DB.

💻 코드 패턴

Postgres logical replication 활성화

-- postgresql.conf
wal_level = logical
max_replication_slots = 5
max_wal_senders = 5

-- publication 생성
CREATE PUBLICATION app_pub FOR TABLE orders, users;

-- replication slot
SELECT pg_create_logical_replication_slot('debezium', 'pgoutput');

Debezium config (Postgres → Kafka)

{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg",
    "database.user": "debezium",
    "database.dbname": "app",
    "topic.prefix": "app",
    "plugin.name": "pgoutput",
    "publication.name": "app_pub",
    "slot.name": "debezium",
    "table.include.list": "public.orders,public.users"
  }
}

자동 생성: app.public.orders, app.public.users Kafka topic.

CDC 메시지 형식

{
  "before": { "id": 1, "status": "open", ...},
  "after":  { "id": 1, "status": "shipped", ...},
  "op": "u",            // c=create, u=update, d=delete
  "ts_ms": 1234567890,
  "source": {...}
}

Consumer (검색 인덱스 동기)

const consumer = kafka.consumer({ groupId: 'es-indexer' });
await consumer.subscribe({ topic: 'app.public.orders' });

await consumer.run({
  eachMessage: async ({ message }) => {
    const e = JSON.parse(message.value!.toString());
    switch (e.op) {
      case 'c': case 'u':
        await es.index({ index: 'orders', id: e.after.id, body: e.after });
        break;
      case 'd':
        await es.delete({ index: 'orders', id: e.before.id });
        break;
    }
  },
});

Snapshot (초기 동기화)

Debezium 시작 시 초기 SELECT * → CDC stream 으로.
snapshot.mode: initial / when_needed / never.

Outbox via CDC (Debezium EventRouter)

-- outbox 테이블 (위 Outbox 패턴)
INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload) VALUES (...);

"transforms": "outbox",
"transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
"transforms.outbox.route.by.field": "aggregate_type",
"transforms.outbox.route.topic.replacement": "events.${routedByValue}"

→ 자동으로 events.order topic 등으로 라우팅.

Schema evolution

ADD COLUMN: 자동 호환
DROP COLUMN: consumer 가 안 쓰면 OK
RENAME: 보통 깨짐 — schema registry 호환 정책

Lag 모니터링

-- replication slot lag
SELECT slot_name, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag
FROM pg_replication_slots;

알람: lag > 1GB.

Retention

-- 안 쓰는 slot = WAL 무한 누적
SELECT pg_drop_replication_slot('unused');

🤔 의사결정 기준

동기화 대상	추천
Search (Elasticsearch)	CDC → Kafka → indexer
Cache (Redis)	CDC + invalidation
분석 DW (Snowflake/BQ)	CDC → Fivetran / Airbyte
마이크로서비스 read model	CDC outbox pattern
단순 동기화	App-level event
복잡 변환	Materialize / Flink

❌ 안티패턴

Slot drop 안 함: WAL 무한 — 디스크 채움.
모든 테이블 CDC: 불필요 트래픽. include.list.
Schema 변경 무사고 가정: 신중 + 테스트.
Consumer 못 따라감: lag 무한. parallelism / 처리 빠르게.
Snapshot 트랜잭션 큰 테이블: 메모리 / 시간. parallelism + chunked.
CDC 만 + 앱 이벤트 무시: app intent 와 row 변경이 다름 (UPDATE 시 의미 추측).
Replica 에서 CDC: lag 위험. primary 권장.

🤖 LLM 활용 힌트

Debezium + Kafka + outbox EventRouter 조합.
App 변경 0 + 무관 시스템 동기화.
Slot lag / retention 모니터링.

4.7 KiB Raw Blame History