[Server] 신뢰성 확보 WITH Transactional OutBox Pattern

 

안녕하세요, 절박한 개발자입니다.

이번 글에서는 MSA(Microservices Architecture) 환경에서 서버 간 이벤트를 어떻게 처리했는지, 그리고 오류 발생 시 어떤 전략을 사용해 문제를 해결했는지에 대해 공유하려 합니다.

 

그럼, 본격적으로 시작해보겠습니다.

프로젝트 프로세스

판매자는 애완견 관련 상품을 등록하고, 구매자는 자체 페이 서비스인 ‘멍페이’(카카오페이와 유사)를 이용해 안심결제 또는 계좌 이체로 상품을 구매할 수 있습니다.

구매자는 안심결제를 통해 물건을 안전하게 받은 후 '구매 확정' 버튼을 누릅니다. 그러면 중고 물품 서비스에서 해당 상품이 '거래 완료' 상태로 변경되고, 이벤트가 발행되어 페이 서비스가 안심 결제 수수료를 제외한 금액을 판매자에게 입금합니다.

메시지 발행 실패

앞서 말했듯이 사용자가 중고 물품 구매 확정합니다. 그리고 중고물품 서버는 메시지를 발행시킵니다.

      
        
        if (command.status().equals(UsedProductStatus.SO) && command.paymentType().equals(PaymentType.SAFE)) {
            usedProductEntity.modifyStatusBySafe(command.status());
            UsedProductSoldOutBySafeEvent event = UsedProductSoldOutBySafeEvent.builder()
                    .price(usedProductEntity.getPrice())
                    .sellerId(usedProductEntity.getMemberId())
                    .buyerId(command.memberId())
                    .build();
            kafkaProducer.sendKafkaMessage(KafkaTopic.USED_PRODUCT_SAFE_SOLD,event);
        }

하지만, 위의 코드는 심각한 문제를 내포하고있습니다.

modifyStatusBySafe()라는 메소드로 상태를 변경하는 트랜잭션은 커밋되어 중고물품 데이터베이스에는 구매 확정 상태로 변경이 되었습니다. 하지만 kafkaProducer.sendKafkaMessage()에 예외가 발생하여 메시지가 발행되지 않아 판매자게에 판매 금액이 입금되지 않았습니다. 이는 판매자에게 피해로 이어집니다.

해당 문제를 해결하기 위해 찾아보는 도중 Transactional Outbox Pattern 사용하여 해결할 수 있는걸 발견하였습니다.

Transactional Outbox Pattern

Transactional Outbox Pattern은 데이터베이스 트랜잭션과 이벤트 발행을 원자적으로 처리할 수 있습니다.

데이터 변경과 함께 발행할 이벤트를 동일한 트랜잭션 내 아웃박스 테이블에 저장한 뒤, 별도의 배치 프로세스가 이 테이블을 읽어 이벤트를 안정적으로 전달합니다. 이를 통해 반드시 이벤트가 발행되도록 보장할 수 있습니다.

 

@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Entity
@EntityListeners(AuditingEntityListener.class)
@Table(name = "used-product-outbox")
public class UsedProductOutbox {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Integer usedProductMessageId;

    private String topic;

    @Enumerated(EnumType.STRING)
    private EventType evenType;

    @Column(columnDefinition = "JSON", nullable = false)
    private String payload;

    @Enumerated(EnumType.STRING)
    private Status status;

    @CreatedDate
    private LocalDateTime createdAt;

    @Builder
    public UsedProductOutbox(String topic, EventType evenType, String payload, Status status) {
        this.topic = topic;
        this.evenType = evenType;
        this.payload = payload;
        this.status = status;
    }

    public enum Status {
        PENDING,
        PUBLISHED
    }

    public enum EventType {
        USED_PRODUCT_SAFE_SOLD
    }
}

  @Override
    @Transactional
    public Integer modifyStatus(ModifyUsedProductStatusCommand command) {
     	
        ...
        
        if (command.status().equals(UsedProductStatus.SO) && command.paymentType().equals(PaymentType.SAFE)) {
			...

            UsedProductOutbox outbox = outBoxFactory.create(event, UsedProductOutbox.EventType.USED_PRODUCT_SAFE_SOLD, KafkaTopic.USED_PRODUCT_SAFE_SOLD);
            usedProductEntity.modifyStatusBySafe(command.status());
            usedProductOutboxRepository.save(outbox);
        }
        
        return usedProductEntity.getUsedProductId();
    }

