이벤트가 한번 만 처리 되도록 보장하기 (feat. 멱등성)

이벤트 발행으로 결제 후처리 하기

이전 포스팅에서는 결제 이후 이벤트를 발행하여 후처리 과정을 처리하도록 하였습니다.

Transactional OutBox Pattern을 이용하여 이벤트 발행을 보장하였고 최소 1회 이상 이벤트가 발행되도록 하였습니다.

그런데 최소 1회 이상이란 것은 이벤트가 2번, 3번, 4번 발행될 수 있다는 의미입니다.

또한 동일한 결제 이벤트가 2번 이상 발행이 된다면 사용자는 한번의 결제로 똑같은 물건을 2개 구매한 것과 같은 이슈가 발생될 수

있기 때문에 이벤트가 1회만 발행되도록 보장하는 것이 중요할 수 있습니다.

AWS SNS FIFO - 이벤트 / 메시지 1회 발행 보장

AWS의 SNS의 경우 유형이 FIFO와 표준으로 나누어져 있습니다.

FIFO의 경우 Topic이 발행 된 순서대로 처리되며 정확히 1회 처리됨을 보장받을 수 있습니다.

저는 별도로 Topic 발행의 중복을 처리하는 로직을 구성하는 것은 번거로울 것 같아 사용중인 SNS의 유형은 전부 FIFO로 지정하였습니다.

SNS를 FIFO 유형으로 사용하는 경우 Topic을 발행할 때 메시지 중복 제거 ID를 지정 해야합니다.

class QuestionPaymentEvent(
    override val eventId: String,
    val questionPayment: QuestionPayment
) : SQSEvent {
    override fun toRequest(): PublishRequest {
        return PublishRequest.builder()
            .topicArn("arn:aws:sns:*****************************************")
            .messageGroupId(questionPayment.order.orderId)
            .messageDeduplicationId(questionPayment.order.orderId)
            .message(SQSEvent.objectMapper.writeValueAsString(this))
            .build()
    }
}

 

위 코드처럼 SNS에 Topic을 발행하는 과정에서 메시지 중복 제거 ID (messageDeduplicationId)를 지정해주고 있으며

결제의 경우는 결제 번호, 리뷰 작성의 경우 UUID/TSID와 같이 중복되지 않는 고유한 값을 지정해주고 있습니다.

 

이렇게 네트워크 이슈 혹은 여러가지 이유로 SNS Topic 발행이 중복되어 요청되었을 때 SNS FIFO는 메시지 중복 제거 ID를 확인하여

1회만 발행되도록 보장합니다.

메시지 / 이벤트는 1회만 소비되도록 보장하기 - 멱등성

Spring Boot와 SQS의 통신 과정을 매우 매우 간단하게 표현하자면 위와 같습니다.

 

Spring boot는 주기적으로 SQS를 polling하며 이벤트/메시지를 소비합니다.

SQS로부터 메시지를 소비하였다면 Spring boot는 적절한 로직을 수행하고 정상적으로 소비했다는 ACK응답을 주게 됩니다.

 

ACK 응답을 받은 SQS는 Spring boot가 소비한 메시지를 SQS에서 제거하게 됩니다.

만약 소비자(Spring boot)가 ACK 응답을 주지 않는다면 SQS는 해당 메시지를 지우지 않고 다른 소비자에게 전달을 합니다.

 

이러한 과정에서 메시지/이벤트가 2회 이상 소비될 수 있는 문제가 발생할 수 있습니다.

@SqsListener("append-user-question.fifo")
fun appendUserQuestion(@Payload event: QuestionPaymentEvent) {
    userQuestionRepository.saveAll(create(event.questionPayment.userId, event.questionPayment.order.questionIds))
}

 

위 코드는 문제 결제 이벤트가 발행되었을 때 구매한 문제를 나의 문제집에 추가하는 로직입니다.

 

만약 userQuestionRepository.saveAll를 정상적으로 처리했지만 갑작스런 네트워크 이슈로 인해 SQS로 ACK 응답을 주지 못했다면

SQS는 해당 이벤트를 제거하지 않고 다른 소비자에게 전달할 것 입니다.

그렇게 되면 해당 유저의 문제집엔 같은 문제가 존재할 수 있습니다.

 

이와 같은 문제를 해결하기 위해 이벤트를 처리하는 소비자가 멱등성을 제공하는 방법이 있습니다.

멱등성 - MDN Web Docs 용어 사전: 웹 용어 정의 | MDN

 

멱등성이란 용어를 간단하게 표현하면 동일한 요청을 계속 보냈을 때 서버의 상태가 동일해야 한다는 것입니다.

appendUserQuestion가 멱등하다면 매 요청마다 문제가 추가되는 것이 아닌 (매 요청마다 서버의 상태가 바뀌는 것이 아닌)

첫번째 요청에서만 문제가 추가되어야 한다는 것입니다. 그리고 동일한 요청이 온다면 이미 문제가 추가 되었기 때문에 별다른 로직을

수행하지 않고 처리되었다는 응답을 주면 됩니다.

 

일반적으로 멱등성 API의 경우 매 요청마다 멱등성 키를 받게 되는데요.

요청을 받은 서버는 DB에 멱등성 키 / 처리 결과가 있는지 확인하고 있다면 해당 응답을 바로 주고

