예외 처리의 예외 처리 (feat. RabbitMQ 지연 큐)

결제 로직 에서 발생하는 예외 처리하기(feat.트랜젝션 의존 줄이기)

지난 게시글에서는 결제 로직에서 발생할 수 있는 예외 케이스들을 정의하고 케이스마다 별도의 예외 처리 로직을 부여하여
결제 플로우를 분리하였습니다.

 

기존 예외 처리 방식

public ChargePointPayment approve(PGPayment pgPayment) {
    return lockManager.executeWithLock(
        LockKeyGenerator.generateChargePointPaymentKey(pgPayment.getPaymentId()),
        () -> {
            try {
                ChargePointPayment chargePointPayment = chargePointPaymentRepository.findByPaymentId(pgPayment.getPaymentId());
                chargePointPayment.validatePayment(pgPayment.getAmount());
                chargePointPayment.approve(pgPayment.getReceiptUrl());
                return chargePointPaymentRepository.save(chargePointPayment);
            } catch (CoreException coreException) {
                throw coreException;
            } catch (Exception unknownException) {
                chargePointPaymentFailHandler.failHandler(pgPayment.getPaymentId());
                throw unknownException;
            }
        }
    );
}
    
// ----- fail Handler ----- //
    
public void failHandler(String paymentId) {
    ChargePointPayment chargePointPayment = chargePointPaymentRepository.findByPaymentId(paymentId);
    chargePointPayment.fail();
    chargePointPaymentRepository.save(chargePointPayment);

    pgAPI.cancel(chargePointPayment.getPaymentId());
}

기존 예외 처리 로직을 보면 롤백 처리가 필요한 Exception이 발생했을 시 직접 failHandler를 호출하여 예외 처리를 하였습니다.

 

하지만 문득 저는 예외 처리 도중 또 예외가 발생하면 어떻게 처리해야하는가에 대해 의문이 들었습니다.

 

예외 처리 도중 예외

failHandler가 호출되는 경우는 결제 로직 처리 중 예상치 못한 Exception이 발생하는 경우입니다.

 

예상치 못한 Exception이란 것은 여러가지 상황이 있겠지만 저는 아래 두가지 경우를 가장 먼저 떠올리게 되었습니다.

1. 외부 PG API 장애 - 외부 API의 상태는 예상가능 한 범주가 아니며 항상 정상적이라고 보장할 수 없기 때문입니다.

2. DB 장애 - DB 서버의 장애 발생으로 DB 서버 통신 중에 예외가 발생할 수 있습니다.

 

만약 1번 2번과 같은 케이스로 인해 예외가 발생하였다고 가정했을 때 failHandler를 통해 예외 처리를 진행하면 어떻게 될까요?

 

운이 좋게도 외부 PG API 장애가 0.1초도 안되어서 복구가 되었다고 하면 PG API 요청을 통해 결제 취소를 하고 정상적으로 
사용자에게 다시 결제 요청을 할 수 있을 것 같습니다.

 

그러나 만약 외부 PG API가 복구 되는데 오랜 시간이 소요되었다고 가정한다면 failHandler에서
PG API를 이용해 결제 취소를 요청하는 부분에서 또 예외가 발생할 것입니다.

 

DB 장애의 경우도 마찬가지입니다. DB서버가 바로 복구되었다고 가정하면 failHandler 내부에서 바로 결제 상태를 fail로 전환하고
결제 정보를 DB에 반영할 수 있겠지만 DB서버 복구에 오랜시간이 소요되고 있다면
결제 정보를 반영하는 save메서드에서 또 예외가 발생할 것입니다.

 

예외 처리 도중 예외 발생 시 대응 방안

spring-retry

GitHub - spring-projects/spring-retry

 

스프링에서는 spring-retry라는 재시도 처리를 도와주는 기능이 있습니다.

@Service
class Service {
    @Retryable(maxAttempts=12, backoff=@Backoff(delay=100, maxDelay=500))
    public service() {
        // ... do something
    }
}

 

간단하게 어노테이션을 사용하면 예외가 발생 할 경우 몇번 재시도 할 건지, 어느 간격으로 재시도를 할 지 정할 수 있습니다.

@Retryable(maxAttempts=12, backoff=@Backoff(delay=100, maxDelay=500))
public void failHandler(String paymentId) {
    ChargePointPayment chargePointPayment = chargePointPaymentRepository.findByPaymentId(paymentId);
    chargePointPayment.fail();
    chargePointPaymentRepository.save(chargePointPayment);

    pgAPI.cancel(chargePointPayment.getPaymentId());
}

 

따라서 failHandler에 spring-retry를 적용 시키면 예외 처리 실패 시 예외를 처리 할 수 있을 것 같습니다.

 