@Transactional 어노테이션을 활용하여, 중고물품 상태 변경과 동시에 Outbox 테이블 및 UsedProduct 테이블에 원자적으로 데이터를 저장합니다.  중고물품 상태가 변경되었음에도 불구하고, 발행되어야 할 메시지가 Outbox 테이블에 저장되지 않는 상황은 발생하지 않습니다.

 

Outbox 메시지 처리

이제 Outbox에 저장된 메시지를 처리할 시스템을 구성해봅시다.

    @Scheduled(fixedRate = 10000)
    public void publishMessage() {
        List<UsedProductOutbox> messages = usedProductOutboxRepository.read();
        messages.forEach(message -> {
            producer.sendKafkaMessage(message.getTopic(), message.getPayload())
                    .whenComplete((result, throwable) -> {
                        message.markAsPublished();
                        usedProductOutboxRepository.save(message);
                    });
        });
    }

Kafka 메시지는 UsedProductOutbox에 저장된 데이터를 주기적으로 Polling하여, 전송에 성공할 때까지 발행되도록 구현하였습니다. 발행이 성공하였다면 데이터베이스에서 발행되었다고 상태를 변경합니다. 이는 At-Least-Once Delivery 전략을 적용한 것으로, 메시지가 최소 한 번 이상 전송되도록 보장합니다.

실제 메시지 소비

@KafkaListener(topics = "used-product-safe-sold")
    public void processUsedProductSafeSold(String message, Acknowledgment ack) {
        try {
			...
            
            safePayService.confirmSafeTransaction(command);
            ack.acknowledge();
        } catch (JsonProcessingException e) {
            log.error(e.getMessage());
        }
    }

 

Kafka는 기본적으로 auto-commit 설정이 되어 있어, 메시지 처리에 실패하더라도 offset이 커밋될 수 있습니다. 이러한 상황은 판매자에게 금전적인 피해로 이어질 수 있기 때문에, 저는 수동 커밋 전략을 적용하여, 메시지가 정상적으로 처리된 경우에만 커밋되도록 하였습니다.

 

다만, At-Least-Once 전략의 특성상 중복 메시지 처리 가능성이 존재하므로, 이에 대한 멱등성 처리도 함께 고려되어야 합니다.

멱등성을 보장하기 위한 방법으로는 다음과 같은 방식이 있습니다:

• 메시지에 고유 식별자(ID)를 부여하고, 이미 처리된 ID의 메시지는 무시
• 메시지를 수신할 때마다 상품의 현재 상태를 조회하여, 동일한 결과가 되도록 처리

이러한 멱등성 처리 방식들을 통해, 중복 메시지로 인한 부작용 없이 안정적인 메시지 처리를 구현할 수 있습니다.

 

 

마무리

이렇게 해서 Transactional Outbox 패턴 도입기를 공유드렸습니다.
진행하면서도 모든 것을 완벽히 해낸 것 같진 않지만, 구현 과정을 통해 이 패턴의 개념을 좀 더 정확하게 이해할 수 있었던 것 같습니다.

참고로, Transactional Outbox 패턴을 구현하는 방법에는 이번에 소개한 방식 외에도 CDC기반의 방식도 있습니다. 관심 있는 분들은 해당 방식도 함께 공부해보시면 좋을 것 같습니다.

 

이번 글은 유난히 흐름이 길었던 것 같네요.
하지만 처음으로 아키텍처 흐름을 스스로 그림으로 정리해보며, 제 나름의 도입기를 체계적으로 정리해볼 수 있어서 꽤 뿌듯한 시간이었습니다.다음 글에서는 JPA의 LOCK을 활용한 동시성 제어에 대해 다뤄보겠습니다.
끝까지 읽어주셔서 감사합니다. 😊

참조 : 

https://ridicorp.com/story/transactional-outbox-pattern-ridi/

Transactional Outbox 패턴으로 메시지 발행 보장하기 - 리디주식회사

 

https://blog.gangnamunni.com/post/transactional-outbox/

분산 시스템에서 메시지 안전하게 다루기

 

https://dkswnkk.tistory.com/741

[Kafka] 프로듀서 멱등성 보장하기