이중 쓰기(Dual Write)란 하나의 비즈니스 로직에서 둘 이상의 시스템에 데이터를 쓰는 작업이다. 이 방식은 서로 다른 시스템에 데이터를 쓰기 때문에 원자성을 보장하지 못하고 정합성이 훼손될 수 있다.
Transactional Outbox Pattern은 이 문제를 해결할 수 있는 방법 중 하나다.
이 글에서는 Outbox Pattern이 무엇인지 알아보고, 원자성 및 정합성 문제를 어떻게 해결하는지 살펴본다.
Dual Write Problem
MSA 환경에서 DB에 데이터를 저장하고 메시지 브로커를 통해 이벤트를 다른 서비스로 전파하는 경우, 다음의 시나리오가 모두 발생할 수 있다.
- DB 저장(O), 이벤트 발행(O)
- DB 저장(O), 이벤트 발행(X)
- DB 저장(X), 이벤트 발행(O)
- DB 저장(X), 이벤트 발행(X)
1번과 같은 정상 케이스를 제외한 2, 3, 4번 모두 시스템 간 데이터 정합성을 지킬 수 없게 된다.
이 문제의 근본적인 원인은 DB와 메시지 브로커가 서로 다른 트랜잭션 경계를 가지기 때문이다. 결국 두 작업을 하나의 원자적 연산으로 처리할 방법이 필요하다.
Transactional Outbox Pattern
Outbox Pattern은 메시지를 외부 시스템에 직접 발행하지 않는다. 그 대신 동일 DB 내 별도 Outbox 테이블에 이벤트와 관련 정보를 저장한다.
동일 트랜잭션 내에서 데이터 처리와 이벤트 저장이 동시에 발생하기 때문에 원자성을 보장할 수 있는 것이다.
Outbox Pattern의 동작 과정은 다음과 같다.
- 데이터 변경 시점에 Outbox 테이블에 이벤트를 함께 저장한다.
- 별도 프로세스(Message Relay)가 주기적으로 Outbox 테이블을 읽는다.
- Outbox 테이블에서 조회한 이벤트를 메시지 브로커로 발행한다. 발행이 성공하면 Outbox 테이블의 해당 이벤트를 처리 완료 상태로 변경한다.
- 컨슈머에서 발행된 이벤트를 처리한다.
이제 간단한 코드 구현을 통해 Outbox Pattern을 적용해보자.
enum class AggregateType { ORDER, PAYMENT }
enum class EventType { ORDER_CREATED, ORDER_CANCELLED }
enum class OutboxStatus { PENDING, COMPLETED, FAILED }
@Entity
class OutboxEvent(
@Id @GeneratedValue(strategy = GenerationType.IDENTITY)
val id: Long = 0,
@Enumerated(EnumType.STRING)
val aggregateType: AggregateType,
val aggregateId: String,
@Enumerated(EnumType.STRING)
val eventType: EventType,
val payload: String,
@Enumerated(EnumType.STRING)
var status: OutboxStatus,
val createdAt: LocalDateTime
)Outbox Pattern의 핵심은 데이터 처리와 이벤트 저장을 동일 트랜잭션 내에서 수행하는 것이다. 주문 저장과 Outbox 데이터 저장이 모두 같은 트랜잭션에서 실행된다. 둘 중 하나라도 실패하면 모두 롤백된다.
data class OrderCreatedEvent(
val orderId: Long,
val productId: Long,
val quantity: Int
)@Service
class OrderService(
private val orderRepository: OrderRepository,
private val outboxEventRepository: OutboxRepository,
private val objectMapper: ObjectMapper
) {
@Transactional
fun createOrder(productId: Long, quantity: Int): Order {
val order = orderRepository.save(
Order(productId = productId, quantity = quantity)
)
val event = OrderCreatedEvent(
orderId = order.id,
productId = order.productId,
quantity = order.quantity
)
outboxEventRepository.save(
OutboxEvent(
aggregateType = AggregateType.ORDER,
aggregateId = order.id.toString(),
eventType = EventType.ORDER_CREATED,
payload = objectMapper.writeValueAsString(event)
)
)
return order
}
}저장된 이벤트는 Message Relay 컴포넌트가 별도 프로세스에서 조회한다. 조회된 이벤트는 메시지 브로커에 발행한다.
아래는 Relay의 간단한 예시 코드이다.
@Component
class MessageRelay(
private val outboxRepository: OutboxRepository,
private val kafkaTemplate: KafkaTemplate<String, String>
) {
private val logger = LoggerFactory.getLogger(javaClass)
@Scheduled(fixedDelay = 1000)
@Transactional
fun publish() {
val events = outboxRepository.findByStatus(OutboxStatus.PENDING)
events.forEach { event ->
try {
kafkaTemplate.send(
"order-events",
event.aggregateId,
event.payload
)
event.status = OutboxStatus.COMPLETED
} catch (e: Exception) {
logger.error("Failed to publish event: ${event.id}", e)
event.status = OutboxStatus.FAILED
}
}
}
}지금까지 Outbox Pattern을 간단한 예시 코드로 살펴보았다.
위 구현에는 몇 가지 고려해야 할 점이 있다.
- MessageRelay가 이벤트를 발행한 후 상태 변경 전 장애가 발생하면, 같은 이벤트가 중복 발행될 수 있다.
- 여러 이벤트가 동시에 처리되면 순서가 보장되지 않을 수 있다.
이 문제의 해결 방법을 살펴보자.
멱등성
멱등성이란 동일한 연산을 몇 번 하더라도 결과가 동일함을 보장하는 성질이다. 그럼 Outbox Pattern에서 멱등성을 보장하기 위해서는 어떻게 해야 할까?
각 Outbox Event는 고유한 id 값을 갖는다. 따라서 메시지를 소비하는 쪽에서 이벤트 식별자를 저장해두고, 처리 시마다 처리된 이력이 있는지 조회하면 된다.
만약 이미 처리된 이력이 있는 경우 해당 이벤트를 무시하고, 없는 경우 이벤트 ID를 저장하고 처리한다. 이를 코드에 적용해보면 다음과 같다.
@Entity
class DomainEventLog(
@Id
val eventId: Long,
val createdAt: LocalDateTime = LocalDateTime.now()
)@Service
class OrderEventConsumer(
private val domainEventLogRepository: DomainEventLogRepository,
private val inventoryService: InventoryService
) {
@KafkaListener(topics = ["order-events"])
@Transactional
fun consume(message: String, @Header("eventId") eventId: Long) {
if (domainEventLogRepository.existsByEventId(eventId)) {
return
}
val event = objectMapper.readValue(message, OrderCreatedEvent::class.java)
inventoryService.decreaseStock(event.productId, event.quantity)
domainEventLogRepository.save(DomainEventLog(eventId = eventId))
}
}순서 보장
여러 이벤트가 동시에 발행되면 순서가 보장되지 않을 수 있다.
예를 들어 같은 주문에 대해 생성, 취소 이벤트가 발생할 수 있다. 만약 두 이벤트가 순서대로 처리되지 않으면, 데이터 정합성이 훼손된다.
Kafka와 같은 메시지 브로커는 같은 파티션 내에서의 메시지 순서를 보장한다.
앞선 코드 예시에서는 aggregateId 값을 파티션 키로 사용하면, 같은 aggregate 내에서 처리 순서를 보장할 수 있다.
kafkaTemplate.send(
"order-events",
event.aggregateId, // 같은 파티션 키를 사용하면 동일한 파티션으로 전송된다.
event.payload
)