그런데 재시도 처리를 위해 failHandler를 처리하던 기존 쓰레드가 block 되면서 대기하는 건지
일정 시간이 지나면 새로운 쓰레드가 호출되어 재시도를 처리하는지 궁금해졌는데요.

 

@Retryable(maxAttempts = 12)
public void retryMethod() {
    System.out.println("Thread ID: " + Thread.currentThread().getId());
    if (true) {
        throw new RuntimeException();
    }
}

// -------------------- //

@Test
void retryTest() {
    chargePointService.retryMethod();
}

// ------------------- // 

// Console

Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1
Thread ID: 1

 

동일한 쓰레드가 호출되어 처리가 되고 있음을 확인할 수 있었습니다.

 

어찌보면 당연한 것이 재시도를 하여 응답을 주려면 쓰레드는 재시도 처리가 성공될 때 까지 block이 되어야 합니다.

 

그렇다면 만약 외부 API 장애 및 DB 장애로 인해 모든 요청이 전부 retry를 처리하고 있다면
모든 쓰레드는 재시도 처리를 위해 쓰레드를 점유하고 있을 것이고

그동안 애플리케이션 서버 자체는 요청을 받지 못하는 상황이 될 것 입니다.

 

또한 결제 실패 예외 처리 경우에는 굳이 결제 실패의 예외 처리를 전부 완료하고 응답을 줄 필요없이

예외가 발생했을 경우 사용자에게 결제가 실패하였다고 응답을 주고 비동기적으로 결제 실패 처리를 해도 문제가 없다고 판단했습니다.

 

메시징 기반 예외 처리

재시도 처리를 할 때 효율적으로 처리하는 방법은 실패할 때 마다 지연시간을 주어 처리하는 방법일 것입니다.

그렇다면 예외가 발생하였다면 어딘가에 실패 로그를 보관해두고 특정 시간이 지날 때 마다 꺼내어 처리하면 될 것 같은데요.

 

데이터베이스를 이용해 실패 로그를 보관해 배치 프로그램을 작성하여 처리하는 방법과
메시징 큐에 예외 메시지를 발행 해 구독하는 방법이 있을 것 같습니다.

 

배치 프로그램을 작성하여 처리하는 경우는 예외 처리 로직뿐만 아니라 데이터를 특정 개수만큼
불러오는 로직을 작성해야하는 번거로움이 있을 것으로 예상됩니다.

메시징 큐의 경우는 Listener 설정을 해주면 메시지가 발행될 때 별도 처리없이 수신이 가능하기 때문에
예외 처리에만 집중하면 될 것 같았습니다.

 

결론적으로 저는 메시징 큐를 이용한 방식을 통해 예외 처리를 진행하기로 결정하였습니다.

 

제가 생각하는 가장 이상적인 처리 방식이라고 생각한 플로우입니다.

예외 발생 시 메시징 큐로 예외를 전송하고 예외 처리 Listener가 예외를 처리합니다.
그런데 예외 처리 중 예외가 발생한다면 다시 메시징 큐로 예외를 전송하는데 지연시간을 주어 메시지를 수신할 수 있도록 합니다.

 

위와 같은 방식을 이용해 처리하면 예외 처리에 성공할 때 까지 쓰레드를 점유하지 않고 처리할 수 있을 것 같습니다.

 

Rabbit MQ Delay Queue (지연 큐)

RabbitMQ를 통해 메시징 큐를 처리하기로 하였고 여러 자료를 찾아본 결과
RabbitMQ의 DelayMessage Plugin을 사용하는 방법과 직접 Delay Queue를 구현하는 방법 두가지가 있었습니다.

 

GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ

 

기본으로 제공되는 플러그인이 아니기 때문에 별도로 설치해줘야하는데요.

저는 현재 RabbitMQ를 CloudAMQP를 통해 프리티어 버전으로 사용중인데 프리티어 버전에선 플러그인 설치가 지원되지 않아
사용할 수 없었습니다.

 

또한 플러그인의 경우 여러가지 제약이 있는데요.

 

1. 지연 된 메시지는 별도의 저장소 저장되기 때문에 RabbitMQ상에서 어느 메시지가 현재 대기중인지 확인할 방법이 없음

2. 대량의 메시지를 Delay 처리하기엔 아직 다듬어지지 않은 상태

3. Delay 메시지는 RAM에 보관되기 때문에 유실가능성이 있음

 

따라서 저는 직접 Delay Queue를 구현해야 했습니다.

 

RabbitMQ delayed messages without delayed plugin

위 블로그 포스팅에서 플러그인 없이 지연 큐 구현 방법을 참고하여 진행 해보았습니다.

Delay Queue가 동작하는 방식은 위와 같습니다.

 

1. 만약 포인트 충전 결제 예외 처리를 지연 메시지로 발행하고 싶을 경우 routing-key를 최종 목적지 queue인 fail-charge-point로
지정하고 Message의 속성 expiration(delay)를 1초(1000)로 지정합니다. 해당 메시지를 Delay Queue로 매핑시켜주는
Fanout Exchange에 발행합니다.

