feat: 알림 전송 Retry 및 DLQ 처리 구현

docker-compose.yml

@@ -21,6 +21,8 @@ services:
       SPRING_RABBITMQ_PASSWORD: notification
       NOTIFICATION_CONSUMER_CONCURRENCY: ${NOTIFICATION_CONSUMER_CONCURRENCY:-1}
       RABBITMQ_PREFETCH: ${RABBITMQ_PREFETCH:-250}
+      MOCK_PROVIDER_FORCE_FAILURE: ${MOCK_PROVIDER_FORCE_FAILURE:-false}
+      MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS: ${MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS:-0}
 
     healthcheck:
       test:

src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessage.java

@@ -6,6 +6,17 @@ public record DeliveryMessage(
         Long notificationId,
         Long deliveryId,
         NotificationChannel channel,
-        String destination
+        String destination,
+        int attempt
 ) {
+
+    public DeliveryMessage nextAttempt() {
+        return new DeliveryMessage(
+                notificationId,
+                deliveryId,
+                channel,
+                destination,
+                attempt + 1
+        );
+    }
 }

src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java

@@ -0,0 +1,58 @@
+package com.backendsystemdesignlab.notification.messaging;
+
+import com.backendsystemdesignlab.notification.notification.provider.EmailProvider;
+import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
+import com.backendsystemdesignlab.notification.notification.provider.PushProvider;
+import com.backendsystemdesignlab.notification.notification.provider.SmsProvider;
+import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class DeliveryMessageHandler {
+
+    private static final int MAX_ATTEMPTS = 3;
+
+    private final PushProvider pushProvider;
+    private final SmsProvider smsProvider;
+    private final EmailProvider emailProvider;
+
+    private final DeliveryRetryPublisher retryPublisher;
+    private final NotificationTransactionService transactionService;
+
+    public void handle(DeliveryMessage message) {
+
+        boolean success = send(message);
+
+        if (success) {
+            transactionService.recordSuccess(message.notificationId(), message.deliveryId());
+            return;
+        }
+
+        if (message.attempt() < MAX_ATTEMPTS) {
+            DeliveryMessage retryMessage = message.nextAttempt();
+            retryPublisher.publishRetry(retryMessage);
+            transactionService.recordRetryFailure(message.deliveryId());
+            return;
+        }
+
+        retryPublisher.publishDlq(message);
+        transactionService.recordFinalFailure(message.notificationId(), message.deliveryId());
+    }
+
+    private boolean send(DeliveryMessage message) {
+
+        try {
+            ProviderResult result =
+                    switch (message.channel()) {
+                        case PUSH -> pushProvider.send(message.destination());
+                        case SMS -> smsProvider.send(message.destination());
+                        case EMAIL -> emailProvider.send(message.destination());
+                    };
+            return result.success();
+        } catch (RuntimeException e) {
+            return false;
+        }
+    }
+}

src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryPublisher.java

