[refactor] @Async 풀 분리, 거부정책, MDC 전파로 비동기 작업 격리

src/main/java/com/matzip/admin/event/RequestReviewEventListener.java

@@ -3,6 +3,7 @@
 import com.matzip.common.analytics.AnalyticsRecorder;
 import com.matzip.common.analytics.domain.EventType;
 import com.matzip.common.analytics.domain.TargetType;
+import com.matzip.common.config.AsyncConfiguration;
 import com.matzip.common.exception.BusinessException;
 import com.matzip.common.exception.code.ErrorCode;
 import com.matzip.lottery.service.LotteryEventService;
@@ -28,7 +29,7 @@ public RequestReviewEventListener(LotteryEventService lotteryEventService, Place
         this.analyticsRecorder = analyticsRecorder;
     }
 
-    @Async
+    @Async(AsyncConfiguration.GENERAL_EXECUTOR)
     @TransactionalEventListener(condition = "#event.reviewStatus().name() == 'APPROVED'")
     public void enterCurrentEventOnApproval(RequestReviewEvent event) {
         Place place = placeRepository.findById(event.placeId())

src/main/java/com/matzip/common/analytics/AnalyticsEventListener.java

@@ -2,6 +2,7 @@
 
 import com.matzip.common.analytics.domain.entity.EventLog;
 import com.matzip.common.analytics.repository.EventLogRepository;
+import com.matzip.common.config.AsyncConfiguration;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
@@ -24,7 +25,7 @@ public AnalyticsEventListener(EventLogRepository eventLogRepository) {
         this.eventLogRepository = eventLogRepository;
     }
 
-    @Async
+    @Async(AsyncConfiguration.GENERAL_EXECUTOR)
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
     public void on(AnalyticsEvent event) {
         try {

src/main/java/com/matzip/common/config/AsyncConfiguration.java

@@ -1,30 +1,64 @@
 package com.matzip.common.config;
 
-import org.springframework.boot.task.ThreadPoolTaskExecutorBuilder;
+import com.matzip.common.logging.MdcTaskDecorator;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.core.task.TaskExecutor;
+import org.springframework.core.task.TaskDecorator;
 import org.springframework.scheduling.annotation.AsyncConfigurer;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import java.util.concurrent.Executor;
-
+import java.util.concurrent.RejectedExecutionHandler;@Slf4j
 @EnableAsync
 @Configuration
 public class AsyncConfiguration implements AsyncConfigurer {
 
+    public static final String GENERAL_EXECUTOR = "generalExecutor";
+    public static final String EXTERNAL_EXECUTOR = "externalExecutor";
+
+    private static final int SHUTDOWN_AWAIT_SECONDS = 15;
+
     @Override
     public Executor getAsyncExecutor() {
-        return asyncTaskExecutor();
+        return generalExecutor();
+    }
+
+    @Bean(GENERAL_EXECUTOR)
+    public ThreadPoolTaskExecutor generalExecutor() {
+        // Hikari 기본 커넥션 풀(10)을 웹 요청과 나눠 쓰도록 max 4로 제한
+        return buildExecutor("async-", 3, 4, 200, loggingDiscardPolicy(GENERAL_EXECUTOR));
+    }
+
+    @Bean(EXTERNAL_EXECUTOR)
+    public ThreadPoolTaskExecutor externalExecutor() {
+        return buildExecutor("ext-", 1, 2, 30, loggingDiscardPolicy(EXTERNAL_EXECUTOR));
+    }
+
+    private ThreadPoolTaskExecutor buildExecutor(String threadNamePrefix, int corePoolSize, int maxPoolSize,
+                                                 int queueCapacity, RejectedExecutionHandler rejectedHandler) {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(corePoolSize);
+        executor.setMaxPoolSize(maxPoolSize);
+        executor.setQueueCapacity(queueCapacity);
+        executor.setThreadNamePrefix(threadNamePrefix);
+        executor.setRejectedExecutionHandler(rejectedHandler);
+        executor.setTaskDecorator(mdcTaskDecorator());
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.setAwaitTerminationSeconds(SHUTDOWN_AWAIT_SECONDS);
+        executor.initialize();
+        return executor;
+    }
+
+    private TaskDecorator mdcTaskDecorator() {
+        return new MdcTaskDecorator();
     }
 
-    @Bean
-    public TaskExecutor asyncTaskExecutor() {
-        return new ThreadPoolTaskExecutorBuilder()
-                .corePoolSize(10)
-                .maxPoolSize(50)
-                .queueCapacity(100)
-                .threadNamePrefix("Async-")
-                .build();
+    // ViewCountService가 REQUIRES_NEW라 CallerRunsPolicy는 커넥션 데드락 위험이 있어 log & discard 사용
+    private RejectedExecutionHandler loggingDiscardPolicy(String poolName) {
+        return (rejected, executor) ->
+                log.warn("[비동기 작업 유실] {}} 풀 포화로 작업을 버립니다. activeThreads={}, queueSize={}",
+                        poolName, executor.getActiveCount(), executor.getQueue().size());
     }
 }

src/main/java/com/matzip/common/infra/discord/DiscordWebhookSender.java

@@ -1,5 +1,6 @@
 package com.matzip.common.infra.discord;
 
+import com.matzip.common.config.AsyncConfiguration;
 import com.matzip.common.config.DiscordWebhookProperties;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
@@ -21,7 +22,7 @@ public DiscordWebhookSender(WebClient.Builder webClientBuilder,
         this.discordWebhookProperties = discordWebhookProperties;
     }
 
-    @Async
+    @Async(AsyncConfiguration.EXTERNAL_EXECUTOR)
     public void sendAsync(String content) {
         String webhookUrl = discordWebhookProperties.getWebhookUrl();
         if (!StringUtils.hasText(webhookUrl)) {

src/main/java/com/matzip/common/logging/MdcTaskDecorator.java

@@ -0,0 +1,26 @@
+package com.matzip.common.logging;
+
+import org.slf4j.MDC;
+import org.springframework.core.task.TaskDecorator;
+
+import java.util.Map;
+
+public class MdcTaskDecorator implements TaskDecorator {
+
+    @Override
+    public Runnable decorate(Runnable runnable) {
+        Map<String, String> parentContext = MDC.getCopyOfContextMap();
+        return () -> {
+            if (parentContext != null) {
+                MDC.setContextMap(parentContext);
+            } else {
+                MDC.clear();
+            }
+            try {
+                runnable.run();
+            } finally {
+                MDC.clear(); // 풀 스레드 재사용 시 이전 요청 컨텍스트가 남지 않도록 정리
+            }
+        };
+    }
+}

src/main/java/com/matzip/place/service/ViewCountService.java

@@ -1,5 +1,6 @@
 package com.matzip.place.service;
 
+import com.matzip.common.config.AsyncConfiguration;
 import com.matzip.place.domain.DailyViewCount;
 import com.matzip.place.repository.DailyViewCountRepository;
 import com.matzip.place.repository.PlaceRepository;
@@ -18,7 +19,7 @@ public class ViewCountService {
     private final PlaceRepository placeRepository;
     private final DailyViewCountRepository dailyViewCountRepository;
 
-    @Async
+    @Async(AsyncConfiguration.GENERAL_EXECUTOR)
     @Transactional(propagation = Propagation.REQUIRES_NEW)
     public void incrementAllCounts(Long placeId) {
         placeRepository.incrementViewCount(placeId);

src/test/java/com/matzip/common/config/AsyncConfigurationTest.java

@@ -0,0 +1,37 @@
+package com.matzip.common.config;
+
+import com.matzip.AbstractMatzipApplicationTest;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AsyncConfigurationTest extends AbstractMatzipApplicationTest {
+
+    @Autowired
+    @Qualifier("generalExecutor")
+    private ThreadPoolTaskExecutor generalExecutor;
+
+    @Autowired
+    @Qualifier("externalExecutor")
+    private ThreadPoolTaskExecutor externalExecutor;
+
+    @Test
+    void 내부작업_풀과_외부IO_풀이_분리되어_등록된다() {
+        assertThat(generalExecutor).isNotSameAs(externalExecutor);
+    }
+
+    @Test
+    void generalExecutor는_Hikari_커넥션을_고려한_크기로_구성된다() {
+        assertThat(generalExecutor.getCorePoolSize()).isEqualTo(3);
+        assertThat(generalExecutor.getMaxPoolSize()).isEqualTo(4);
+    }
+
+    @Test
+    void externalExecutor는_저빈도_외부IO용_작은_크기로_구성된다() {
+        assertThat(externalExecutor.getCorePoolSize()).isEqualTo(1);
+        assertThat(externalExecutor.getMaxPoolSize()).isEqualTo(2);
+    }
+}

src/test/java/com/matzip/common/logging/MdcTaskDecoratorTest.java

@@ -0,0 +1,68 @@
+package com.matzip.common.logging;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.MDC;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MdcTaskDecoratorTest {
+
+    private static final String TRACE_ID = "traceId";
+
+    private final MdcTaskDecorator decorator = new MdcTaskDecorator();
+
+    @AfterEach
+    void clearMdc() {
+        MDC.clear();
+    }
+
+    @Test
+    void 부모_스레드의_MDC를_실행_스레드로_복사한다() throws Exception {
+        MDC.put(TRACE_ID, "abc123"); // 제출(부모) 스레드에서 데코레이트
+        AtomicReference<String> seenDuringRun = new AtomicReference<>();
+        Runnable decorated = decorator.decorate(() -> seenDuringRun.set(MDC.get(TRACE_ID)));
+
+        // 다른 스레드에서 실행 → ThreadLocal이라 데코레이터 없이는 전파되지 않음
+        runAndJoin(new Thread(decorated));
+
+        assertThat(seenDuringRun.get()).isEqualTo("abc123");
+    }
+
+    @Test
+    void 실행_후_스레드의_MDC를_정리한다() throws Exception {
+        MDC.put(TRACE_ID, "abc123");
+        Runnable decorated = decorator.decorate(() -> {});
+
+        // 실행이 끝난 뒤 워커 스레드에 값이 남으면 재사용 시 다음 작업을 오염시킨다
+        AtomicReference<String> afterRun = new AtomicReference<>("SENTINEL");
+        runAndJoin(new Thread(() -> {
+            decorated.run();
+            afterRun.set(MDC.get(TRACE_ID));
+        }));
+
+        assertThat(afterRun.get()).isNull();
+    }
+
+    @Test
+    void 부모_MDC가_없으면_워커에_남은_잔여값을_지운다() throws Exception {
+        MDC.clear(); // 부모(제출) 스레드에 컨텍스트 없음
+        AtomicReference<String> seenDuringRun = new AtomicReference<>("SENTINEL");
+        Runnable decorated = decorator.decorate(() -> seenDuringRun.set(MDC.get(TRACE_ID)));
+
+        // 재사용된 워커 스레드에 이전 작업의 값이 남아있는 상황을 모사
+        runAndJoin(new Thread(() -> {
+            MDC.put(TRACE_ID, "STALE");
+            decorated.run();
+        }));
+
+        assertThat(seenDuringRun.get()).isNull();
+    }
+
+    private void runAndJoin(Thread worker) throws InterruptedException {
+        worker.start();
+        worker.join();
+    }
+}