없다면 요청을 처리하고 DB에 멱등성 키와 처리 결과를 저장하게 됩니다.

 

이벤트 소비자의 경우 일반적인 API처럼 별도의 응답을 주는 경우는 없기 때문에 단순히 EventID가 이미 처리되었는지

확인하고 이미 처리되었다면 별다른 로직을 수행하지 않고 SQS로 ACK을 보내면 될 것 같습니다.

interface SQSEvent {
    val eventId: String

    fun toRequest(): PublishRequest

    fun toJson(): String {
        return objectMapper.writeValueAsString(this)
    }

    companion object {
        val objectMapper: ObjectMapper = ObjectMapper()
            .registerKotlinModule()
            .registerModule(JavaTimeModule())
            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    }
}

 

 

SNS에 발행되는 Topic들은 기본적으로 제가 별도로 구성한 SQSEvent 인터페이스를 상속하고 있으며 eventId 필드를 가지게 됩니다.

 

하지만 하나의 Topic을 발행해서 여러 개의 SQS로 BroadCast하는 형태이기 때문에 문제 결제 이벤트의 경우

통계 처리 SQS, 나의 문제집 추가 SQS, 장바구니 정리 SQS에 전달되는 이벤트 모두 동일한 eventId를 가지게 됩니다.

 

따라서 단순히 eventId를 멱등성 키로 선택한다면 4개 중 1개의 요청만 처리되고 나머지 3개의 요청은 이미 처리된 것으로

판정되기 때문에 약간의 수정이 필요했습니다.

@Component
@Aspect
class IdempotentEventAspect(
    private val eventProcessLogRepository: EventProcessLogRepository,
    private val transactionTemplate: TransactionTemplate,
) {
    @Around("@annotation(io.awspring.cloud.sqs.annotation.SqsListener)")
    fun processingEventIdempotency(joinPoint: ProceedingJoinPoint) {
        val event = joinPoint.args.first { it is SQSEvent } as SQSEvent
        val idempotentKey = event.eventId + "-" + joinPoint.signature.name

        if (eventProcessLogRepository.existsByIdempotentKey(idempotentKey)) {
            return
        }

        transactionTemplate.execute {
            joinPoint.proceed()
            eventProcessLogRepository.save(EventProcessLog.create(idempotentKey))
        }
    }
}

 

저의 경우 SqsListener 어노테이션이 사용 된 메서드를 대상으로 AOP를 이용해서 처리하였습니다.

joinPoint로부터 SQSEvent Type의 파라미터를 받아와 event 객체를 불러옵니다.

이후 eventId와 listener의 메서드 이름을 이용하여 멱등성 키를 만들었습니다.

 

이와 같은 방식으로 멱등성 키를 생성하면 문제 결제 이벤트처럼 동일한 eventId로 broadcast 되더라도

SQS 소비자마다 고유한 멱등성 키를 생성할 수 있습니다.

 

DB에 해당 멱등성 키가 존재하는 지 확인하고 존재한다면 return을 통해 함수를 종료하고

존재하지 않다면 proceed를 통해 비즈니스 로직을 처리합니다.

이후 해당 멱등성 키는 처리되었음을 남기기 위해 DB에 저장하는 과정으로 마무리 됩니다.

 

추가로 proceed(비즈니스 로직)과 멱등성 키를 DB에 저장하는 로직을 하나의 트랜젝션으로 일단 묶었습니다.

이것이 제가 멱등성 키를 비교적 효율적이고 빠른 Redis에 저장하지 않은 이유인데요.

 

일반적으로 멱등성 키를 RDBMS로 저장하는 경우 비교적 이벤트처리 마다 DB에 접근해야하는 것이 번거로울 수 있고 성능상 문제가

발생 할 수도 있다고 생각합니다.

 

하지만 여러가지 예외 상황을 생각하는 습관을 가지게되면서 Redis로 멱등성 키를 관리한다면 발생할 수 있는 상황을 생각해보았습니다.

 

proceed(이벤트 처리)는 되었지만 Redis에 멱등성 키가 저장되지 않는 경우 어떻게 될까? 라는 생각을 해보았고

이렇게 되면 데이터는 변경되었지만 멱등성 키가 저장되지 않았기 때문에 결론적으로 멱등성을 보장할 수 없다는 의미였습니다.

 

따라서 이벤트 처리 비즈니스 로직과 멱등성 키를 저장하는 것은 하나의 원자성으로 보장되어야 한다는 뜻이였고

RDBMS와 Redis를 하나의 원자성으로 묶기 위해서는 너무 복잡할 것 같아 일단은 기존에 사용중인 RDBMS에서

멱등성 키를 관리하도록 하였습니다.

마무리

이렇게 몇몇 기존 로직을 이벤트 처리방식으로 전환하면서 발생할 수 있는 문제를 계속해서 보완하는 과정을 거치고 있는데요.

아직까지도 어딘가 미숙해 보이고 제대로 처리되고 있는 것은 맞는걸 까 계속 확인하게 되는 것 같습니다.

꾸준히 여러 기술 블로그나 레퍼런스를 보면서 안정적인 이벤트 처리 방식이 되도록 공부하고 도입해봐야겠습니다.

 

참고

AWS SQS를 도입하면서 했던 고민들

 

멱등성이 뭔가요? | 토스페이먼츠 개발자센터