Spring SSE의 내부 동작 탐색(AsyncRequestNotUsableException)

SSE 연결 중, Client가 연결을 종료했을 때 AsyncRequestNotUsableException이 발생했습니다. 왜 이런 에러 로그가 찍히는 거고, 비동기 처리를 한 적이 없는데 왜 에러에 Async가 붙은 건지 궁금해 글을 작성하게 되었습니다.

 

이를 이해하기 위해서는 우선 Spring에서의 SSE 연결 흐름에 대해 알아야 합니다.

---

SSE 요청 흐름

SSE의 초기 요청 흐름입니다. 여기서 중요한 점은, 연결을 유지하는 동안 스레드를 계속 점유하고 있지 않는다는 점입니다.

 

하나의 스레드가 하나의 연결을 담당하는 대신에, NIO Selector에게 연결 정보를 넘겨준 후 스레드 풀에 반환이 됩니다. NIO Selector 내부에서는 하나의 스레드로 여러 개의 연결을 관리할 수 있는데요, 상당히 흥미로운 내용이지만, 내용이 너무 깊어질 것 같아 넘어가도록 하겠습니다.

 

그렇다면 Spring은 이를 어떻게 구현했을까요?

 

Spring MVC에서는 SseEmitter를 반환하면, 내부적으로 Servlet 3.0의 비동기 처리(Async Processing) 모드가 활성화됩니다. 우리가 명시적으로 비동기 코드를 작성하지 않았더라도, 스레드를 반환하고 연결만 살려두기 위해 Spring이 요청을 비동기 상태(Async State)로 전환한 것이죠. 즉, SSE 연결 중인 모든 요청은 기술적으로 Async Request입니다.

 

Spring의 흐름을 살펴봅시다.

 

SSE를 사용하기 위해서는 SseEmitter 객체를 만들어서 반환해줘야 합니다. 해당 객체를 만들어서 반환을 하면 HandlerAdapter가 SseEmitter가 반환된 것을 확인하고, ResponseBodyEmitterReturnValueHandler를 호출합니다. 이후에 해당 클래스에서 

request.startAsync() 호출을 진행한다고 이해하면 될 것 같습니다. (간략하게)

---

Client가 연결을 끊으면 무슨 일이 일어날까?

저는 실제 운영코드에서 SseEmitter의 timeout 시간을 무제한으로 설정했습니다.

public SseEmitter createEmitter(
            final UUID organizationUuid,
            final String userId,
            final ConnectionType connectionType
    ) {
        final SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        final String emitterId = generateEmitterId(organizationUuid, userId, connectionType);

 

타임아웃을 무제한으로 설정하는 대신, client가 정상적으로 연결이 되어있는지 확인하는 하트비트 스케줄링을 5분 간격으로 날려 연결을 확인하고 있습니다.

@Scheduled(cron = "0 */5 * * * *")
public void sendHeartbeat() {
    final Map<String, SseEmitter> emitters = sseEmitterRepository.findAll();

    if (emitters.isEmpty()) {
        log.debug("하트비트 스케줄링 스킵");
        return;
    }

    log.info("Heartbeat 전송 - 대상: {}개", emitters.size());

    //TODO : 순차전송이 아닌, 비동기 전송을 고려해보자
    for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) {
        final String emitterId = entry.getKey();
        final SseEmitter emitter = entry.getValue();

        try {
            emitter.send(SseEmitter.event()
                    .name("heartbeat")
                    .data("ping"));
        } catch (IOException e) {
            log.info("좀비 연결 제거 - ID : {} " , emitterId);
            sseEmitterRepository.remove(emitterId);
        }
    }

    log.debug("Heartbeat 완료 ");
}

 

이 과정에서 emitter.send()가 실패하면 요청이 성공적으로 보내지지 않았다는 것이고, Client가 연결을 끊었다는 것을 유추할 수 있습니다. 연결을 끊었으니 서버에서도 해당 emitter를 관리할 필요가 없기에 삭제를 진행했고요.

 

하지만 로그를 확인해 본 결과, 제가 생각한 것과는 사뭇 다르게 로직이 돌아가고 있었습니다.

 

(SSE 연결 코드)

public SseEmitter createEmitter(
            final UUID organizationUuid,
            final String userId,
            final ConnectionType connectionType
    ) {
        final SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        final String emitterId = generateEmitterId(organizationUuid, userId, connectionType);

        sseEmitterRepository.save(emitterId, emitter);
        log.info("SSE 연결 생성 - Type: {}, Emitter ID: {}", connectionType, emitterId);

        emitter.onCompletion(() -> {
            log.info("SSE 연결 정상 종료 - {}", emitterId);
            sseEmitterRepository.remove(emitterId);
        });

        emitter.onTimeout(() -> {
            log.warn("SSE 연결 타임아웃 - {}", emitterId);
            sseEmitterRepository.remove(emitterId);
        });

        emitter.onError((e) -> {
            log.error("SSE 연결 에러 - {}: {}", emitterId, e.getMessage());
            sseEmitterRepository.remove(emitterId);
        });

        sendToClient(emitter, emitterId, "connect", "EventStream Created");

        return emitter;
    }

 

 

