이벤트 발행 트랜잭션 처리

@TransactionalEventListener 적용 시 주의할 점

• 하지만 구현된 코드에서 알림을 보내는 이벤트에서 에러가 발생한다면 작성된 댓글이 저장이 되나? 생각했다.

내가 생각한 알림 서비스

사용자가 댓글을 달았는데, 알림쪽에서 문제가 발생했다. 근데 같은 트랜잭션 안에 있기 때문에 댓글이 작성되지 않을 것이다. 알림은 유실되어도 괜찮다고 생각하지만 댓글 작성 같은 경우에는 절대 유실되어서는 안 된다고 생각한다.

그렇기 때문에 댓글 등록과 알림은 서로 독립적인 논리 트랜잭션 안에서 동작하도록 개선하려고 한다.

@EventListener와 @TransactionEventListener

1. @EventListener

• 트랜잭션에 참여하지 않고 트랜잭션이 없으면 동기적으로 수행한다. 즉, 이벤트가 발생하는 즉시 리스너가 동작한다. 트랜잭션의 상태와 관련없이 이벤트 핸들러가 실행된다.

• 트랜잭션 정보 활용 불가능

• 이벤트 핸들러에서 예외가 발생해도 트랜잭션에 영향을 주지 않는다.

• 회원가입 후 환영 메일이 오는것과 같은 트랜잭션과 관련없는 이벤트에서 사용할 수 있다.

1. @TransactionEventListener

• 트랜잭션이 커밋된 이후 트랜잭션 상태(커밋, 롤백)에 따라서 이벤트 핸들러가 실행된다.

• 이벤트 핸들러에서 예외가 발생하면 트랜잭션이 롤백될 수 있다.

이 프로젝트에서 리뷰가 생성되지 않고 rollback 되었는데 사용자한테 알림이 발송되면 안되기 때문에 후자를 선택했다.

기존 코드

✅ NotificationListener

@Component
public class NotificationListener {

  private final NotificationFacadeService notificationFacadeService;

  public NotificationListener(NotificationFacadeService notificationFacadeService) {
    this.notificationFacadeService = notificationFacadeService;
  }

  @EventListener
  public void reviewCommentNotificationHandler(ReviewCommentEvent event) {
    notificationFacadeService.reviewCommentNotifyOnEvent(event);
  }
}

✅ NotificationService & ReviewCommentService

@Override
@Transactional
public ReviewCommentResult createComment(CreateReviewCommentCommand command, LocalDateTime now) {
  ReviewResult reviewResult =
      ...
  ReviewCommentResult comment = reviewCommentService.createComment(command, now);

  eventPublisher.publishEvent(reviewCommentEvent);
  return comment;
}

-----------------------------------------------------------------------------

@Service
@RequiredArgsConstructor
public class NotificationService {
  private final CreateNotificationService createNotificationService;
  private final SendNotificationService sendNotificationService;

  public void reviewCommentNotifyOnEvent(ReviewCommentEvent event) {
    var now = LocalDateTime.now();
    var type = NotificationType.COMMENT;
    var command =
        new CreateNotificationCommand(event.content(), null, event.receiveUserId(), type, now);
    var notificationResult = createNotificationService.create(command);
    sendNotificationService.send(notificationResult);
  }
}

✅ CreateotificationService

@Service
@RequiredArgsConstructor
public class CreateNotificationService
    implements CreateNotificationUseCase, CreateTimeUserIdUseCase {

  private final NotificationRepository notificationRepository;

  @Override
  @Transactional
  public NotificationResult create(CreateNotificationCommand command) {
    var notification = notificationRepository.save(Notification.create(..));
    return NotificationResult.from(notification);
  }
}

createReviewComment() 메소드는 comment를 생성하고, event를 보내고 있다.

