🤔 개발 배경
고객으로부터 주문 요청이 발생하면 본 서비스의 DB에 주문 정보를 INSERT 할 것이다. 그리고 그 주문 정보에 해당하는 외부 파트너 API를 호출해야 하는데, 이 과정은 둘 다 성공하면 COMMIT, 둘 중 하나라도 실패하면 둘 다 ROLLBACK해야 원자성을 지킬 수 있다.
하지만… 외부 파트너 호출은 프로젝트 DB 트랜잭션의 일부가 될 수 없다는 점에서 문제가 발생했다. 즉, 서로 다른 시스템 간에 진짜 원자성은 불가능하다고 생각했다. 이게 진짜 심각한 문제인게, 대충만 생각해봐도 말도 안 되는 상황이 벌어질 수도 있었다. 몇 가지만 생각해본다면,
💥 DB는 커밋됐는데, 파트너 전송이 실패한 경우
- 주문 정보 INSERT 성공
- 파트너 API 호출
- 파트너 서버 응답을 못 받아 오거나 네트워크가 끊김(문제 발생!!!)
파트너 호출 실패 했다고 롤백을 하자니 주문 정보 자체가 없어지고, 커밋을 하자니 고객은 본인이 결제했다고 굳게 믿고 있는데, 파트너는 주문이 있는지도 모를 것이다.
💥 파트너 전송은 성공했는데, DB 커밋이 실패
- 주문 INSERT 시도
- 파트너 API 호출 및 파트너 응답 성공
- DB 커밋 단계에서 커넥션이 끊긴다든지, 서버가 다운(문제 발생!!!)
이 상황도 진짜 최악이다. 주문 정보가 없는데, 파트너는 이미 상품 준비 중이다.
어떻게 하면 이 과정을 일관성 있게 유지할 수 있을까? 지금 최종 목표는 아래와 같다.
- 주문 정보는 반드시 저장되어야 한다.
- 파트너 전송도 언젠가는 반드시 성공해야 한다.
- 전송 실패 했을 경우에도 데이터 유실 없이 재처리 가능해야 한다.
- 중복 전송이 되더라도 멱등하게 처리 가능해야 한다.
따라서 설계를 내부 DB 트랜잭션으로 확실하게 주문 정보를 저장하고, 나중에 안전하게 주문 정보를 전송할 수 있어야 한다.
🗃️ Outbox 패턴
주문 정보를 무조건 커밋한 이후에 파트너로 전송해야 떠돌아 다니는 유령 주문이 없을 것이다. 하지만 커밋 후에 전송할 때에도 서버가 죽거나 장애가 발생하면 전송이 유실될 수도 있어서 “Outbox 패턴” 을 도입하기로 결정했다.
개념은 간단하다. 주문 정보를 DB에 INSERT 할 때 커밋된 이후에 파트너에 전송할 이벤트도 같이 저장하면 된다. 즉, 주문 정보를 INSERT 하는 작업과 나중에 전송할 이벤트를 DB에 저장하는 과정이 하나의 트랜잭션 내부에서 일어나게 하면 된다는 것이다.
프로젝트에서는 같은 트랜잭션 내부에 orders 테이블에 주문 정보를 삽입하는 작업과 outbox_events 테이블에 그 주문 정보에 해당하는 이벤트를 삽입하는 작업을 넣어두었다. 이렇게 하면 주문은 저장됐는데 이벤트가 없는 상태가 있을 수 없기 때문에 데이터 유실을 막을 수 있다.
대략 구현 방법은 아래와 같다.
- 고객이 주문을 요청한다.
OrderService가 트랜잭션을 시작한다.orders테이블에 해당 주문 정보를 저장하거나 멱등키로 기존 주문을 반환한다.- 신규 주문인 경우에만
outbox_events에 방금 주문 정보에 대한 이벤트를 저장한다. - 별도의 스케줄러가 주기적으로
outbox_events에서 이벤트들을 처리하도록 한다. - 성공 혹은 실패에 따라 해당 이벤트의 상태를 바꾼다.
일단 outbox_events에 삽입될 데이터의 형태를 아래와 같이 정의해줬다.
package order_system.pickup.outbox.dto;
import java.time.Instant;
public record OutboxEvent(
Long id,
String eventType,
String aggregateType,
Long aggregateId,
String payload,
String status,
int retryCount,
Instant nextRunAt,
String lastError,
Instant createdAt
) {}
그리고 OutboxRepository를 생성해서 그 안에 신규 생성된 주문 정보에 대한 이벤트를 outbox_events 테이블에 삽입하는 로직을 작성해두었다.
...
@Repository
public class OutboxRepository {
private final JdbcTemplate jdbcTemplate;
public OutboxRepository(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
...
public void saveOrderCreated(OrderCreatedEventPayload payload) {
String sql = "INSERT INTO outbox_events(event_type, aggregate_type, aggregate_id, payload) VALUES (?, ?, ?, CAST(? AS JSON))";
jdbcTemplate.update(
sql,
OutboxEventType.ORDER_CREATED,
AggregateType.ORDER,
payload.orderId(),
toJson(payload)
);
}
...다만, 지금 이벤트마다 payload 구조는 다르기 때문에 다른 타입이 필요했다. 현재 payload는 JSON String이라서 타입 안전성이 없다. 그리고 현재는 이벤트 타입이 ORDER_CREATED인 것들만 있어서 망정이지, 향후에는 이벤트 타입이 결제 승인인 payload, 주문 취소인 payload 등 각 이벤트에 맞는 DTO를 쓰는 것이 자연스럽다고 생각했다.
package order_system.pickup.outbox.dto;
import java.time.Instant;
public record OrderCreatedEventPayload(
String eventId,
int schemaVersion,
Instant occurredAt,
Long orderId,
Long storeId,
Integer totalPrice,
String idempotencyKey
) {
public static final int SCHEMA_VERSION = 1;
public static OrderCreatedEventPayload of(Long orderId,
Long storeId,
Integer totalPrice,
String idempotencyKey,
String eventId,
Instant occurredAt) {
return new OrderCreatedEventPayload(
eventId,
SCHEMA_VERSION,
occurredAt,
orderId,
storeId,
totalPrice,
idempotencyKey
);
}
}
이렇게 처리를 하고, OrderService에서 주문 정보를 orders 테이블에 저장하는 작업과 해당 주문 정보에 대한 이벤트를 outbox_events 테이블에 삽입하는 과정을 하나의 트랜잭션으로 묶어 주었다.
@Transactional
public OrderResponse createOrder(OrderCreateRequest req, String idempotencyKey) {
validateIdempotencyKey(idempotencyKey);
var existOrder = orderRepository.findByIdempotencyKey(idempotencyKey);
if (existOrder.isPresent()) {
return existOrder.get();
}
boolean existStore = storeRepository.findById(req.storeId()).isPresent();
if (!existStore) {
throw new StoreNotFoundException(req.storeId());
}
Long orderId;
try {
// 신규 주문의 경우에 orders 테이블에 주문 정보를 저장
orderId = orderRepository.save(req, idempotencyKey);
} catch (DuplicateKeyException e) {
return orderRepository.findByIdempotencyKey(idempotencyKey)
.orElseThrow(() -> new IdempotencyKeyInconsistentStateException(idempotencyKey, e));
}
var payload = OrderCreatedEventPayload.of(
orderId,
req.storeId(),
req.totalPrice(),
idempotencyKey,
java.util.UUID.randomUUID().toString(),
java.time.Instant.now()
);
// ORDER_CREATED 이벤트를 outbox_events 테이블에 저장
outboxRepository.saveOrderCreated(payload);
return orderRepository.findById(orderId)
.orElseThrow(() -> new OrderNotFoundException(orderId));
}이로써 커밋이 성공했다면 주문 정보와 그 주문 정보에 해당하는 이벤트도 반드시 존재하고, 커밋에 실패하면 둘 다 없어지는 것을 보장할 수 있게 되었다.
🪝 Dispatcher Polling
지금까지의 내용을 다시 복기해보자. 주문 생성 API에서 해야 할 일은 “주문 정보를 DB에 저장하는 것”, “파트너로 주문을 전송하는 것” 이다. 문제는 파트너로 주문을 전송하는 작업은 외부 네트워크에 의존한다는 점이다.
둘 다 그냥 한 번에 묶어서 처리하면 되지 않나? 여기서 둘은 같은 트랜잭션 내부에 둘 수 없기 때문에 둘 중 무언가에 문제가 생기면 중복 전송이나 데이터 유실, 상태가 불일치하는 등의 불상사가 생길 수 있었다. 그래서 파트너로 이벤트를 전송하는 작업은 별도의 Dispatcher가 처리하는 것이 더 효율적이라는 생각을 했다. 나름대로 Dispatcher가 해야 하는 일을 정리해보니 아래와 같았다.
outbox_events에서 처리해야 할 이벤트를 조회한다.- 이벤트를 선점해서 처리하도록 한다.
- 성공하면 PROCESSED 처리한다.
- 실패하면 재시도 횟수를 증가시키고, 일정 시간이 지난 이후에 재시도하도록 설정한다.
이때 이벤트를 선점한다는 것은 여러 워커가 떠도 같은 이벤트를 동시에 처리하지 않도록 하기 위함이다. 따라서 아래와 같이 OutboxRepository에 이벤트 status가 PENDING인 이벤트를 찾아서 락을 해주도록 findAndLockPending() 메서드를 추가하고, 선점에 성공하면 status를 PROCESSING으로 바꿔줄 markProcessed() 메서드도 추가해주었다.
public List<OutboxEvent> findAndLockPending(int limit, String workerId) {
String sql = """
SELECT id, event_type, aggregate_type, aggregate_id, payload, status,
retry_count, next_run_at, last_error, created_at
FROM outbox_events
WHERE status = 'PENDING'
AND (next_run_at IS NULL OR next_run_at <= CURRENT_TIMESTAMP)
ORDER BY id
LIMIT ?
FOR UPDATE SKIP LOCKED
""";
List<OutboxEvent> events = jdbcTemplate.query(sql, OUTBOX_ROW_MAPPER, limit);
if (events.isEmpty()) {
return events;
}
String lockSql = """
UPDATE outbox_events
SET status = 'PROCESSING',
locked_by = ?,
locked_at = CURRENT_TIMESTAMP
WHERE id = ?
""";
jdbcTemplate.batchUpdate(
lockSql,
events,
events.size(),
(ps, event) -> {
ps.setString(1, workerId);
ps.setLong(2, event.id());
}
);
return events;
}
...
public int markProcessed(Long id, String workerId) {
String sql = """
UPDATE outbox_events
SET status = 'PROCESSED',
processed_at = CURRENT_TIMESTAMP,
next_run_at = NULL,
last_error = NULL,
locked_by = NULL,
locked_at = NULL
WHERE id = ?
AND status = 'PROCESSING'
AND locked_by = ?
""";
return jdbcTemplate.update(sql, id, workerId);
}
...
그리고 실제 outbox_events에서 이벤트들을 꺼내 파트너에게 전송할 OutboxDispatcher를 구현했다. 아래는 코드 중 일부를 발췌한 부분이다.
@Scheduled(fixedDelay = 1000)
public void dispatch() {
List<OutboxEvent> events = outboxRepository.findAndLockPending(BATCH_SIZE, workerId);
if (events.isEmpty()) {
return;
}
log.info("Outbox 디스패처 실행: size={}, workerId={}", events.size(), workerId);
for (OutboxEvent event : events) {
try {
handle(event);
int updated = outboxRepository.markProcessed(event.id(), workerId);
if (updated == 0) {
log.warn("markProcessed 실패: id={}, workerId={}", event.id(), workerId);
}
} catch (Exception e) {
onFailure(event, e);
}
}
}
private void handle(OutboxEvent event) {
String eventType = event.eventType();
if ("ORDER_CREATED".equals(eventType)) {
// TODO: 향후에 실제 파트너 어댑터 혹은 클라이언트 호출
log.debug("Handle ORDER_CREATED: id={}, aggregateId={}", event.id(), event.aggregateId());
return;
}
throw new IllegalArgumentException("알 수 없는 이벤트 타입입니다. " + eventType);
}
private void onFailure(OutboxEvent event, Exception e) {
int nextRetry = event.retryCount() + 1;
boolean toFailed = nextRetry > MAX_RETRY_ATTEMPTS;
String reason = e.getClass().getSimpleName() + ": " + e.getMessage();
if (toFailed) {
int updated = outboxRepository.markFailed(event.id(), workerId, reason);
log.warn("outbox 실패 → FAILED(DLQ): id={}, retry={}, updated={}, reason={}",
event.id(), nextRetry, updated, reason, e);
return;
}
Instant nextRunAt = Instant.now().plusSeconds(backoffTime(nextRetry));
int updated = outboxRepository.markRetry(event.id(), workerId, nextRetry, nextRunAt, reason);
log.warn("outbox 실패 → 재시도 예약: id={}, retry={}, nextRunAt={}, updated={}, reason={}",
event.id(), nextRetry, nextRunAt, updated, reason, e);
}
🗳️ Polling이란?
폴링(Polling)은 DB를 일정 주기로 확인해서 해야 할 일을 가져오는 방식을 말한다. 지금 위의 dispatch() 메서드를 보면 @Scheduled 애노테이션을 통해 1초마다 PENDING 이벤트를 조회하도록 했다.
일단 현재 Kafka나 RabbitMQ와 같은 메시지 브로커에 대한 지식이 전혀 없기도 하고, 이 정도까지는 너무 과한 것 같아 폴링 방식을 채택했다. 하지만 폴링 방식은 트래픽이 없는데도 계속 쿼리를 날리기 때문에 DB에 부하가 발생할 수 있고, 긴 간격으로 처리하면 지연이 커질 수도 있다. 따라서 향후 기능을 확장해나가면서 메시지 브로커를 도입해볼 예정이다.
지금은 폴링 방식으로 여러 워커들이 돌아도 같은 이벤트를 동시에 처리하지 않도록 락 처리를 해주었고, 만약 파트너에게 이벤트 전송을 실패한다면 재시도하도록 처리했다. 장애가 계속 지속되어 이벤트가 계속 쌓이고, 워커는 계속 같은 이벤트만 재시도하느라 리소스를 소모하게 된다. 따라서 일정 재시도 횟수를 초과할 경우에는 status를 FAILED로 전환하고, 자동 처리 대상에서 제외하고 “DLQ(Dead Letter Queue)” 에 따로 보관하도록 했다.
이로써 OutboxDispatcher를 통해 API를 안정화시키고, 이벤트 유실을 방지할 수 있게 되었다.
