TL;DR
- Producer의 At-least-once 전송과 Consumer의 Exactly-Once 처리를 보장하기 위한 방법을 제시합니다.
- 멱등성 보장을 위한 이벤트 핸들링 테이블과 이벤트 로그를 저장하는 감사 로그 테이블을 분리한 이유를 설명 합니다.
- 카프카를 어떤 배경과 요구사항에서 사용하는 것이 적절한지 살펴봅니다.
0. 들어가며
지난주에는 Spring의 ApplicationEvent를 활용해서 이벤트 구조를 설계하였습니다.
이번주에 새로운 요구사항이 생겼습니다. 카프카 기반 이벤트 파이프라인을 구현하세요.
이 글에서는 실제 주문 시스템에 Kafka를 도입하면서 겪은 시행착오를 공유합니다. At-Least-Once 전송에서 발생하는 중복 메시지 문제를 어떻게 해결했는지, 그리고 왜 테이블을 분리해야 했는지 실제 코드와 함께 살펴보겠습니다.
이번주에 구현해야될 요구사항 입니다.
- Producer에서 At-Least-Once을 보장하여 이벤트를 발행
- Transactional Outbox Pattern을 구현
- Consumer에서 이벤트를 수취해 상품의 집계(좋아요 수, 판매량, 상세 페이지 조회 수)를 upsert
1. Kafka Producer의 At-Least-Once 보장 - Transactional Outbox 패턴
Producer는 카프카에 메시지를 발행하는 역할을 수행합니다. 역할 수행을 완수하기 위해 At-Least-Once을 만족시켜야 합니다.
At-Least-Once은 '최소 한번'의 뜻으로 메시지를 적어도 한번 이상 전달합니다. 메시지 유실은 없지만, 중복 전달이 발생할 수 있습니다.
메시지가 유실되지 않음을 보장하지만 네트워크 오류나 재시도 로직으로 동일한 메시지가 둘 이상 중복 전달될 수 있습니다. 중복 전달되어도 한 번만 처리되는 'exactly one'을 보장하기 위해 컨슈머에서 멱등성을 구현해야 합니다.
문제 상황
카프카로 이벤트 발행할때 문제가 발생할 수 있습니다.
@Transactional
fun placeOrder(command: OrderCommand) {
// 1. DB에 주문 저장
orderRepository.save(order)
// 2. Kafka로 이벤트 발행
kafkaTemplate.send("order-events", orderEvent) // ← 여기서 실패하면?
}
위 코드의 문제점
- DB 저장은 성공했는데 Kafka 발행이 실패하면 → 데이터 불일치
- Kafka 발행은 성공했는데 트랜잭션 롤백되면 → 유령 이벤트 발생
해결 과정
위 문제를 해결하기 위해 transactional outbox 패턴을 사용하였습니다.
핵심 아이디어는
- 이벤트를 Kafka로 직접 보내지 말고, DB의 Outbox 테이블에 먼저 저장하고,
- 별도의 폴러가 Outbox를 주기적으로 읽어 Kafka로 발행 합니다.
구현 상세
Outbox 엔티티
@Entity
@Table(
name = "outbox",
indexes = [
Index(name = "idx_outbox_status_created_at", columnList = "status, created_at"),
Index(name = "idx_outbox_aggregate_id_type", columnList = "aggregate_id, aggregate_type"),
],
)
class Outbox(
val eventId: String, // Time-ordered UUID
val aggregateType: AggregateType, // PRODUCT, ORDER, USER
val aggregateId: String, // productId, orderId 등
val eventType: String, // LIKE_COUNT_CHANGED, ORDER_COMPLETED 등
val payload: String, // JSON 직렬화된 이벤트
var status: OutboxStatus = OutboxStatus.PENDING,
var publishedAt: ZonedDateTime? = null,
var errorMessage: String? = null,
) : BaseEntity()
enum class OutboxStatus {
PENDING, // 발행 대기
PROCESSING, // 발행 중
PUBLISHED, // 발행 완료
FAILED, // 발행 실패
}
설계 포인트:
- eventId는 Time-ordered UUID로 이벤트 순서와 유일성 보장
- status로 이벤트 발행 상태 추적
- idx_outbox_status_created_at 인덱스로 PENDING 이벤트 빠르게 조회
비즈니스 트랜잭션에서 outbox 저장
@Component
class OrderFacade(
private val orderService: OrderService,
private val outboxService: OutboxService,
) {
@Transactional
fun placeOrder(command: OrderCommand.Place) {
// 1. 주문 생성
val orderResult = orderService.createOrder(...)
// 2. 재고 차감
productService.deductAllStock(orderResult.orderDetails)
// 3. Outbox에 이벤트 저장 (같은 트랜잭션!)
saveOrderCompletedOutbox(
orderId = orderResult.order.id,
userId = user.id,
totalAmount = finalAmount,
orderDetails = orderResult.orderDetails,
)
}
private fun saveOrderCompletedOutbox(...) {
val event = OutboxEvent.OrderCompleted(
orderId = orderId,
userId = userId,
totalAmount = totalAmount,
items = orderDetails.map { OrderItem(it.productId, it.quantity) },
)
outboxService.save(
aggregateType = AggregateType.ORDER,
aggregateId = orderId.toString(),
eventType = OutboxEvent.OrderCompleted.EVENT_TYPE,
payload = event,
)
}
}
@Transactional 안에서 주문 저장과 Outbox 저장이 함께 이루어집니다. 둘 다 성공하거나 둘 다 실패합니다.
Outbox Poller (스케줄러)
@Component
class OutboxPollScheduler(
private val outboxFacade: OutboxFacade,
) {
@Scheduled(fixedDelay = 1000) // 1초마다 폴링
fun poll() {
outboxFacade.publishPendingEvents()
}
}Outbox Facade (발행 오케스트레이션)
@Component
class OutboxFacade(
private val outboxService: OutboxService,
private val outboxEventPublisher: OutboxEventPublisher,
) {
fun publishPendingEvents(batchSize: Int = 100): PublishResult {
// PENDING과 FAILED 상태의 이벤트 조회
val events = outboxService.findPendingEvents(batchSize)
for (event in events) {
try {
// 1. PROCESSING으로 상태 변경 (새 트랜잭션)
outboxService.markProcessing(event)
// 2. Kafka 발행
val success = outboxEventPublisher.publish(event)
if (success) {
// 3. PUBLISHED로 상태 변경
outboxService.markCompleted(event.id)
} else {
// 4. FAILED로 상태 변경 (다음 폴링에서 재시도)
outboxService.markFailed(event.id, "Kafka 발행 실패")
}
} catch (e: Exception) {
outboxService.markFailed(event.id, e.message)
}
}
return PublishResult(successCount, failCount)
}
}
Kafka Producer 설정
spring:
kafka:
producer:
acks: all # 모든 ISR 복제본이 ACK해야 성공
properties:
enable.idempotence: true # 프로듀서가 레코드 쓰기 작업을 단 한 번만 허용 (멱등성 보장)
At-Least-Once을 보장하기 위해 kafka producer의 주요 설정입니다
| 설정 | 의미 |
| acks: all | 프로듀서가 요청을 보내고 리더가 레플리카의 수신을 확인해야 하는 개수를 결정 |
| enable.idempotence: true | 프로듀서가 레코드 쓰기 작업을 단 한 번만 허용할 것인지를 결정. 'true'로 설정해야 멱등성을 보장한다. |
acks=all만으로는 부족한 이유:
acks=all은 "브로커가 메시지를 안전하게 저장했음"을 보장하지만, 네트워크 타임아웃으로 Producer가 ACK를 못 받으면 재전송합니다.
이때 동일 메시지가 브로커에 두 번 저장될 수 있습니다.
enable.idempotence=true는 Producer가 각 메시지에 시퀀스 번호를 부여해 브로커가 중복 메시지를 감지하고 무시하게 합니다. 두 설정을 함께 사용해야 "메시지 유실 없이 + 중복 없이" 전송을 보장합니다.
2. Kafka Consumer의 Exactly-Once 시맨틱
분산 시스템에서 메시지를 "정확히 한 번만" 처리하는 것은 가장 어려운 문제 중 하나입니다. 네트워크 장애, 컨슈머 재시작, Kafka 리밸런싱 등 다양한 상황에서 메시지가 중복 전달될 수 있기 때문입니다.
이 글에서는 commerce-streamer 모듈에서 Exactly-Once 시맨틱을 어떻게 구현했는지 설명합니다.
1. Manual Acknowledgment (수동 커밋)
왜 Auto-Commit을 끄는가?
spring:
kafka:
consumer:
enable-auto-commit: false # 자동 커밋 비활성화
listener:
ack-mode: manual # 수동 ACK 모드Kafka의 기본 설정은 enable.auto.commit=true로, 주기적으로 오프셋을 자동 커밋합니다. 문제는 처리가 완료되기 전에 오프셋이 커밋될 수 있다는 것입니다.
// ProductMetricsConsumer.kt
@KafkaListener(
topics = [OutboxEvent.LikeCountChanged.TOPIC],
groupId = "product-metrics-like-consumer",
containerFactory = KafkaConfig.BATCH_LISTENER,
)
fun consumeLikeCountChanged(
records: List<ConsumerRecord<Any, Any>>,
acknowledgment: Acknowledgment,
) {
try {
productMetricsFacade.handleLikeEvents(records, "product-metrics-like-consumer")
acknowledgment.acknowledge() // 성공 시에만 ACK
} catch (e: Exception) {
throw e // 예외 발생 → ACK 없음 → Kafka가 재전송
}
}
핵심 포인트:
- acknowledgment.acknowledge()는 **모든 처리가 성공한 후에만** 호출
- 예외 발생 시 ACK하지 않음 → Kafka가 메시지 재전송
- 이로써 "at-least-once" 보장
2. Idempotency Check (멱등성 검사)
재전송된 메시지를 중복 처리하지 않으려면 멱등성 검사가 필요합니다.
EventHandled 엔티티
// EventHandled.kt
@Entity
@Table(
name = "event_handled",
uniqueConstraints = [
UniqueConstraint(
name = "uk_event_id_aggregate_id",
columnNames = ["event_id", "aggregate_id"]
),
],
)
class EventHandled(
@Column(name = "event_id", nullable = false)
val eventId: String,
@Column(name = "aggregate_id", nullable = false)
val aggregateId: String,
@Column(name = "event_type", nullable = false)
val eventType: String,
@Column(name = "event_timestamp", nullable = false)
val eventTimestamp: ZonedDateTime,
) : BaseEntity()
왜 eventId + aggregateId인가?
처음에는 eventId만으로 중복 검사를 했지만, 문제가 발생했습니다.
문제 시나리오: 주문 완료 이벤트
// 주문 완료 이벤트: 상품 2개 포함
OrderCompleted(
orderId = 5001,
items = [
OrderItem(productId = 1001, quantity = 1),
OrderItem(productId = 1002, quantity = 2),
]
)
이 이벤트 하나로 두 상품의 판매 카운트를 각각 증가시켜야 합니다.
// eventId만 사용할 경우 (잘못된 방식)
eventId = "uuid-123"
// 상품 1001 처리: OK
existsByEventId("uuid-123") → false → 처리 → 저장
// 상품 1002 처리: BUG!
existsByEventId("uuid-123") → true → SKIP! // 처리 안됨!
해결: aggregateId 추가
// eventId + aggregateId 조합 (올바른 방식)
// 상품 1001 처리
existsByEventIdAndAggregateId("uuid-123", "1001") → false → 처리 → 저장
// 상품 1002 처리
existsByEventIdAndAggregateId("uuid-123", "1002") → false → 처리 → 저장
// 재전송 시
existsByEventIdAndAggregateId("uuid-123", "1001") → true → SKIP (정상)
existsByEventIdAndAggregateId("uuid-123", "1002") → true → SKIP (정상)
정리
| 문제 | 해결책 | 구현 |
| 처리 전 오프셋 커밋 | Manual ACK | ackMode = MANUAL |
| 메시지 재전송으로 인한 중복 | 멱등성 검사 | EventHandled 테이블 |
| 단일 이벤트에 여러 aggregate | aggregateId 추가 | (eventId, aggregateId) 복합키 |
3. 이벤트 핸들링 테이블과 감사 로그 테이블 분리: Kafka Consumer의 관심사 분리
Kafka 기반 이벤트 드리븐 아키텍처에서 이벤트를 처리할 때, 멱등성 보장과 감사 로그 기록은 모두 중요한 요구사항입니다. 하지만 이 두 가지는 본질적으로 다른 목적을 가지고 있습니다.
이번 글에서는 commerce-streamer 모듈에서 event_handled 테이블과 audit_log 테이블을 분리한 이유와 구현 방식을 살펴보겠습니다.
테이블 설계
event_handled: 멱등성 체크 전용
@Entity
@Table(
name = "event_handled",
uniqueConstraints = [
UniqueConstraint(
name = "uk_event_id_aggregate_id",
columnNames = ["event_id", "aggregate_id"]
),
],
)
class EventHandled(
@Column(name = "event_id", nullable = false)
val eventId: String,
@Column(name = "aggregate_id", nullable = false)
val aggregateId: String,
@Column(name = "event_type", nullable = false)
val eventType: String,
@Column(name = "event_timestamp", nullable = false)
val eventTimestamp: ZonedDateTime,
) : BaseEntity()
핵심 포인트:
- (eventId, aggregateId) 조합으로 유니크 제약
- 왜 aggregateId가 필요한가? → 같은 이벤트 ID라도 서로 다른 애그리게이트(상품, 주문 등)에 대한 처리는 별도로 추적해야 함
- 페이로드 없음 → 빠른 조회를 위해 최소한의 정보만 저장
audit_log: 감사 기록 전용
@Entity
@Table(name = "audit_logs")
class AuditLog(
@Column(name = "event_id", nullable = false, unique = true)
val eventId: String,
@Column(name = "event_type", nullable = false)
val eventType: String,
@Column(name = "topic_name", nullable = false)
val topicName: String,
@Column(name = "aggregate_id", nullable = false)
val aggregateId: String,
@Lob
@Column(name = "raw_payload", columnDefinition = "TEXT")
val rawPayload: String,
) : BaseEntity()
핵심 포인트:
- eventId만 유니크 → 이벤트 자체는 시스템에서 한 번만 기록
- rawPayload로 원본 JSON 보존 → 추후 분석/디버깅에 필수
- topicName 포함 → 어떤 토픽에서 왔는지 추적 가능
컨슈머 그룹 분리 전략
분리의 핵심은 독립적인 컨슈머 그룹입니다.
// 비즈니스 로직 처리 (메트릭 집계)
@KafkaListener(
topics = [OutboxEvent.LikeCountChanged.TOPIC],
groupId = "product-metrics-consumer", // 비즈니스 전용
)
fun consumeLikeEvents(records: List<ConsumerRecord<Any, Any>>, ack: Acknowledgment)
// 감사 로그 수집
@KafkaListener(
topics = [
OutboxEvent.LikeCountChanged.TOPIC,
OutboxEvent.ViewCountIncreased.TOPIC,
OutboxEvent.OrderCompleted.TOPIC,
// ... 모든 비즈니스 토픽
],
groupId = "audit-collector", // 감사 전용
)
fun auditAllEvents(records: List<ConsumerRecord<Any, Any>>, ack: Acknowledgment)
이렇게 하면:
1. 비즈니스 로직 실패가 감사 로그에 영향을 주지 않음
2. 감사 로그 처리 지연이 비즈니스 로직에 영향을 주지 않음
3. 각각 독립적으로 스케일링 가능
정리
| 구분 | event_handled | audit_log |
| 목적 | 멱등성 보장 (중복 처리 방지) | 감사 추적성 (이벤트 이력 보관) |
| 수명 주기 | 일정 기간 후 삭제 가능 (TTL/배치) | 장기 보관 필요 (법적/감사 요건) |
| 쿼리 패턴 | EXISTS 체크 (빠른 조회 필요) | 범위 검색, 분석용 쿼리 |
| 인덱스 전략 | (event_id, aggregate_id) 고유 제약 | timestamp, event_type 등 다양한 인덱스 |
| 데이터 크기 | 최소한의 메타데이터만 저장 | 전체 페이로드 포함 |
| 목적 | 멱등성 보장 (중복 처리 방지) | 감사 추적성 (이벤트 이력 보관) |
두 테이블을 분리함으로써 각각의 요구사항에 최적화된 스키마 설계와 인덱스 전략을 적용할 수 있습니다.
멱등성 검사는 간결하고 빠른 조회가 필요하고, 감사 로그는 전체 이벤트 데이터를 장기 보관하며 다양한 분석 쿼리를 지원해야 합니다.
4. Kafka 사용 여부 판단 기준
카프카를 반드시 도입하는게 좋은지 고민이 들었습니다. 적합한 경우와 적합하지 않은 경우를 정리하였습니다.
Kafka가 적합한 경우
짧고 빠른 트랜잭션 작업에 Kafka가 강점을 발휘합니다.
Kafka는 높은 처리량과 낮은 지연 시간이 필요한 이벤트 스트리밍에 최적화되어 있습니다. 다음과 같은 작업에 적합합니다.
- 실시간 알림: 푸시 알림, 이메일 발송 트리거 등 즉각적인 반응이 필요한 작업
- 결제 승인: 빠른 응답이 중요하고 트랜잭션이 짧게 끝나는 작업
- 로그 수집: 대량의 로그를 빠르게 수집하고 전달해야 하는 작업
Kafka가 적합하지 않은 경우
장시간 실행되는 복잡한 작업에는 Kafka가 오히려 독이 될 수 있습니다.
실제로 발생한 문제
https://techblog.woowahan.com/23625/
장시간 비동기 작업, Kafka 대신 RDB 기반 Task Queue로 해결하기 | 우아한형제들 기술블로그
장시간 비동기 작업, Kafka 대신 RDB 기반 Task Queue로 해결하기 | 전자계약서 시스템에서는 다양한 업무 목적에 따라 여러 형태의 대용량 엑셀 파일을 생성할 수 있습니다. 예를 들어 생산성 지표 엑
techblog.woowahan.com
글에서 소개된 사례가 좋은 예시입니다. 장시간 실행되는 복잡한 작업에는 부적합합니다.
글에서 겪은 문제가 대표적인 예시입니다. 30분 이상 걸리는 엑셀 생성 작업에서 max.poll.interval.ms(기본 5분) 타임아웃을 초과하면서 Consumer가 죽은 것으로 판단되어 리밸런싱이 발생했고, 동일 메시지가 중복 처리되는 문제가 생겼습니다.
타임아웃을 늘리면 해결되는 것처럼 보이지만, 실제 Worker 장애 시 감지가 그만큼 지연되는 부작용이 생깁니다.
Kafka가 맞지 않는 작업의 특성
다음과 같은 특성을 가진 작업이라면 Kafka 대신 다른 방식을 고려해볼 필요가 있습니다.
- 처리 시간이 수십 분에서 수 시간에 달하는 배치 작업: 대용량 엑셀 생성, 대량 데이터 마이그레이션, 복잡한 리포트 생성 등
- 중간 상태 추적이 필요한 작업: 작업이 어디까지 진행되었는지 알아야 하고, 중단 시 이어서 처리해야 하는 경우
- 복잡한 재시도 로직이 필요한 작업: 단순 재시도가 아니라 재시도 횟수, 백오프 전략 등을 세밀하게 제어해야 하는 경우
- 생산자와 소비자가 동일 서비스 내에 있는 경우: "우리가 발행하고 우리가 소비하는" 구조에서는 Kafka의 분산 메시징 이점을 거의 누리지 못합니다
결국 핵심은 작업의 트랜잭션 특성에 맞는 도구를 선택하는 것입니다. Kafka는 분명 강력한 도구이지만, 모든 비동기 작업에 정답은 아닙니다. 글에서 얻은 인사이트처럼 "좋은 시스템은 복잡한 기술 위에 세워지는 것이 아니라, 명확한 문제 정의와 단순한 설계 원칙 위에서 만들어진다"는 점을 기억하면 좋겠습니다.
끝으로
이번 글에서 다룬 내용을 정리하면 다음과 같습니다.
- Producer의 At-Least-Once 보장: DB 트랜잭션과 Kafka 발행의 원자성 문제는 Transactional Outbox 패턴으로 해결할 수 있습니다. 이벤트를 Kafka로 직접 보내지 말고, 비즈니스 트랜잭션 안에서 Outbox 테이블에 먼저 저장한 뒤 별도 폴러가 발행하는 방식입니다.
- Consumer의 Exactly-Once 처리: 메시지 중복 전달은 피할 수 없으므로, Consumer에서 멱등성을 보장해야 합니다. 이때 eventId만으로는 부족하고, (eventId, aggregateId) 조합으로 중복 검사를 해야 하나의 이벤트가 여러 애그리게이트에 영향을 주는 경우도 처리할 수 있습니다.
- 테이블 분리의 중요성: 멱등성 체크용 테이블(event_handled)과 감사 로그 테이블(audit_log)은 목적이 다르므로 분리해야 합니다. 전자는 빠른 EXISTS 체크에, 후자는 전체 페이로드 보관과 분석에 최적화할 수 있습니다.
루프백 3기 오픈
https://www.loopers.im/education
교육 과정 | Loop:Pak
Loop:Pak 부트캠프 교육 과정 상세 정보. NextNode Backend Edition을 포함한 전문 개발자 양성 커리큘럼과 현업 멘토링 프로그램을 만나보세요.
www.loopers.im
루퍼스 3기가 오픈되었습니다.
추천인 코드를 입력하면 할인가로 수강할 수 있습니다. 다만 결코 작은 금액은 아니기에, 신청 전 적지 않은 고민을 했는데요. AI 기술의 빠른 발전과 함께 연차가 쌓이면서, 스스로의 실력에 대한 의구심이 들던 시기였습니다.
며칠간 고민한 끝에 이번 기회를 마지막이라 생각하고 수강을 결정했습니다. 지금 돌이켜보면, 후회 없는 선택이었다고 자신 있게 말할 수 있습니다.
만약 저와 비슷한 고민을 하고 있거나, 루퍼스 과정에 대해 궁금한 점이 있다면 언제든지 편하게 연락 주셔도 좋습니다.
LinkedIn 또는 jikimee64@gmail.com으로 언제든 문의 부탁 드립니다. 감사합니다.
추천인 코드
TZFI0
'외부활동 > 루퍼스 2기' 카테고리의 다른 글
| [루퍼스/루프백 2기] 이커머스 실시간 랭킹 구현기 - Redis ZSET부터 윈도우 설계 관점에서 콜드 스타트 해결까지 (0) | 2025.12.21 |
|---|---|
| [Loop:PAK] 8주차 WIL (0) | 2025.12.16 |
| [Loop:PAK] 7주차 WIL (0) | 2025.12.06 |
| [루퍼스/루프백 2기] Event-Driven Architecture 도입기: Command와 Event, 그리고 트랜잭션 경계 (0) | 2025.12.06 |
| [Loop:PAK] 6주차 WIL (0) | 2025.12.04 |