2. 발행한 메시지는 Delay Queue에 도착합니다.

 

3. Delay Queue는 소비자가 아무도 없기 때문에 도착한 메시지들은 가만히 있게 됩니다..
그리고 Expiration을 1초로 설정했기 때문에 1초가 지나면 메시지는 dead 하게 됩니다.

Delay Queue의 Dead-Letter-Exchange는 A Topic Exchange이므로 A Topic Exchange로 전달이 됩니다.

 

4. A Topic Exchange는 메시지의 routing-key를 확인합니다. 해당 메시지의 routing-key는 fail-charge-point이기 떄문에
최종 목적지인 fail-charge-point Queue에 도착합니다.

 

5. 예외를 처리하는 Message Listener(Spring Application)은 메시지를 수신하고 예외 처리를 진행합니다.

 

6. 만약 아직까지 예상치 못한 예외가 계속 발생한다면 다시 Fanout Exchnage로 지연 메시지를 발행하며 위 과정을 반복합니다.

 

Fanout Exchange를 이용해 별도의 지정없이 바로 Delay Queue에 메시지를 발행할 수 있었습니다.
덕분에 최종 목적지인 fail-charge-point를 routing-key에 명시할 수 있으며
이후 Dead-Letter-Exchange(A-Topic-Exchange)에서는 routing-key를 통해 원하는 목적지로 메시지를
일정 시간 지연을 주어 발행할 수 있었습니다.

지연 큐를 활용 해 예외 처리 하기 

// RabbitConfig.java

// queue 

@Bean
public Queue failChargePointQueue() {
    return QueueBuilder.durable(FAIL_CHARGE_POINT_QUEUE)
        .build();
}

@Bean
public Queue delayQueue() {
    return QueueBuilder.durable(DELAY_QUEUE)
        .deadLetterExchange(EXCHANGE)
        .build();
}


// exchange

@Bean
public TopicExchange exchange() {
    return new TopicExchange(EXCHANGE);
}

@Bean
public FanoutExchange delayQueueFanoutExchange() {
    return new FanoutExchange(DELAY_EXCHANGE);
}


// binding

@Bean
public Binding bindingFailChargePoint(Queue failChargePointQueue, TopicExchange exchange) {
    return BindingBuilder.bind(failChargePointQueue).to(exchange).with(FAIL_CHARGE_POINT_QUEUE);
}

@Bean
public Binding bindingDelay(Queue delayQueue, FanoutExchange delayQueueFanoutExchange) {
    return BindingBuilder.bind(delayQueue).to(delayQueueFanoutExchange);
}


// ------ //

 

저는 비교적 복잡한 설정이 없기 때문에 @Configuration를 사용하는 RabbitConfig에 Queue, Exchange, Binding을 정의하여
자동으로 Spring Application 실행 시 RabbitMQ에 반영되도록 하였습니다.

 

@Transactional
public ChargePointPayment approve(PGPayment pgPayment) {
    try {
        ChargePointPayment chargePointPayment = chargePointPaymentRepository.findByPaymentIdWithLock(pgPayment.getPaymentId());
        chargePointPayment.validatePayment(pgPayment.getAmount());
        chargePointPayment.approve(pgPayment.getReceiptUrl());
        return chargePointPaymentRepository.save(chargePointPayment);
    } catch (CoreException coreException) {
        throw coreException;
    } catch (Exception unknownException) {
        messageSender.sendMessage(MessageType.FAIL_CHARGE_POINT, FailChargePointPaymentMessage.create(pgPayment.getPaymentId()));
        throw unknownException;
    }
}

 

기존에는 Excepction Catch에서 바로 failHandler(예외 처리 로직)을 호출했지만 messageSender를 통해 메시지를 발행합니다.

첫 예외는 바로 failHandler를 호출해도 되겠지만 굳이 예외 처리가 완료될 때 까지 사용자는 기다릴 필요가 없으므로
비동기적으로 예외를 처리하기 위해 메시지 발행 후 응답을 주는 방향으로 결정했습니다.

@RabbitListener(id = "fail-charge-point", queues = "fail-charge-point")
public void failHandler(FailChargePointPaymentMessage message) {
    try {
        ChargePointPayment chargePointPayment = chargePointPaymentRepository.findByPaymentId(message.getPaymentId());
        chargePointPayment.fail();
        chargePointPaymentRepository.save(chargePointPayment);

        pgPaymentProcessor.cancel(chargePointPayment.getPaymentId());
    } catch (Exception e) {
        message.increaseFailCount();
        messageSender.sendDelayMessage(MessageType.FAIL_CHARGE_POINT, message, message.getFailCount());
    }
}

 