기존 구현 방식의 알림을 전송하기까지의 트랜잭션 흐름을 보면, 한 트랜잭션 안에서 진행되고 있다. 로직의 순서만 봤을 때, 코드는 분리되어 있지만 트랜잭션은 그렇지 않다. 이 경우 알림에 실패하면 댓글 생성까지 rollback 되어버린다. 알림 서비스는 중요한 비즈니스 로직이 아니라고 생각한다. 어디까지나 사용자의 편의를 위한 부가 기능일 뿐이다.

여기서 중요한 것은 commit 시점이다. 위 코드의 작업은 하나의 쓰레드에서 동기적으로 수행되므로 작업이 순차적으로 처리된다. 이벤트 작업을 무사히 끝내야만 다시 createReviewComment()로 돌아와서 트랜잭션이 commit 될 수 있다.

트랜잭션을 확인해보면 다음과 같다.

Getting transaction for [createComment]
Getting transaction for [reviewCommentService.createComment]
Getting transaction for [notificationService.create]

Completing transaction for [notificationService.create]
Completing transaction for [reviewCommentService.createComment]
Completing transaction for [createComment]

이벤트 발행 시점 변경하기

@TransactionalEventistener어노테이션은 @Transactional이 붙은 메소드가 커밋 된 후 실행하는 메소드이다. 즉, 해당 트랜잭션이 commit된 이후 리스너가 동작하게 해준다.

• EventListener 실행 시점을 설정할 수 있게 도와준다.

phase 옵션 :

• AFTER_COMMIT : 호출자의 트랜잭션이 commit된 후 이벤트를 발생시킵니다. (DEFAULT)

• BEFORE_COMMIT : 호출자의 트랜잭션이 commit되기 직전에 이벤트를 발생시킵니다.

• AFTER_COMPLETION : 호출자의 트랜잭션 commit의 성공여부와 관계없이 끝나면 실행시킵니다.

• AFTER_ROLLBACK : 호출자의 트랜잭션이 rollback 된 후 실행시킵니다.

그럼 다시 코드를 변경해 보자.

✅ NotificationListener 변경

@Component
public class NotificationListener {
  ...
  
  @TransactionalEventListener
  public void reviewCommentNotificationHandler(ReviewCommentEvent event) {
    notificationFacadeService.reviewCommentNotifyOnEvent(event);
  }
}

이 상태로 트랜잭션을 다시 확인해 보자.

Getting transaction for [createComment]
Getting transaction for [reviewCommentService.createComment]
Completing transaction for [reviewCommentService.createComment]
Completing transaction for [createComment]

Getting transaction for [notificationService.create]
Completing transaction for [notificationService.create]

댓글 생성과 알림 기능이 잘 분리된 것 같다. 하지만 DB를 확인해 보니 새로운 문제가 발생했다.

트랜잭션 로그를 보면 notification이 생성되는 것을 기대하지만, 실제로 쿼리 로그를 보면 insert 쿼리는 발생하지 않는다.

문제 상황과 원인 분석

이벤트를 수행하면 이벤트 객체를 생성하고 어떤 이벤트가 생성 되었는지 DB에 기록하는 것을 목표로 한다,

하지만 유저에게 전송까지 잘 되는데 왜 DB에 반영이 안됐을까?

데이터베이스 Write 작업은 트랜잭션이 commit 되었을 때 완료된다. 스프링에서는 @Transactional 이 메소드 시작에 트랜잭션을 열고 메소드가 끝나는 시점에 commit을 도와준다.

변경 전 상태와 변경 후 트랜잭션 상태를 비교해 보자.

변경 전 : createReviewComment() -> (메소드 마지막에) commit -> 이벤트 핸들러 처리 -> 돌아와서 comment return; 변경 후 : createReviewComment() 시작, 종료 → create() 시작, 종료

여기서 주의할 점은 여전히 하나의 트랜잭션으로 묶여 있다는 것이다.

이미 comment를 save() 하고 commit까지 마친 것이 아닌가?

트랜잭션 로그를 보면 comment 생성과 이벤트 기능이 서로 다른 트랜잭션 안에서 동작하는 것이 아닌가? 라고 생각할 수 있다.