@@ -20,7 +20,7 @@ public void publishAll(Long notificationId, List<DeliveryCommand> deliveries) {
     }
 
     private void publish(Long notificationId, DeliveryCommand delivery) {
-        DeliveryMessage message = new DeliveryMessage(notificationId, delivery.deliveryId(), delivery.channel(), delivery.destination());
+        DeliveryMessage message = new DeliveryMessage(notificationId, delivery.deliveryId(), delivery.channel(), delivery.destination(), 1);
 
         String routingKey = switch (delivery.channel()) {
             case PUSH -> RabbitMqConfig.PUSH_ROUTING_KEY;

src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryRetryPublisher.java

@@ -0,0 +1,36 @@
+package com.backendsystemdesignlab.notification.messaging;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class DeliveryRetryPublisher {
+
+    private final RabbitTemplate rabbitTemplate;
+
+    public void publishRetry(DeliveryMessage message) {
+        rabbitTemplate.convertAndSend(
+                RabbitMqConfig.RETRY_EXCHANGE,
+                routingKey(message),
+                message
+        );
+    }
+
+    public void publishDlq(DeliveryMessage message) {
+        rabbitTemplate.convertAndSend(
+                RabbitMqConfig.DLX,
+                routingKey(message),
+                message
+        );
+    }
+
+    private String routingKey(DeliveryMessage message) {
+        return switch (message.channel()) {
+            case PUSH -> RabbitMqConfig.PUSH_ROUTING_KEY;
+            case SMS -> RabbitMqConfig.SMS_ROUTING_KEY;
+            case EMAIL -> RabbitMqConfig.EMAIL_ROUTING_KEY;
+        };
+    }
+}

src/main/java/com/backendsystemdesignlab/notification/messaging/RabbitMqConfig.java

@@ -1,9 +1,6 @@
 package com.backendsystemdesignlab.notification.messaging;
 
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.DirectExchange;
-import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.*;
 import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.context.annotation.Bean;
@@ -20,13 +17,32 @@ public class RabbitMqConfig {
     public static final String SMS_ROUTING_KEY = "notification.sms";
     public static final String EMAIL_ROUTING_KEY = "notification.email";
 
+    public static final String RETRY_EXCHANGE = "notification.retry.exchange";
+    public static final String DLX = "notification.dlx";
+    public static final String PUSH_RETRY_QUEUE = "notification.push.retry.queue";
+    public static final String SMS_RETRY_QUEUE = "notification.sms.retry.queue";
+    public static final String EMAIL_RETRY_QUEUE = "notification.email.retry.queue";
+    public static final String PUSH_DLQ = "notification.push.dlq";
+    public static final String SMS_DLQ = "notification.sms.dlq";
+    public static final String EMAIL_DLQ = "notification.email.dlq";
+
     // Exchange: 메시지를 받아서 적절한 Queue로 전달하는 라우터
     @Bean
     public DirectExchange notificationExchange() {
         return new DirectExchange(EXCHANGE, true, false); // Routing Key가 정확하게 일치하는 Queue로 보내기
         // durable: RabbitMQ 서버가 재시작되어도 Exchange 정의를 유지하겠다, autoDelete: Consumer 등이 없어지면 Exchange 자동 삭제
     }
 
+    @Bean
+    public DirectExchange retryExchange() {
+        return new DirectExchange(RETRY_EXCHANGE, true, false);
+    }
+
+    @Bean
+    public DirectExchange deadLetterExchange() {
+        return new DirectExchange(DLX, true, false);
+    }
+
     @Bean
     public Queue pushQueue() {
        return new Queue(PUSH_QUEUE, true);
@@ -42,6 +58,57 @@ public Queue emailQueue() {
         return new Queue(EMAIL_QUEUE, true);
     }
 
+    @Bean
+    public Queue pushRetryQueue() {
+        return QueueBuilder
+                .durable(PUSH_RETRY_QUEUE)
+                .ttl(5000) // 5초 동안 Retry Queue에 보관
+                .deadLetterExchange(EXCHANGE) // 시간이 지나면 다시 원래 Exchange
+                .deadLetterRoutingKey(PUSH_ROUTING_KEY) // PUSH Queue로 돌아가도록
+                .build();
+    }
+
+    @Bean
+    public Queue smsRetryQueue() {
+        return QueueBuilder
+                .durable(SMS_RETRY_QUEUE)
+                .ttl(5000)
+                .deadLetterExchange(EXCHANGE)
+                .deadLetterRoutingKey(SMS_ROUTING_KEY)
+                .build();
+    }
+
+    @Bean
+    public Queue emailRetryQueue() {
+        return QueueBuilder
+                .durable(EMAIL_RETRY_QUEUE)
+                .ttl(5000)
+                .deadLetterExchange(EXCHANGE)
+                .deadLetterRoutingKey(EMAIL_ROUTING_KEY)
+                .build();
+    }
+
+    @Bean
+    public Queue pushDlq() {
+        return QueueBuilder
+                .durable(PUSH_DLQ)
+                .build();
+    }
+
+    @Bean
+    public Queue smsDlq() {
+        return QueueBuilder
+                .durable(SMS_DLQ)
+                .build();
+    }
+
+    @Bean
+    public Queue emailDlq() {
+        return QueueBuilder
+                .durable(EMAIL_DLQ)
+                .build();
+    }
+
     @Bean
     public Binding pushBinding(DirectExchange notificationExchange, Queue pushQueue) {
         // pushQueue를 notificationExchange에 연결하고, notification.push라는 Routing Key로 연결해라
@@ -67,6 +134,54 @@ public Binding emailBinding(DirectExchange notificationExchange, Queue emailQueu
                 .with(EMAIL_ROUTING_KEY);
     }
 
+    @Bean
+    public Binding pushRetryBinding(DirectExchange retryExchange, Queue pushRetryQueue) {
+        return BindingBuilder
+                .bind(pushRetryQueue)
+                .to(retryExchange)
+                .with(PUSH_ROUTING_KEY);
+    }
+
+    @Bean
+    public Binding smsRetryBinding(DirectExchange retryExchange, Queue smsRetryQueue) {
+        return BindingBuilder
+                .bind(smsRetryQueue)
+                .to(retryExchange)
+                .with(SMS_ROUTING_KEY);
+    }
+
+    @Bean
+    public Binding emailRetryBinding(DirectExchange retryExchange, Queue emailRetryQueue) {
+        return BindingBuilder
+                .bind(emailRetryQueue)
+                .to(retryExchange)
+                .with(EMAIL_ROUTING_KEY);
+    }
+
+    @Bean
+    public Binding pushDlqBinding(DirectExchange deadLetterExchange, Queue pushDlq) {
+        return BindingBuilder
+                .bind(pushDlq)
+                .to(deadLetterExchange)
+                .with(PUSH_ROUTING_KEY);
+    }
+
+    @Bean
+    public Binding smsDlqBinding(DirectExchange deadLetterExchange, Queue smsDlq) {
+        return BindingBuilder
+                .bind(smsDlq)
+                .to(deadLetterExchange)
+                .with(SMS_ROUTING_KEY);
+    }
+
+    @Bean
+    public Binding emailDlqBinding(DirectExchange deadLetterExchange, Queue emailDlq) {
+        return BindingBuilder
+                .bind(emailDlq)
+                .to(deadLetterExchange)
+                .with(EMAIL_ROUTING_KEY);
+    }
+
     @Bean
     public MessageConverter messageConverter() {
         return new JacksonJsonMessageConverter();

src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/EmailDeliveryConsumer.java

@@ -1,10 +1,8 @@
 package com.backendsystemdesignlab.notification.messaging.consumer;
 
 import com.backendsystemdesignlab.notification.messaging.DeliveryMessage;
+import com.backendsystemdesignlab.notification.messaging.DeliveryMessageHandler;
 import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig;
-import com.backendsystemdesignlab.notification.notification.provider.EmailProvider;
-import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
-import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
@@ -13,29 +11,13 @@
 @RequiredArgsConstructor
 public class EmailDeliveryConsumer {
 
-    private final EmailProvider emailProvider;
-    private final NotificationTransactionService transactionService;
+    private final DeliveryMessageHandler messageHandler;
 
     @RabbitListener(
             queues = RabbitMqConfig.EMAIL_QUEUE,
             concurrency = "${NOTIFICATION_CONSUMER_CONCURRENCY:1}"
     )
     public void consume(DeliveryMessage message) {
-
-        boolean success;
-
-        try {
-            ProviderResult result = emailProvider.send(message.destination());
-            success = result.success();
-        } catch (RuntimeException e) {
-            success = false;
-        }
-
-        transactionService.recordDeliveryResult(
-                message.notificationId(),
-                message.deliveryId(),
-                success
-        );
+        messageHandler.handle(message);
     }
-
 }

src/main/java/com/backendsystemdesignlab/notification/messaging/consumer/PushDeliveryConsumer.java

@@ -1,10 +1,8 @@
 package com.backendsystemdesignlab.notification.messaging.consumer;
 
 import com.backendsystemdesignlab.notification.messaging.DeliveryMessage;
+import com.backendsystemdesignlab.notification.messaging.DeliveryMessageHandler;
 import com.backendsystemdesignlab.notification.messaging.RabbitMqConfig;
-import com.backendsystemdesignlab.notification.notification.provider.ProviderResult;
-import com.backendsystemdesignlab.notification.notification.provider.PushProvider;
-import com.backendsystemdesignlab.notification.notification.service.NotificationTransactionService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
@@ -13,29 +11,12 @@
 @RequiredArgsConstructor
 public class PushDeliveryConsumer {
 
-    private final PushProvider pushProvider;
-    private final NotificationTransactionService transactionService;
-
+    private final DeliveryMessageHandler messageHandler;
     @RabbitListener(
             queues = RabbitMqConfig.PUSH_QUEUE,
             concurrency = "${NOTIFICATION_CONSUMER_CONCURRENCY:1}"
     )
     public void consume(DeliveryMessage message) {
-
-        boolean success;
-
-        try {
-            ProviderResult result = pushProvider.send(message.destination());
-            success = result.success();
-        } catch (RuntimeException e) {
-            success = false;
-        }
-
-        transactionService.recordDeliveryResult(
-                message.notificationId(),
-                message.deliveryId(),
-                success
-        );
+        messageHandler.handle(message);
     }
-
 }