기존에 예외를 처리했던 failHandler는 FailChargePointPaymentMessage를 수신하는 Listener가 되었습니다.

예외 처리중 발생하는 예외 처리를 대비하기 위해 failHandler 로직은 try catch로 감싸져있습니다.

catch문에서는 예외 처리 중 예외가 발생했을 경우 sendDelayMessage를 통해 지연 메시지를 전송하도록 하였습니다.

 

@Getter
@AllArgsConstructor
@NoArgsConstructor
public class FailChargePointPaymentMessage {
    private int failCount;
    private String paymentId;

    public static FailChargePointPaymentMessage create(String paymentId) {
        return new FailChargePointPaymentMessage(0, paymentId);
    }

    public void increaseFailCount() {
        this.failCount++;
    }
}

 

failCount의 경우는 실패 횟수에 따라 지연 시간을 다르게 주기 위해 생성한 필드입니다.

 

// MessageSender

public void sendDelayMessage(MessageType messageType, Object message, int failCount) {
    rabbitTemplate.convertAndSend(
        RabbitConfig.DELAY_EXCHANGE,
        messageType.getQueueName(),
        message,
        m -> {
            m.getMessageProperties().setExpiration(selectDelay(failCount));
            return m;
        });
}

private String selectDelay(int failCount) {
    switch (failCount) {
        default:
            return "10000";
    }
}

 

sendDelayMessage 메서드입니다. rabbitTemplate는 오버로딩 된 converAndSend 메서드 굉장히 많은데요.

 

위 코드가 사용해야할 convertAndSend 메서드입니다.

 

첫번째 인자로는 DelayQueue로 전송하는 Fanout Exchange

두번째 인자로는 최종 목적지 Queue의 routing key (Queue 이름)

세번째 인자로는 메시지 객체

네번째 인자로는 MessagePostProcessor입니다.

 

MessagePostProcessor의 경우 Pojo 객체를 메시지로 바로 발행할 때 쉽게 Message 속성을 설정할 수 있게 도와줍니다.
expiration의 경우도 Message 속성이기 때문에 MessagePostProcessor로 쉽게 설정할 수 있습니다.

 

아직 실패횟수에 따른 지연 시간을 별도로 구성하지 않았기 때문에 임시로 10000 (10초)로 설정하였습니다.

 

이렇게 기본적인 지연 큐를 활용 한 예외 처리 로직 구성도 마무리되었는데요. 한번 제대로 처리 되는지 확인해보겠습니다.

 

@Test
public void testSend() {
    messageSender.sendDelayMessage(MessageType.FAIL_CHARGE_POINT, FailChargePointPaymentMessage.create("qwer1qwer"), 0);
}

 

임의로 지연 메시지를 발행해보겠습니다.

 

 

RabbitMQ 웹 콘솔에서 확인해보면 delay queue에 메시지가 도착했음을 볼 수 있었습니다.

 

 

10초 이후 다시 Queue를 조회하면 비어있다고 나오게 됩니다.

 

이후 최종 목적지 Queue였던 fail-charge-point를 확인해보면 지연 큐에서 메시지가 이동되었음을 확인할 수 있습니다.

 

이제 Spring Application을 통해 메시지를 수신하고 임의로 예외처리 도중 예외를 발생시켜보겠습니다.

 

catch (Exception e) {
    message.increaseFailCount();
    System.out.println("message fail count: " + message.getFailCount());
    messageSender.sendDelayMessage(MessageType.FAIL_CHARGE_POINT, message, message.getFailCount());
}

 

로깅을 위해 임시로 메시지를 수신할 때 마다 failCount를 확인해보았습니다.

10초마다 메시지가 지연되어 발행되는 것을 볼 수 있었습니다.

 

 

이후 생각해볼 점

이렇게 결제 프로세스에서 예외 처리 도중 예외가 발생하면 어쩌지? 라는 의문 덕분에
결제 실패라는 사실을 손실시키지 않고 끝까지 처리하기 위한 방법을 고민해보았는데요.

 

최종적으로 마지막 고려할 점이 있는 것 같습니다.

 

DB 혹은 외부 API 장애가 발생했을 때 최악의 상황으로
Rabbit MQ도 장애가 발생하여 메시지가 발행되지 않거나 손실되는 경우입니다.

 

이런 경우는 흔하게 RabbitMQ 서버를 한 대로 처리하는 것이 아닌 여러 대의 서버를 두어 Master 서버에 장애가 발생하면
다른 slave 서버를 Master 서버로 위임하여 처리하는 방법이 있을 것 같습니다.

제 개인 프로젝트는 포인트 충전 결제 뿐만 아니라 서비스 재화를 이용해 상품을 구매하는 로직도 존재하기 때문에

상품 구매 로직에서도 지연 큐를 통한 예외 처리를 적용하고 고민해봐야겠습니다.