9단계 — Kafka 토픽 설계

6단계에서 Kafka 를 "쓸지 말지" 를 정했습니다. 쓰기로 했다면 다음 질문은 "어떻게" 입니다. 토픽 이름 · 파티션 · 키 · Producer/Consumer 설정 · 멱등 컨슈머까지, 한 토픽을 처음부터 끝까지 설계해 봅니다. 코드는 리팩터로 고치지만 토픽 설계는 마이그레이션이 필요하므로 처음이 중요합니다.

1. 토픽 이름부터 정한다

토픽 이름은 API 입니다. 컨슈머가 의존하면 바꾸기 어렵습니다. <domain>.<entity>.<event> 패턴을 권장합니다.

orders.order.created
orders.order.cancelled
billing.invoice.paid

• 구분자는 dot 또는 kebab 중 하나로 통일 (섞으면 정렬·필터가 어긋남)
• 환경은 prefix (dev.orders.order.created) 또는 별도 클러스터로 분리 — 클러스터 분리가 더 안전
• 이름을 코드에 하드코딩하지 말 것 — 설정값으로 주입

// config.ts — 토픽명을 설정으로
export const TOPICS = {
  orderCreated: process.env.ORDER_CREATED_TOPIC ?? "orders.order.created",
};

2. 파티션 수와 키를 정한다

파티션은 두 가지를 동시에 정합니다 — 처리량 상한과 컨슈머 병렬도 상한.

Topic: orders.order.created  (partitions = 6)
├── Partition 0  ◀── key hash
├── Partition 1
├── ...
└── Partition 5

컨슈머 그룹 안 컨슈머 ≤ 6

• 같은 키 → 같은 파티션 → 파티션 내 순서 보장. 한 주문 id 를 키로 쓰면 그 주문의 이벤트가 한 파티션에 모입니다.
• 키가 없으면 라운드로빈 — 순서 보장이 사라집니다.
• 파티션 수를 늘리면 hash(key) % count 가 바뀌어 같은 키가 다른 파티션으로 갑니다. 순서 가정이 깨지므로 처음에 예상 처리량의 2~3 배로 약간 넉넉히.

await producer.send({
  topic: TOPICS.orderCreated,
  messages: [
    { key: order.id, value: JSON.stringify(event) },  // key = 순서 단위
  ],
});

3. Producer 설정 — 유실 없이 보내기

"보냈다" 가 아니라 "브로커가 받았다" 까지 확인합니다.

import { Kafka } from "kafkajs";

const kafka = new Kafka({ clientId: "my-service", brokers: ["kafka:9092"] });
const producer = kafka.producer({
  idempotent: true,        // 재시도로 인한 중복 발행 방지
});

await producer.connect();
await producer.send({
  topic: TOPICS.orderCreated,
  acks: -1,                // -1 = all (ISR 전체 확인) · 유실 방지
  messages: [{ key: order.id, value: JSON.stringify(event) }],
});

• acks: -1 (= all) — ISR 전체가 받을 때까지 확인. 데이터가 중요하면 필수
• idempotent: true — Producer 재시도가 중복을 만들지 않도록 브로커가 막음
• 매 메시지를 await 로 하나씩 기다리면 처리량이 무너집니다 — 배치로 보내고 결과는 한 번에 확인

DB 커밋 전 발행 금지 — 트랜잭션 안에서 메시지를 먼저 보내면 롤백돼도 메시지는 나갑니다. outbox 테이블에 같은 트랜잭션으로 INSERT 하고, 별도 발행기가 커밋된 행만 읽어 발행합니다.

4. Consumer 그룹 — 오프셋을 직접 관리

const consumer = kafka.consumer({
  groupId: "order-notifier",          // 같은 그룹 = 파티션 분배
});

await consumer.connect();
await consumer.subscribe({
  topic: TOPICS.orderCreated,
  fromBeginning: true,                // 새 그룹은 처음부터 — 유실 방지
});

