Skip to content

비동기 Executor 안정화 및 AI 연동 실패 처리 전환 #257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ckdals4600 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

비동기 Executor 안정화 및 AI 연동 실패 처리 전환 #257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d81410d
feat: @Async Executor에 거부 정책 및 graceful shutdown 추가 (#256)
ckdals4600 Jun 30, 2026
0e71134
refactor: AI 응답 실패를 null 반환에서 typed exception으로 (#256)
ckdals4600 Jun 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 20 additions & 14 deletions src/main/java/com/sofa/linkiving/domain/chat/ai/RagAnswerClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@

import com.sofa.linkiving.domain.chat.dto.request.RagAnswerReq;
import com.sofa.linkiving.domain.chat.dto.response.RagAnswerRes;
import com.sofa.linkiving.global.error.exception.BusinessException;
import com.sofa.linkiving.infra.feign.EmptyAiResponseException;
import com.sofa.linkiving.infra.feign.ExternalApiErrorCode;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
Expand Down Expand Up @@ -43,23 +46,26 @@ private Counter buildCounter(String result) {

@Override
public RagAnswerRes generateAnswer(RagAnswerReq request) {
List<RagAnswerRes> ragAnswerRes;
try {
List<RagAnswerRes> ragAnswerRes = ragAnswerFeign.generateAnswer(request);

if (ragAnswerRes == null || ragAnswerRes.isEmpty()) {
log.warn("RagAnswerClient generateAnswer empty response");
emptyCounter.increment();
return null;
}

log.info("RagAnswerClient generateAnswer ragAnswerRes={}", ragAnswerRes);
successCounter.increment();
return ragAnswerRes.get(0);

ragAnswerRes = ragAnswerFeign.generateAnswer(request);
} catch (BusinessException e) {
failureCounter.increment();
log.warn("[AI Server] generateAnswer failed - code={}", e.getErrorCode().getCode());
throw e;
} catch (Exception e) {
log.error("RagAnswerClient generateAnswer error", e);
failureCounter.increment();
return null;
log.warn("[AI Server] generateAnswer failed - reason={}", e.getMessage());
throw new BusinessException(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
}

if (ragAnswerRes == null || ragAnswerRes.isEmpty()) {
emptyCounter.increment();
log.warn("[AI Server] generateAnswer empty response");
throw new EmptyAiResponseException();
}

successCounter.increment();
return ragAnswerRes.get(0);
}
}
65 changes: 37 additions & 28 deletions src/main/java/com/sofa/linkiving/domain/link/ai/RagSummaryClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@
import com.sofa.linkiving.domain.link.dto.request.RagRegenerateSummaryReq;
import com.sofa.linkiving.domain.link.dto.response.RagInitialSummaryRes;
import com.sofa.linkiving.domain.link.dto.response.RagRegenerateSummaryRes;
import com.sofa.linkiving.global.error.exception.BusinessException;
import com.sofa.linkiving.infra.feign.EmptyAiResponseException;
import com.sofa.linkiving.infra.feign.ExternalApiErrorCode;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
Expand Down Expand Up @@ -51,47 +54,53 @@ private Counter buildCounter(String operation, String result) {

@Override
public RagInitialSummaryRes initialSummary(Long linkId, Long userId, String title, String url, String memo) {
List<RagInitialSummaryRes> response;
try {
RagInitialSummaryReq req = new RagInitialSummaryReq(linkId, userId, title, url, memo);
List<RagInitialSummaryRes> response = ragSummaryFeign.requestInitialSummary(req);

if (response != null && !response.isEmpty()) {
log.info("[AI Server] Initial Summary Requested Success. LinkId: {}", linkId);
initialSuccess.increment();
return response.get(0);
}

initialEmpty.increment();
return null;

response = ragSummaryFeign.requestInitialSummary(req);
} catch (BusinessException e) {
initialFailure.increment();
log.warn("[AI Server] Initial summary failed - linkId={}, code={}", linkId, e.getErrorCode().getCode());
throw e;
} catch (Exception e) {
log.error("[AI Server Error] Failed to request initial summary for LinkId: {}. Error: {}", linkId,
e.getMessage());
initialFailure.increment();
return null;
log.warn("[AI Server] Initial summary failed - linkId={}, reason={}", linkId, e.getMessage());
throw new BusinessException(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
}

if (response == null || response.isEmpty()) {
initialEmpty.increment();
log.warn("[AI Server] Initial summary empty response - linkId={}", linkId);
throw new EmptyAiResponseException();
}

initialSuccess.increment();
return response.get(0);
}

@Override
public RagRegenerateSummaryRes regenerateSummary(Long linkId, Long userId, String url, String existingSummary) {
List<RagRegenerateSummaryRes> response;
try {
RagRegenerateSummaryReq req = new RagRegenerateSummaryReq(linkId, userId, url, existingSummary);
List<RagRegenerateSummaryRes> response = ragSummaryFeign.requestRegenerateSummary(req);

if (response != null && !response.isEmpty()) {
log.info("[AI Server] Regenerate Summary Success. LinkId: {}", linkId);
regenerateSuccess.increment();
return response.get(0);
}

regenerateEmpty.increment();
return null;

response = ragSummaryFeign.requestRegenerateSummary(req);
} catch (BusinessException e) {
regenerateFailure.increment();
log.warn("[AI Server] Regenerate summary failed - linkId={}, code={}", linkId, e.getErrorCode().getCode());
throw e;
} catch (Exception e) {
log.error("[AI Server Error] Failed to regenerate summary for LinkId: {}. Error: {}", linkId,
e.getMessage());
regenerateFailure.increment();
return null;
log.warn("[AI Server] Regenerate summary failed - linkId={}, reason={}", linkId, e.getMessage());
throw new BusinessException(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
}

if (response == null || response.isEmpty()) {
regenerateEmpty.increment();
log.warn("[AI Server] Regenerate summary empty response - linkId={}", linkId);
throw new EmptyAiResponseException();
}

regenerateSuccess.increment();
return response.get(0);
}
}
15 changes: 5 additions & 10 deletions src/main/java/com/sofa/linkiving/domain/link/worker/SummaryWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import com.sofa.linkiving.domain.link.event.SummaryStatusEvent;
import com.sofa.linkiving.domain.link.facade.SummaryWorkerFacade;
import com.sofa.linkiving.global.logging.LogContext;
import com.sofa.linkiving.infra.feign.EmptyAiResponseException;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
Expand Down Expand Up @@ -95,8 +96,6 @@ private void processQueue() throws InterruptedException {
String userEmail = null;
log.info("Processing link for summary - linkId: {}", linkId);

generateFailureCounter.increment();

try {
Link link = summaryWorkerFacade.getLinkWithMember(linkId);
userEmail = link.getMember().getEmail();
Expand All @@ -123,6 +122,7 @@ private void processQueue() throws InterruptedException {
));
}
} catch (Exception e) {
generateFailureCounter.increment();
log.error("Failed to generate summary for linkId: {}", linkId, e);

try {
Expand All @@ -146,25 +146,20 @@ private void processQueue() throws InterruptedException {
}

@Retryable(
value = {Exception.class},
retryFor = {Exception.class},
noRetryFor = {EmptyAiResponseException.class},
maxAttempts = 3,
backoff = @Backoff(delay = 2000)
)
public RagInitialSummaryRes callAiServerWithRetry(Link link) {
log.info("Attempting summary request to AI server - linkId: {}", link.getId());

RagInitialSummaryRes res = summaryClient.initialSummary(
return summaryClient.initialSummary(
link.getId(),
link.getMember().getId(),
link.getTitle(),
link.getUrl(),
link.getMemo()
);

if (res == null) {
throw new RuntimeException("Received a null response from the AI server.");
}

return res;
}
}
15 changes: 12 additions & 3 deletions src/main/java/com/sofa/linkiving/global/config/AsyncConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package com.sofa.linkiving.global.config;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
Expand All @@ -24,11 +25,19 @@ public AsyncConfig(TaskDecorator taskDecorator) {
@Bean
public ThreadPoolTaskExecutor applicationTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(50);
executor.setQueueCapacity(200);

executor.setCorePoolSize(5);
executor.setMaxPoolSize(20);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("async-");

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(30);

executor.setTaskDecorator(taskDecorator);

executor.initialize();
return executor;
}
Comment on lines +34 to 43

@Goder-0 Goder-0 Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

CallerRunsPolicy 자체가 잘못된 건 아니지만, 현재처럼 AI 연동이나 후속 이벤트 처리까지 같은 전역 async executor에 태우는 구조에서는 포화 시 비동기 작업이 요청/이벤트 스레드에서 직접 실행될 수 있다는 점이 조금 우려됩니다.
이 작업들이 사용자 응답 시간과 분리되어야 하는 성격이라면, 워크로드별로 executor를 분리하거나 포화 시에는 명시적으로 reject/관측 가능하게 처리하는 쪽이 운영상 더 예측 가능할 것 같습니다.
만약 CallerRunsPolicy를 유지하는 게 의도라면, 포화 시에도 호출 스레드 실행을 허용해도 괜찮은 이유가 함께 있으면 좋겠습니다.

@ckdals4600 ckdals4600 Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

좋은 지적 감사합니다.
먼저 CallerRunsPolicy를 선택한 이유는 해당 executor에 올라가는 작업들의 성격때문입니다.
현재 이 전역 async executor에 태우는 작업들(링크 동기화, 요약 큐 적재, 상태 알림 등)은 대부분 유실되면 안 되는 정합성 작업입니다.
특히 이들은 대부분 @TransactionalEventListener(AFTER_COMMIT)에서 트리거되는데, 여기서 AbortPolicy로 드롭되면 트랜잭션은 이미 커밋됐는데 후속 작업만 유실되어 정합성이 깨져 동기화 누락이나 요약 영구 미실행 등의 문제가 발생할 수 있습니다.
게다가 해당 rejection은 submit 시점에 호출 스레드로 던져져 리스너의 @Retryable로도 잡기 불가능합니다.
이러한 이유로 포화 시 작업을 잃기보다 프로듀서를 느리게 만드는 백프레셔로 CallerRunsPolicy를 택했습니다.
또한 큐의 경우도 bounded(100)으로 설정하여 무한정 쌓이지 않고, 각 설정 값에 대해서도 아래와 같은 의 보수적sizing이라 caller-runs 발동은 지속 포화 상황의 edge 케이스라고 생각됩니다.

  • core 5(평상시 상주 스레드 5개)
  • max 20(부하 시 최대 20개까지 확장)
  • queue 100(스레드가 다 찼을 때 대기 작업을 최대 100개까지 queuing)

다만 말씀하신 워크로드 혼재와 포화 시 요청/이벤트 스레드 지연 문제에 대해서는 충분히 발생 가능한 문제로 확인되어 아래와 같은 방법을 통해 보완하도록 하겠습니다.

  • 관측: applicationTaskExecutor는 자동계측되어 executor_active_threads / executor_queued_tasks / executor_pool_size_threads 원본 지표가 /actuator/prometheus에 노출됩니다. 다만 이걸 보는 대시보드 패널과 포화 알림 룰은 존재하지 않아, executor 포화 패널과 알림 룰(queued가 capacity(100)에·active가 max(20)에 근접)을 이번 PR에 함께 작업해서 올리도록 하겠습니다.
  • 분리: AI 연동 계열과 이벤트 후속 처리 계열을 별도 executor로 나누는 것에 대해서는 동의합니다. 다만 이번 PR 범위가 "전역 executor bounded화 + graceful shutdown 안정화"라, executor 분리 작업은 별도 이슈를 통해 작업하는 것이 좋을 것 같습니다.

@Goder-0 Goder-0 Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

좋습니다. 그렇게 하시죠.

@ckdals4600 ckdals4600 Jul 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Goder-0 말씀 주신 포화 관측 작업 진행했고, PR에도 반영했습니다.

  • Grafana: AI/Async 대시보드에 executor 포화 패널 3개 추가 (큐 대기 / 스레드 active·pool·max / 큐 사용률 게이지)
  • Prometheus: 포화 알림 룰 2개 추가 (AsyncExecutorQueueSaturation, AsyncExecutorPoolExhausted, for: 2m)

이제 CallerRunsPolicy가 발동하는 포화 상황을 그래프·알림으로 사전 감지할 수 있습니다.

워크로드별 executor 분리는 이번 PR 범위와 성격이 달라 별도 이슈(#260)로 분리했습니다.

Expand Down
3 changes: 2 additions & 1 deletion src/main/java/com/sofa/linkiving/global/fillter/RequestLoggingFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@ public class RequestLoggingFilter extends OncePerRequestFilter {
"/favicon.ico",
"/swagger",
"/v3/api-docs",
"/health-check"
"/health-check",
"/h2-console"
};

private static final int MAX_BODY_LENGTH = 2000;
Expand Down
9 changes: 9 additions & 0 deletions src/main/java/com/sofa/linkiving/infra/feign/EmptyAiResponseException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
package com.sofa.linkiving.infra.feign;

import com.sofa.linkiving.global.error.exception.BusinessException;

public class EmptyAiResponseException extends BusinessException {
public EmptyAiResponseException() {
super(ExternalApiErrorCode.EXTERNAL_API_INVALID_RESPONSE);
}
}
29 changes: 15 additions & 14 deletions src/test/java/com/sofa/linkiving/domain/chat/ai/RagAnswerClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@

import com.sofa.linkiving.domain.chat.dto.request.RagAnswerReq;
import com.sofa.linkiving.domain.chat.dto.response.RagAnswerRes;
import com.sofa.linkiving.global.error.exception.BusinessException;
import com.sofa.linkiving.infra.feign.EmptyAiResponseException;
import com.sofa.linkiving.infra.feign.ExternalApiErrorCode;

import io.micrometer.core.instrument.simple.SimpleMeterRegistry;

Expand Down Expand Up @@ -60,34 +63,32 @@ void shouldReturnFirstElement_WhenGenerateAnswerSuccess() {
}

@Test
@DisplayName("Feign 요청 중 예외가 발생하면 예외를 잡고 null을 반환한다")
void shouldCatchExceptionAndReturnNull_WhenGenerateAnswerThrowsException() {
@DisplayName("Feign 요청 중 예외가 발생하면 통신 오류 예외로 전환하고 failure 로 집계한다")
void shouldThrowCommunicationError_WhenGenerateAnswerThrowsException() {
// given
RagAnswerReq req = mock(RagAnswerReq.class);
given(ragAnswerFeign.generateAnswer(any(RagAnswerReq.class)))
.willThrow(new RuntimeException("AI Server Error"));

// when
RagAnswerRes actualRes = ragAnswerClient.generateAnswer(req);

// then
assertThat(actualRes).isNull();
// when & then
assertThatThrownBy(() -> ragAnswerClient.generateAnswer(req))
.isInstanceOf(BusinessException.class)
.extracting("errorCode")
.isEqualTo(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
assertThat(counterCount("failure")).isEqualTo(1.0);
}

@Test
@DisplayName("Feign 응답이 비어있으면 null 을 반환하고 empty 로 집계한다")
void shouldReturnNullAndCountEmpty_WhenResponseIsEmpty() {
@DisplayName("Feign 응답이 비어있으면 EmptyAiResponseException 을 던지고 empty 로 집계한다")
void shouldThrowEmptyResponse_WhenResponseIsEmpty() {
// given
RagAnswerReq req = mock(RagAnswerReq.class);
given(ragAnswerFeign.generateAnswer(any(RagAnswerReq.class)))
.willReturn(Collections.emptyList());

// when
RagAnswerRes actualRes = ragAnswerClient.generateAnswer(req);

// then
assertThat(actualRes).isNull();
// when & then
assertThatThrownBy(() -> ragAnswerClient.generateAnswer(req))
.isInstanceOf(EmptyAiResponseException.class);
assertThat(counterCount("empty")).isEqualTo(1.0);
}
}
56 changes: 35 additions & 21 deletions src/test/java/com/sofa/linkiving/domain/link/ai/RagSummaryClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@
import com.sofa.linkiving.domain.link.dto.request.RagRegenerateSummaryReq;
import com.sofa.linkiving.domain.link.dto.response.RagInitialSummaryRes;
import com.sofa.linkiving.domain.link.dto.response.RagRegenerateSummaryRes;
import com.sofa.linkiving.global.error.exception.BusinessException;
import com.sofa.linkiving.infra.feign.EmptyAiResponseException;
import com.sofa.linkiving.infra.feign.ExternalApiErrorCode;

import io.micrometer.core.instrument.simple.SimpleMeterRegistry;

Expand Down Expand Up @@ -74,32 +77,30 @@ void shouldReturnInitialSummaryResWhenSuccess() {
}

@Test
@DisplayName("최초 요약 요청 시 응답이 비어있으면 null을 반환한다")
void shouldReturnNullWhenInitialSummaryResponseIsEmpty() {
@DisplayName("최초 요약 요청 시 응답이 비어있으면 EmptyAiResponseException 을 던지고 empty 로 집계한다")
void shouldThrowEmptyResponse_WhenInitialSummaryResponseIsEmpty() {
// given
given(ragSummaryFeign.requestInitialSummary(any(RagInitialSummaryReq.class)))
.willReturn(Collections.emptyList());

// when
RagInitialSummaryRes result = ragSummaryClient.initialSummary(1L, 100L, "Title", "URL", "Memo");

// then
assertThat(result).isNull();
// when & then
assertThatThrownBy(() -> ragSummaryClient.initialSummary(1L, 100L, "Title", "URL", "Memo"))
.isInstanceOf(EmptyAiResponseException.class);
assertThat(counterCount("initial", "empty")).isEqualTo(1.0);
}

@Test
@DisplayName("최초 요약 요청 중 예외 발생 시 로그를 남기고 null을 반환한다")
void shouldReturnNullWhenInitialSummaryThrowsException() {
@DisplayName("최초 요약 요청 중 예외 발생 시 통신 오류 예외로 전환하고 failure 로 집계한다")
void shouldThrowCommunicationError_WhenInitialSummaryThrowsException() {
// given
given(ragSummaryFeign.requestInitialSummary(any(RagInitialSummaryReq.class)))
.willThrow(new RuntimeException("AI Server Error"));

// when
RagInitialSummaryRes result = ragSummaryClient.initialSummary(1L, 100L, "Title", "URL", "Memo");

// then
assertThat(result).isNull();
// when & then
assertThatThrownBy(() -> ragSummaryClient.initialSummary(1L, 100L, "Title", "URL", "Memo"))
.isInstanceOf(BusinessException.class)
.extracting("errorCode")
.isEqualTo(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
assertThat(counterCount("initial", "failure")).isEqualTo(1.0);
}

Expand Down Expand Up @@ -130,17 +131,30 @@ void shouldReturnRegenerateSummaryResWhenSuccess() {
}

@Test
@DisplayName("요약 재생성 요청 중 예외 발생 시 null을 반환한다")
void shouldReturnNullWhenRegenerateSummaryThrowsException() {
@DisplayName("요약 재생성 요청 중 예외 발생 시 통신 오류 예외로 전환하고 failure 로 집계한다")
void shouldThrowCommunicationError_WhenRegenerateSummaryThrowsException() {
// given
given(ragSummaryFeign.requestRegenerateSummary(any(RagRegenerateSummaryReq.class)))
.willThrow(new RuntimeException("Connection Timeout"));

// when
RagRegenerateSummaryRes result = ragSummaryClient.regenerateSummary(1L, 100L, "URL", "Old");

// then
assertThat(result).isNull();
// when & then
assertThatThrownBy(() -> ragSummaryClient.regenerateSummary(1L, 100L, "URL", "Old"))
.isInstanceOf(BusinessException.class)
.extracting("errorCode")
.isEqualTo(ExternalApiErrorCode.EXTERNAL_API_COMMUNICATION_ERROR);
assertThat(counterCount("regenerate", "failure")).isEqualTo(1.0);
}

@Test
@DisplayName("요약 재생성 요청 시 응답이 비어있으면 EmptyAiResponseException 을 던지고 empty 로 집계한다")
void shouldThrowEmptyResponse_WhenRegenerateSummaryResponseIsEmpty() {
// given
given(ragSummaryFeign.requestRegenerateSummary(any(RagRegenerateSummaryReq.class)))
.willReturn(Collections.emptyList());

// when & then
assertThatThrownBy(() -> ragSummaryClient.regenerateSummary(1L, 100L, "URL", "Old"))
.isInstanceOf(EmptyAiResponseException.class);
assertThat(counterCount("regenerate", "empty")).isEqualTo(1.0);
}
}
Loading
Loading