feat(messaging): RabbitMQ DLX·DLQ + 컨슈머 재시도 정책 (US-28)

backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java

@@ -2,6 +2,9 @@
 
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Positive;
+import jakarta.validation.constraints.PositiveOrZero;
+import java.time.Duration;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.validation.annotation.Validated;
 
@@ -19,7 +22,11 @@ public record RabbitMqProperties(
 	@NotNull
 	Queues queues,
 	@NotNull
-	RoutingKeyProperties routingKeys
+	RoutingKeyProperties routingKeys,
+	@NotNull
+	DeadLetter deadLetter,
+	@NotNull
+	Retry retry
 ) {
 
 	public record Message(
@@ -76,4 +83,18 @@ public record RoutingKeyProperties(
 		@NotBlank String realtimeSessionNotify
 	) {
 	}
+
+	public record DeadLetter(
+		@NotBlank String exchange,
+		@NotBlank String routingKeyPrefix
+	) {
+	}
+
+	public record Retry(
+		@PositiveOrZero int maxRetries,
+		@NotNull Duration initialInterval,
+		@Positive double multiplier,
+		@NotNull Duration maxInterval
+	) {
+	}
 }

backend/src/main/java/com/stackup/stackup/common/messaging/RabbitMqConfig.java

@@ -1,12 +1,20 @@
 package com.stackup.stackup.common.messaging;
 
 import com.stackup.stackup.common.config.properties.RabbitMqProperties;
+import java.util.Map;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Declarables;
+import org.springframework.amqp.core.DirectExchange;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
 import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.context.annotation.Bean;
@@ -15,6 +23,8 @@
 @Configuration
 public class RabbitMqConfig {
 
+    static final String LISTENER_CONTAINER_FACTORY = "rabbitListenerContainerFactory";
+
     private final RabbitMqProperties properties;
 
     public RabbitMqConfig(RabbitMqProperties properties) {
@@ -48,66 +58,96 @@ public TopicExchange realtimeExchange() {
         );
     }
 
+    @Bean
+    public DirectExchange deadLetterExchange() {
+        return new DirectExchange(
+            properties.deadLetter().exchange(),
+            properties.exchanges().durable(),
+            properties.exchanges().autoDelete()
+        );
+    }
+
     @Bean
     public Queue aiAnalyzeResumeQueue() {
-        return new Queue(properties.queues().names().aiAnalyzeResume(), properties.queues().durable());
+        return workQueue(properties.queues().names().aiAnalyzeResume());
     }
 
     @Bean
     public Queue aiAnalyzeRepositoryQueue() {
-        return new Queue(properties.queues().names().aiAnalyzeRepository(), properties.queues().durable());
+        return workQueue(properties.queues().names().aiAnalyzeRepository());
     }
 
     @Bean
     public Queue aiGenerateQuestionsQueue() {
-        return new Queue(properties.queues().names().aiGenerateQuestions(), properties.queues().durable());
+        return workQueue(properties.queues().names().aiGenerateQuestions());
     }
 
     @Bean
     public Queue aiGenerateFollowupQueue() {
-        return new Queue(properties.queues().names().aiGenerateFollowup(), properties.queues().durable());
+        return workQueue(properties.queues().names().aiGenerateFollowup());
     }
 
     @Bean
     public Queue coreCallbackAnalysisQueue() {
-        return new Queue(properties.queues().names().coreCallbackAnalysis(), properties.queues().durable());
+        return workQueue(properties.queues().names().coreCallbackAnalysis());
     }
 
     @Bean
     public Queue coreCallbackQuestionsQueue() {
-        return new Queue(properties.queues().names().coreCallbackQuestions(), properties.queues().durable());
+        return workQueue(properties.queues().names().coreCallbackQuestions());
     }
 
     @Bean
     public Declarables rabbitDeclarables(
         TopicExchange coreToAiExchange,
         TopicExchange aiToCoreExchange,
         TopicExchange realtimeExchange,
+        DirectExchange deadLetterExchange,
         Queue aiAnalyzeResumeQueue,
         Queue aiAnalyzeRepositoryQueue,
         Queue aiGenerateQuestionsQueue,
         Queue aiGenerateFollowupQueue,
         Queue coreCallbackAnalysisQueue,
         Queue coreCallbackQuestionsQueue
     ) {
+        Queue dlqAiAnalyzeResume = dlq(properties.queues().names().aiAnalyzeResume());
+        Queue dlqAiAnalyzeRepository = dlq(properties.queues().names().aiAnalyzeRepository());
+        Queue dlqAiGenerateQuestions = dlq(properties.queues().names().aiGenerateQuestions());
+        Queue dlqAiGenerateFollowup = dlq(properties.queues().names().aiGenerateFollowup());
+        Queue dlqCoreCallbackAnalysis = dlq(properties.queues().names().coreCallbackAnalysis());
+        Queue dlqCoreCallbackQuestions = dlq(properties.queues().names().coreCallbackQuestions());
+
         return new Declarables(
             coreToAiExchange,
             aiToCoreExchange,
             realtimeExchange,
+            deadLetterExchange,
             aiAnalyzeResumeQueue,
             aiAnalyzeRepositoryQueue,
             aiGenerateQuestionsQueue,
             aiGenerateFollowupQueue,
             coreCallbackAnalysisQueue,
             coreCallbackQuestionsQueue,
+            dlqAiAnalyzeResume,
+            dlqAiAnalyzeRepository,
+            dlqAiGenerateQuestions,
+            dlqAiGenerateFollowup,
+            dlqCoreCallbackAnalysis,
+            dlqCoreCallbackQuestions,
             BindingBuilder.bind(aiAnalyzeResumeQueue).to(coreToAiExchange).with(properties.routingKeys().analyzeResume()),
             BindingBuilder.bind(aiAnalyzeRepositoryQueue).to(coreToAiExchange).with(properties.routingKeys().analyzeRepository()),
             BindingBuilder.bind(aiGenerateQuestionsQueue).to(coreToAiExchange).with(properties.routingKeys().generateQuestions()),
             BindingBuilder.bind(aiGenerateFollowupQueue).to(coreToAiExchange).with(properties.routingKeys().generateFollowup()),
             BindingBuilder.bind(coreCallbackAnalysisQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackAnalysis()),
-            BindingBuilder.bind(coreCallbackQuestionsQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackQuestions())
-            // 주의: q.realtime.session.notify 큐 + binding 은 RealTime 서버가 자체 declare
-            // (또는 infra/rabbitmq/definitions.json). Core 는 exchange 만 declare.
+            BindingBuilder.bind(coreCallbackQuestionsQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackQuestions()),
+            BindingBuilder.bind(dlqAiAnalyzeResume).to(deadLetterExchange).with(dlqAiAnalyzeResume.getName()),
+            BindingBuilder.bind(dlqAiAnalyzeRepository).to(deadLetterExchange).with(dlqAiAnalyzeRepository.getName()),
+            BindingBuilder.bind(dlqAiGenerateQuestions).to(deadLetterExchange).with(dlqAiGenerateQuestions.getName()),
+            BindingBuilder.bind(dlqAiGenerateFollowup).to(deadLetterExchange).with(dlqAiGenerateFollowup.getName()),
+            BindingBuilder.bind(dlqCoreCallbackAnalysis).to(deadLetterExchange).with(dlqCoreCallbackAnalysis.getName()),
+            BindingBuilder.bind(dlqCoreCallbackQuestions).to(deadLetterExchange).with(dlqCoreCallbackQuestions.getName())
+            // 주의: q.realtime.session.notify 큐 + binding (+ DLQ) 은 RealTime 서버 측에서
+            // infra/rabbitmq/definitions.json 으로 import 되므로 Core 는 exchange 만 declare.
         );
     }
 
@@ -123,4 +163,52 @@ public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Messag
         rabbitTemplate.setMandatory(properties.template().mandatory());
         return rabbitTemplate;
     }
+
+    @Bean(name = LISTENER_CONTAINER_FACTORY)
+    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
+        ConnectionFactory connectionFactory,
+        MessageConverter messageConverter
+    ) {
+        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        factory.setMessageConverter(messageConverter);
+        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
+        factory.setDefaultRequeueRejected(false);
+        factory.setAdviceChain(retryInterceptor());
+        return factory;
+    }
+
+    private MethodInterceptor retryInterceptor() {
+        RabbitMqProperties.Retry retry = properties.retry();
+        return RetryInterceptorBuilder.stateless()
+            .maxRetries(retry.maxRetries())
+            .backOffOptions(
+                retry.initialInterval().toMillis(),
+                retry.multiplier(),
+                retry.maxInterval().toMillis()
+            )
+            .recoverer(new RejectAndDontRequeueRecoverer())
+            .build();
+    }
+
+    private Queue workQueue(String name) {
+        return QueueBuilder.durable(name)
+            .withArguments(deadLetterArguments(name))
+            .build();
+    }
+
+    private Queue dlq(String workQueueName) {
+        return QueueBuilder.durable(dlqName(workQueueName)).build();
+    }
+
+    private Map<String, Object> deadLetterArguments(String workQueueName) {
+        return Map.of(
+            "x-dead-letter-exchange", properties.deadLetter().exchange(),
+            "x-dead-letter-routing-key", dlqName(workQueueName)
+        );
+    }
+
+    private String dlqName(String workQueueName) {
+        return properties.deadLetter().routingKeyPrefix() + workQueueName;
+    }
 }