await consumer.run({
  autoCommit: false,                  // 처리 완료 후 직접 커밋
  eachMessage: async ({ message, partition, topic }) => {
    await handle(message);
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

• groupId — 같은 그룹의 컨슈머가 파티션을 나눠 가짐. 다른 그룹은 같은 토픽을 독립적으로 읽음
• fromBeginning: true — 저장된 오프셋이 없을 때 처음부터 (kafkajs 의 auto.offset.reset=earliest 대응). false 면 그 사이 메시지를 조용히 건너뜀
• autoCommit: false + 처리 후 수동 커밋 — at-least-once 와 맞물림
• 컨슈머 안에서 오래 블로킹하면 그룹이 죽은 것으로 보고 리밸런싱이 반복됨

5. 멱등 컨슈머 + DLQ

Kafka 기본 전달 보증은 at-least-once — 같은 메시지가 두 번 올 수 있습니다. 중복 제거는 컨슈머 책임입니다.

async function handle(message: KafkaMessage) {
  const payload = JSON.parse(message.value!.toString());
  const messageId = payload.id;                 // 발행 시 부여한 uuid

  if (await alreadyProcessed(messageId)) return; // 멱등 — 중복 스킵

  try {
    await applyBusinessLogic(payload);
    await markProcessed(messageId);              // 처리완료 기록
  } catch (e) {
    if (isTransient(e)) throw e;                 // 일시적 → 재시도
    await sendToDlq(message, e);                 // 영구적 → DLQ
  }
}

-- 멱등성 테이블 — UNIQUE 제약이 중복 INSERT 를 막음
CREATE TABLE IF NOT EXISTS processed_messages (
  message_id  TEXT PRIMARY KEY,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

• 발행 시 메시지마다 uuid 를 넣고, 컨슈머는 처리 전 조회 후 중복이면 스킵
• DLQ (Dead Letter Queue) — 잘못된 형식 등 영구 실패 메시지를 별도 토픽으로 보냄. 메인 흐름이 막히지 않음
• 일시적 오류는 즉시가 아니라 점점 간격을 늘려 (backoff) 재시도, 한도 초과 시 DLQ

6. 스키마 진화 (간단히)

Producer 와 Consumer 가 따로 배포되므로 메시지 형식 변경은 깨지기 쉽습니다.

• 기본값 있는 (optional) 필드 추가 — 보통 안전. 옛 컨슈머가 무시
• 필드 삭제 · 이름 변경 · 타입 변경 — 깨지는 변경. 단계를 거쳐 진행
• 규모가 커지면 Schema Registry (Avro/Protobuf/JSON Schema) 로 호환성 (backward 권장) 을 자동 검증

MVP 는 JSON 으로 시작하고, 안정화 후 Schema Registry 를 도입하는 순서가 자연스럽습니다.

자주 걸리는 자리

• 매 메시지 동기 대기 — 배치로 보냄. 하나씩 await 하면 처리량 급락
• 파티션 수 과소 — 4 개로 시작했다 컨슈머 8 개로 못 늘림. 늘리면 순서 깨짐
• 컨슈머 안 장시간 블로킹 — max.poll.interval.ms 초과 → 리밸런싱 반복
• fromBeginning: false 로 새 그룹 — 그 사이 메시지 유실. 새 그룹은 true
• 잘못된 메시지에 예외 전파 — 오프셋 미커밋 → 무한 재처리. 영구 실패는 DLQ
• DB 커밋 전 발행 — 롤백돼도 메시지 나감. outbox 또는 after-commit 으로 분리

하고픈 말

토픽 이름 · 파티션 수 · 키 전략은 한 번 정하면 운영 내내 따라옵니다. 처음에 도메인 명명 규칙과 키 전략만 합의해 두면, 멱등 컨슈머 · DLQ · 스키마는 점진적으로 더해 갈 수 있습니다.

Next

• data/13-kafka-topics
• data/05-data-pipeline