SSE 연결 에러는 emitter.onError()가 실행되었을 때 작성되는 로그인데요, 해당 메서드가 왜 호출이 되었을까요?? 또, AsyncRequestNotUsableException은 왜 발생하는 걸까요?

 

제가 기대한 흐름은 다음과 같습니다.

 

 

 

하지만 실제로는 emitter.onError() 메서드가 호출되고, Broken Pipe라는 예외가 발생하는데요, 내부적으로 어떻게 동작을 하는 걸까요?

 

사실은 내부적으로 더 훨씬 복잡하게 동작합니다.

 

1) Client가 연결을 끊고

2) 하트비트 스케줄링을 통해 연결을 확인하는 과정에서, emitter.send()가 실패하는 것을 확인합니다.

3) 이 과정에서 IOException이 발생하는데요, Tomcat은 IOException을 던져줌과 동시에, AsyncContext의 상태를 Error로 변경합니다.

4) 스케줄링 작업이 끝나고 나면, 톰캣은 AsyncListener에게 AsyncListener.onError()를 호출하라고 명령합니다. 위에서 예측한 것처럼 IOException을 던지고 끝나는 게 아니라, 추가적인 작업을 수행하는 것이죠.

5) AsyncListener는 SseEmitter의 emitter.onError() 실행을 명령합니다. 이 과정에서, SSE 연결 에러 로그가 출력이 되었던 것이죠.

6) 이후 모든 콜백이 종료되면, WebAsyncManager은 비동기 요청을 마무리하기 위해 Tomcat에게 dispatch()를 시도합니다.

7) 하지만, 3) 과정에서 AsyncContext의 상태가 Error로 기록되어 있기에, Tomcat은 dispatch() 요청을 거절하고 AsyncRequestNotUsableException을 발생시킵니다.

 

정리해 보자면 Heartbeat Scheduler에서 try-catch 문으로 IOException을 처리해도, Tomcat이 AsyncListener.onError()를 호출하기에 별도의 에러가 발생하는 상황입니다. 이 과정에서 불필요한 Error 로그들이 발생했고요.

---

해결방안

에러 로그는 에러상황에만 찍혀야 한다고 생각합니다. 불필요한 에러 로그가 쌓이면, 추후에 진짜 에러가 발생했을 때 찾기가 힘들어지죠.

 

1) GlobalExceptionHandler에서 AsyncRequestNotUsableException 처리

@ExceptionHandler(AsyncRequestNotUsableException.class)
    public void handleAsyncRequestNotUsable(AsyncRequestNotUsableException e) {
        log.debug("SSE 연결이 끊겨 사용자에게 메세지 응답 불가", e);
    }

 

ExceptionHandler에서 해당 에러를 잡아 debug 레벨로 로깅하는 전략을 선택했습니다. info와 debug 중에 고민을 했는데, 해당 로그는 지속적으로 계속 찍히지만 불필요하다고 판단해 debug 레벨을 선택했습니다.

 

2) 로그 세분화

emitter.onError((e) -> {
    final String message = e.getMessage();
    if (message != null && message.contains("disconnected client")) {
        log.debug("SSE 연결 종료(Client Disconnect)");
    } else {
        log.error("SSE 연결 에러 발생 - ID: {}, 메시지: {}", emitterId, message, e);
    }
    sseEmitterRepository.remove(emitterId);
});

 

emitter.onError() 또한 지속적으로 호출되어 문제가 되었었는데요, Client가 연결을 종료한 상황은 정상적인 흐름이라고 판단했고 위와 마찬가지로 로그가 불필요하다고 판단해 debug 레벨로 분리했습니다. 그 외에는 error 레벨을 설정했고요.

---

SSE는 참 좋은 기술인 것 같네요. 하지만 깊이 있게 공부를 해야 추후에 버그가 났을 때 찾을 수 있겠다는 생각도 드네요ㅎ... (신경써야 할 부분이 은근 있습니다).

 

팀원들에게 공유 + 학습 목적으로 글을 작성했지만, 저처럼 분명 try-catch로 IOException 잡았는데 왜 에러가 뜨지? 하며 궁금증이 생기시는 분들에게 작은 도움이 되면 좋겠네요.