backend/src/main/resources/application.yml

@@ -104,6 +104,14 @@ app:
         callback-analysis: callback.analysis
         callback-questions: callback.questions
         realtime-session-notify: realtime.session.notify
+      dead-letter:
+        exchange: stackup.dlx
+        routing-key-prefix: dlq.
+      retry:
+        max-retries: 3
+        initial-interval: 1s
+        multiplier: 2.0
+        max-interval: 10s
   sse:
     timeout: 30m
     keep-alive-interval: 30s

backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMessagePublisherTest.java

@@ -5,6 +5,7 @@
 import static org.mockito.Mockito.verify;
 
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Map;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -103,7 +104,9 @@ private RabbitMqProperties rabbitMqProperties() {
                 "callback.analysis",
                 "callback.questions",
                 "realtime.session.notify"
-            )
+            ),
+            new RabbitMqProperties.DeadLetter("stackup.dlx", "dlq."),
+            new RabbitMqProperties.Retry(3, Duration.ofSeconds(1), 2.0, Duration.ofSeconds(10))
         );
     }
 }

backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMqConfigTest.java

@@ -0,0 +1,79 @@
+package com.stackup.stackup.common.messaging;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.stackup.stackup.common.config.properties.RabbitMqProperties;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.core.Queue;
+
+class RabbitMqConfigTest {
+
+    private final RabbitMqConfig config = new RabbitMqConfig(rabbitMqProperties());
+
+    @Test
+    void workQueue_carriesDeadLetterArguments() {
+        Queue queue = config.aiAnalyzeResumeQueue();
+
+        assertThat(queue.getName()).isEqualTo("ai.analyze.resume");
+        assertThat(queue.isDurable()).isTrue();
+        assertThat(queue.getArguments())
+            .containsEntry("x-dead-letter-exchange", "stackup.dlx")
+            .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.resume");
+    }
+
+    @Test
+    void allWorkQueues_routeToDeadLetterExchange() {
+        assertThat(config.aiAnalyzeResumeQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.resume");
+        assertThat(config.aiAnalyzeRepositoryQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.repository");
+        assertThat(config.aiGenerateQuestionsQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.ai.generate.questions");
+        assertThat(config.aiGenerateFollowupQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.ai.generate.followup");
+        assertThat(config.coreCallbackAnalysisQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.core.callback.analysis");
+        assertThat(config.coreCallbackQuestionsQueue().getArguments())
+            .containsEntry("x-dead-letter-routing-key", "dlq.core.callback.questions");
+    }
+
+    private RabbitMqProperties rabbitMqProperties() {
+        return new RabbitMqProperties(
+            "core-server",
+            "v1",
+            new RabbitMqProperties.Message("application/json", StandardCharsets.UTF_8.name(), "x-trace-id"),
+            new RabbitMqProperties.Template(true),
+            new RabbitMqProperties.Exchanges(
+                true,
+                false,
+                new RabbitMqProperties.Exchanges.Names(
+                    "stackup.core-to-ai", "stackup.ai-to-core", "stackup.realtime"
+                )
+            ),
+            new RabbitMqProperties.Queues(
+                true,
+                new RabbitMqProperties.Queues.Names(
+                    "ai.analyze.resume",
+                    "ai.analyze.repository",
+                    "ai.generate.questions",
+                    "ai.generate.followup",
+                    "core.callback.analysis",
+                    "core.callback.questions"
+                )
+            ),
+            new RabbitMqProperties.RoutingKeyProperties(
+                "analyze.resume",
+                "analyze.repository",
+                "generate.questions",
+                "generate.followup",
+                "callback.analysis",
+                "callback.questions",
+                "realtime.session.notify"
+            ),
+            new RabbitMqProperties.DeadLetter("stackup.dlx", "dlq."),
+            new RabbitMqProperties.Retry(3, Duration.ofSeconds(1), 2.0, Duration.ofSeconds(10))
+        );
+    }
+}