[Spring] SSE 구축과 Exception: Could not open JPA EntityManager for transaction 에러 해결 - 코루틴 적용하기
Kubernetes로 MinIO에 파일을 업로드하는 Job을 띄워 진행률을 Kafka로 Produce하고, 백엔드서버에서 consume을 통해 업로드 진행률을 확인할 수 있도록 하였다.
이를 FE에 실시간으로 알려주기 위해서 SSE(Server-Sent Events) 컨트롤러를 구현하였는데, 로컬환경에서 테스트할때는 문제 없던 부분을 테스트베드에 배포를 하니까 연동하는데 문제가 발생했다. DB 조회조차 되지 않고 연속적으로 DB Connection 관련 에러 로그를 발생시키고 있었다.
우선적으로 SSE controller를 비활성화 시키고 배포를하니 DB Connection 에러가 발생하지 않아서 SSE를 호출하면서 발생한 문제로 확인되었다. 확인 후 수정이 필요했다.
SSE 구현부
진행률을 가져오는 SSE를 다음과 같이 구현했다.
controller
SseEmitter 를 통해 emitter 변수를 생성하고, 초기 emitter에 {}를 보내 초기화해준다.
@GetMapping("/{pid}")
fun getProgress(
@PathVariable pid: String,
request: HttpServletRequest
): Any {
val emitter = SseEmitter(Long.MAX_VALUE).apply {
onTimeout { complete() }
onError { complete() }
}
val emitterId = request.session.id
val emitterInfo = ExtendedSseEmitter(emitter, pid)
return try {
val initialData = "{}"
emitter.send(initialData)
kafkaService.addSseEmitter(emitterId, emitterInfo)
emitter
} catch (e: Exception) {
emitter.completeWithError(e)
ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(CommonMessage.ERROR)
logger.error(">>> Exception: ${e.message}")
}
}
KafkaListener
KafkaListener에서 emitter에 send 하는 부분은 아래와 같다. KafkaListener에서 구독하고 있는 topic에 메시지가 들어오면 해당 메시지를 emitter로 보내도록 구현했다.
...
sseMessages.values.forEach { emitterInfo ->
if (kafkaMessageJson.namespace == emitterInfo.namespace) {
try {
emitterInfo.emitter.send(rawMessage)
} catch (e: Exception) {
emitterInfo.emitter.completeWithError(e)
logger.info(">>> Emitter error: $e")
}
}
}
...문제 발생
😅 왜 SSE에 비동기 처리를 하지 않았을까
테스트까지는 문제가 없었으나 배포하니까 문제가 발생했다. sse에 비동기 처리 적용이 되지 않은 것이 문제였다.
SSE 이벤트 스트림을 통해 클라이언트와 BE 서버 간에 연결이 지속적으로 열려있었고 자원을 점유하고 있는 상황이 발생한 것이다. SSE가 서버 자원을 계속적으로 점유하고 있었으므로 데이터베이스 작업을 처리할 수 없어 데이터베이스 조회가 불가능하고, 이로 인해 아래와 같은 문제가 지속적으로 발생한 것이다.
- o.h.engine.jdbc.spi.SqlExceptionHelper : SQL Error: 0, SQLState: null
- o.h.engine.jdbc.spi.SqlExceptionHelper : HikariPool-1 - Connection is not available, request timed out after 50000ms.
- Exception: Could not open JPA EntityManager for transaction이 문제는 내가 Spring 환경에 익숙하지 않기 때문에 발생한 문제였다. 비동기처리를 따로 하지 않아도 스프링 프레임워크에서 쓰레드형식으로 자동적으로 관리해줄거라고 생각했던 이상한 믿음이 있었달까 😇
문제 해결
💡 SSE Controller에 비동기 처리 - Coroutine 적용을 통해 간단히 해결했다.
생각해보면 당연했던걸 하지 않았기에 급하게 비동기처리를 진행해주었다. Kotlin으로 개발하고 있었기 때문에 Coroutine을 아래와 같이 적용했다.
@GetMapping("/{pid}")
suspend fun getProgress(
@PathVariable namespace: String,
request: HttpServletRequest
): Any {
val emitter = SseEmitter(Long.MAX_VALUE).apply {
onTimeout { complete() }
onError { complete() }
}
val emitterId = request.session.id
val emitterInfo = ExtendedSseEmitter(emitter, pid)
CoroutineScope(Dispatchers.IO).launch {
try {
val initialData = "{}"
emitter.send(initialData)
kafkaService.addSseEmitter(emitterId, emitterInfo)
} catch (e: Exception) {
emitter.completeWithError(e)
ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(CommonMessage.ERROR)
logger.error(">>> Exception: ${e.message}")
}
}
return emitter
}함수명 fun 앞에 suspend를 추가하고 emitter 부분을 CoroutineScope로 묶어 코루틴처리를 진행하였다. 비동기 처리를 적용하여 배포하니 문제 없이 동작하는 것을 확인하였다.
(참고) CoroutineScope와 GlobalScope
CoroutineScope
코루틴스코프는 코루틴의 생명 주기를 관리하는 데 사용된다. 이를 사용하면 해당 스코프 내에서 시작된 모든 코루틴이 명시적인 취소나 완료 시까지 계속 유지된다.
CoroutineScope(Dispatchers.IO).launch { ... } 를 통해 새로운 코루틴 스코프를 생성하고 Dispatchers.IO 컨텍스트에서 코루틴을 시작한다. 이는 입출력 작업에 최적화된 스레드 풀을 사용한다.
데이터베이스 접근, 네트워크 통신, 파일 입출력 등 블로킹 I/O 작업을 처리하는데 적합하므로 비동기적으로 SSE 이벤트를 전송하고 Kafka 서비스와의 상호작용을 담당하는 부분이므로 이를 사용하였다.
GlobalScope
애플리케이션의 전체 생명 주기 동안 살아있는 전역 코루틴 스코프이다. 글로벌스코프로 시작된 코루틴은 애플리케이션이 종료되거나 명시적으로 취소되기 전까지 계속 실행된다.
GlobalScope의 사용은 권장되지 않는데, 전역 코루틴 스코프는 쉽게 누수될 수 있고 애플리케이션의 다른 부분과의 생명 주기를 공유하지 않기 때문에 리소스 관리가 어렵고 사이드 이펙트를 줄 수 있기 떄문이다.
결론
CoroutineScope 를 사용하는 것이 GlobalScope 보다 리소스 관리 측면에서 안정적이며, 코루틴을 조금 더 세밀하게 제어할 수 있다. 또한, 코루틴이 시작된 특정 컨텍스트 내에서만 동작하게 하므로 코드의 안정성과 가독성을 높일 수 있다는 장점이 있다.