하지만 creatReviewComment()에서 만들어진 트랜잭션은 commit 되었지만, 트랜잭션이 사라진 것은 아니다.

🌟 이벤트 핸들러에서 이미 commit 된 creatReviewComment()의 트랜잭션에 참여한 것이다. 🌟

createReviewComment()의 트랜잭션이 commit되고 이벤트 핸들러가 동작하는데, 그 안에 발생한 이벤트를 save()해서 db에 write하는 작업이 존재하지만 앞에서 참여한 트랜잭션이 commit 되어 반영되지 않는 것이다.

• AFTER_COMMIT이 default

해결 방안

그럼 이벤트 핸들러 내부에서 DB 변화가 발생해서 write를 해야한다면 어떻게 해야할까?

1. 새로운 트랜잭션 생성

@Transactional(propagation = Propagation.REQUIRES_NEW)을 붙여주는 방법이다. 이렇게 하면 이벤트 리스너 로직 안에서 실행되는 @Transactional이 적용된 로직은 이전의 트랜잭션과 별도로 새롭게 트랜잭션을 시작한다. 하지만 트랜잭션이 있던 없던 상관없이 새로운 트랜잭션을 만들어 독립적으로 커밋과 롤백을 진행하므로 커넥션 풀의 커넥션을 한 개 더 차지한다. (독립적으로 열린 트랜잭션 사이에 데드락을 조심해야 한다.)

@Component
public class NotificationListener {
  ...
  
  @Transactional(propagation = Propagation.REQUIRES_NEW)
  @TransactionalEventListener()
  public void reviewCommentNotificationHandler(ReviewCommentEvent event) {
    notificationFacadeService.reviewCommentNotifyOnEvent(event);
  }
}

🌟 이 방법은 maximum-pool-size = 1로 설정되어 있으면 핸들러 메소드를 실행하지 못하고 DB 커넥션을 가져오기 위해 대기하다가 Time Out이 발생할 수 있다.

// setting pool-size
spring.datasource.hikari.maximum-pool-size = 1

// transaction log
Getting transaction for [createComment]
Getting transaction for [reviewCommentService.createComment]
Completing transaction for [reviewCommentService.createComment]
Completing transaction for [createComment]

// error log
SQL Error: 0, SQLState: null
Connection is not available, request timed out after 15007ms 
                      (total=1, active=1, idle=0, waiting=0)
TransactionSynchronization.afterCompletion threw exception
JDBCConnectionException: Unable to acquire JDBC Connection

2. AFTER_COMMIT 대신 BEFORE_COMMIT

위에서 phase에 대해 언급을 했었다. AbstractPlatformTransactionManager 추상 클래스의 processCommit이 받는 status에 따라 트리거가 동작한다.

따라서 우리가 발생한 문제를 해결하기 위해서는 @TransactionalEventListener의 phase 옵션을 BEFORE_COMMIT으로 지정한다.

이렇게 설정하면 커밋되기 전에 리스너가 실행되기 때문에 리스너 로직의 트랜잭션이 커밋될 수 있다.

하지만 리스너쪽에서 예외가 발생하면 이벤트를 발생시키는 핵심 로직의 트랜잭션에도 영향을 줄 수 있기때문에 주의해서 사용하도록 한다.

@Component
public class NotificationListener {
  ...
  
  @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
  public void reviewCommentNotificationHandler(ReviewCommentEvent event) {
    notificationFacadeService.reviewCommentNotifyOnEvent(event);
  }
}

3. 비동기 사용, @Async 어노테이션 추가

어노테이션을 달면 이벤트 리스너가 별도의 스레드에서 실행된다. 또, AFTER_COMMIT 이후에 동일한 데이터 소스를 사용하지 않는다. 하지만 이 방법은 테스트가 까다롭다.

@EnableAsync
@Configuration
public class AsyncConfig implements AsyncConfigurer {

  @Override
  public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(8);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("AsyncThread-");
    executor.initialize();
    return executor;
  }
}

@Component
public class NotificationListener {
  ...
  
  @TransactionalEventListener()
  @Async()
  public void reviewCommentNotificationHandler(ReviewCommentEvent event) {
    notificationFacadeService.reviewCommentNotifyOnEvent(event);
  }
}

+ OSIV가 켜져있을 시 SSE 서비스단에 트랜잭션이 걸려있다면 SSE연결 동안 트랜잭션을 계속 물고 있어 커넥션 낭비가 일어날 수 있으니 트랜잭션을 걸지 않아야 합니다.

테스트 및 방안 채택

도메인 간 결합을 느슨하게 하면서 발생하는 문제를 해결하기 위해 3가지 방법을 고민하고 Locust를 사용해서 동일한 환경에서 특정 트래픽에 어떤 결과가 나오는지 궁금해서 테스트를 진행했다.

1. @Transactional(propagation = Propagation.REQUIRES_NEW)

위에서 언급했던 커넥션 풀 고갈 문제가 바로 나타났다. 하나의 요청에 대해 두 개의 데이터베이스 커넥션을 사용하게 되어, 트래픽이 급증할 경우 커넥션 풀이 고갈되어 서버에서 에러가 발생했고, 이후 지연되는 요청들은 전부 실패했다.

java.sql.SQLTransientConnectionException: perfume-db-pool - Connection is not available, 
request timed out after 15005ms (total=24, active=24, idle=0, waiting=117)

1. BEFORE_COMMIT과 비동기 비교

BEFORE_COMMIT 방식과 비동기 방식을 비교하면 Event 처리량, TPS, 특정 부하에서 Pending이 관찰되는 상태까지 비교적 동일하다. 각 방식의 장점은 다르기 때문에 이를 분석하고 비교하여 적용하려고 한다.

이벤트 리스너의 BEFORE_COMMIT 방식은 트랜잭션이 커밋되기 전 이벤트를 처리한다. 따라서 트랜잭션이 성공적으로 완료되지 않으면 이벤트 처리도 롤백된다.

• 장점: 데이터의 일관성을 보장할 수 있고, DB 변경이 이루어진 후 결과에 따라 후속 작업을 수행할 수 있다.

• 단점: 이벤트 처리는 동기적으로 이루어져, 이벤트가 처리되는 동안 호출 대기가 발생하며 성능 저하로 이어질 수 있다.

@Async를 사용하는 비동기 방식은 이벤트 리스너가 별도의 스레드에서 실행되므로 즉시 호출자는 다음 작업으로 넘어갈 수 있다.

• 장점: 메인 트랜잭션의 완료를 기다리지 않고 다음 작업을 수행하기 때문에 성능 개선에 유리하다.

• 단점: 데이터 일관성 보장이 어려울 수 있다. 예를 들어, 트랜잭션이 롤백 되더라도 비동기로 실행된 이벤트는 처리될 수 있다.

위에서도 언급했지만, 알림 기능은 핵심 비즈니스나 주요 기능을 보완하는 하나의 부가 기능이라고 생각한다.

따라서 이벤트로 동작하는 알림 기능의 실패가 비즈니스 로직에 영향을 주어서는 안되기 때문에 서비스 특성과도 적합하고 성능상 이점을 가져올 수 있는 비동기 방식을 채택했다.

결론적으로, 정답은 없는 것 같다. 위와 같은 경우에는 상황에 따라서 달라질 수 있고, 성능과 데이터 일관성 간의 균형을 고려하여 각 방식의 장단점을 이해하고 서비스 특성이나 요구사항에 맞는 방식을 선택하는 것이 중요하다고 생각된다.

Summary

• 핵심 비즈니스 로직과 이벤트 로직은 한 트랜잭션에 묶으면 좋지 않을 것 같다. (서비스 특성마다 다를 수 있다.)

• 이벤트 리스너를 트랜잭션과 함께 사용하는 경우는 AFTER_COMMIT이 기본값으로 설정되어 있으므로 위와 같은 방법으로 처리할 수 있다.