[ spring ] 통합 테스트 중 RabbitMQ 동작 결과에 대한 문제

NYoun 2025. 7. 1. 22:22

2025. 7. 1. 22:22

문제

RabbitMQ를 어떤 용도에 따라 사용하느냐에 따라 발생하는 문제 및 해결 방법이 달라지겠지만,

이번에는 Consumer에서 데이터 파싱 로직 처리 이후 DB에 insert 또는 update, delete 하는 경우에 대한 문제다.

여기서 문제는 3가지가 발생했다.

테스트 메서드가 RabbitMQ보다 빨리 끝나 처리된 데이터 검증이 제대로 진행되지 않는 문제
Consumer가 테스트 코드에서 저장해둔 데이터에 접근하지 못하는 문제
RabbitMQ를 통해 insert 된 데이터가 롤백되지 않는 문제

예시 및 문제 원인, 문제별 해결 방안

발생 환경에 대한 가정을 먼저 세운다.

테스트는 서비스에 대한 통합 테스트라고 가정.
RabbitMQ에서는 저장 또는 수정, 삭제할 데이터의 파싱을 처리하고 DB로 요청을 보내 데이터 처리를 수행.
내부 비즈니스 로직은 메시지를 통해 받은 데이터의 파싱 정도이기 때문에 중요도가 높지 않아 Consumer 코드는 파싱코드라고 가정하고 생략.
처리되는 테이블로는 SalesSummary에 저장, Product에 수정, Cart에 삭제가 처리된다고 가정하며 3개의 consumer가 호출되는 환경. Product는 stock이라는 재고를 수정, Cart는 무조건 해당 사용자의 장바구니 삭제를 처리.
테스트에서 호출되는 서비스 메서드는 Order 테이블에 주문 데이터를 저장한 뒤 3개의 메시지를 발행해 위 처리를 수행하는 서비스라고 가정. Order 테이블의 경우 동기적으로 처리
테스트 메서드에서는 서비스 메서드 호출 이후 데이터가 정상적으로 저장, 수정, 삭제 되었는지 검증을 해야 한다고 가정.

각 엔티티 코드는 중요도가 높지 않기 때문에 생략하고 서비스와 테스트 코드만 예시로 들어 정리.

서비스메서드 및 테스트 클래스

@Service
@RequiredArgsConstructor
public class TestService {
    
    private final OrderRepository orderRepository;
    
    private final RabbitTemplate rabbitTemplate;
    
    @Transactional
    public void payment(PaymentDTO paymentDTO) {
        //최초 데이터 파싱
        OrderDataDTO orderDataDTO = createOrderDataDTO(paymentDTO);
        Order order = orderDataDTO.getOrder();
        
        orderRepository.save(order);
        
        //RabbitMQ 메시지 발행 메서드
        //3개의 메시지를 발행해 Message Broker에게 보내는 메서드라고 가정.
        //이 메서드를 통해 메시지를 받은 각 Consumer에서는 데이터 파싱 이후 각 처리를 진행.
        sendOrderQueueMessage(paymentDTO, orderDataDTO);
    }
}



@SpringBootTest(classes = TestApplication.class)
@ActiveProfiles("test")
@Transactional
public class TestServiceIntegrationTest {
    
    @Autowired
    private TestService testService;
    
    @Autowired
    private SalesSummaryRepository salesSummaryRepository;
    
    @Autowird
    private OrderRepository orderRepository;
    
    @Autowird
    private ProductRepository productRepository;
    
    @Autowired
    private CartRepository cartRepository;
    
    //...
    
    //paymentDTO에 담아 보낼 상품 엔티티
    private Product product;
    
    //주문 요청하는 사용자 엔티티
    private Member member;
    
    @BeforeEach
    void init() {
        //테스트에 필요한 데이터 생성 및 저장
        //여기에서 생성된 데이터로 product, member 할당
    }
    
    @Test
    void rabbitMQTest() {
        PaymentDTO paymentDTO = OrderFixture.createPaymentDTO(product);
        int productStockFixture = product.getStock() - paymentDTO.getOrderCount();
        
        assertDoesNotThrow(() -> testService.payemnt(paymentDTO));
        
        //저장된 order 데이터 조회
        List<Order> saveOrderList = orderRepository.findAll();
        assertFalse(saveOrderList.isEmpty());
        assertEquals(1, saveOrderList.size());
        
        //저장된 order 데이터 검증
        //...
        
        //저장된 SalesSummary 조회
        List<SalesSummary> saveSummaryList = salesSummaryRepository.findAll();
        assertFalse(saveSummaryList.isEmpty());
        
        //저장된 SalesSummary 검증
        //...
        
        //수정된 Product 재고 검증
        Product patchProduct = productRepository.findById(productId).orElse(null);
        assertNotNull(patchProduct);
        assertEquals(productStockFixture, patchProduct.getStock());
        
        //Cart 삭제 검증
        Cart deleteCart = cartRepository.findCartByUserId(member.getUserId());
        assertNull(deleteCart);
    }
}

문제 원인 1) 테스트 메서드가 RabbitMQ보다 빨리 끝나 처리된 데이터 검증이 제대로 진행되지 않는 문제

위와 같이 코드를 작성하고 테스트를 수행하게 되면 order 데이터 검증까지는 성공하지만, SalesSummary 검증부터는 실패하게 된다.

RabbitMQ는 비동기로 동작하기 때문에 테스트가 끝날때까지 consumer의 처리와 DB 요청이 미처 마무리되지 못할 수 있다.

서비스메서드에서는 메시지를 발행하고 바로 종료되기 때문에 테스트 코드에서는 바로 검증을 시작하려고 시도하게 되기 때문이다.

consumer의 로직이 굉장히 단순하고 빠르게 처리된다고 하더라도 RabbitMQ와의 연결에서 지연시간이 발생하게 되면 오류가 발생할 수 있는 여지가 있기 때문에 테스트의 일관성을 위해서는 RabbitMQ로의 요청이 완료된 것을 기다렸다가 검증해야 한다.

해결 방안

가장 먼저 떠오를 수 있는 방법은 Thread.sleep()이 있다.

이걸 테스트 메서드에서 처리한다면 예상되는 RabbitMQ 처리시간을 대기했다가 검증하는 방법으로 문제를 해결할 수 있다.

하지만, Thread.sleep()은 아무래도 유연성과 효율성에서 단점이 발생하게 된다.

테스트 과정에서 평균 5초 이내에 처리가 되는것을 확인하고 Thread.sleep(5000) 을 걸어뒀다고 가정해보자.
그럼 여기서 2가지의 문제가 발생할 여지가 생기게 된다.

만약 평균 1초로 개선이 되었다면? 불필요하게 5초까지 꼭 대기해야 하는 비효율이 발생한다.

또 다른 문제로 연결이 지연되어 10초가 걸리는 상황이 발생했다면? 5초만 대기했기 때문에 검증에 다시 실패하게 된다.

이러한 문제들로 인해 Thread.sleep()을 사용하는 것은 하나의 방법이 될 수 있지만 너무 정적인 방법이라고 볼 수 있다.

이러한 단점을 상쇄시킬 수 있는 다른 방법은 await().untilAsserted()다.

의존성을 추가해야 사용할 수 있지만, 긴 대기시간을 주더라도 주기적으로 확인해 테스트가 통과되는 즉시 완료되기 때문에 Thread.sleep에서 발생할 수 있는 문제를 해결할 수 있다.

의존성은 아래와 같이 추가해주면 된다.

testImplementation("org.awaitility:awaitility:4.2.0")

그리고 테스트 클래스는 아래와 같이 수정하게 된다.

import static org.awaitility.Awaitility.await;

@SpringBootTest(classes = TestApplication.class)
@ActiveProfiles("test")
@Transactional
public class TestServiceIntegrationTest {
    
    @Autowired
    private TestService testService;
    
    @Autowired
    private SalesSummaryRepository salesSummaryRepository;
    
    @Autowird
    private OrderRepository orderRepository;
    
    @Autowird
    private ProductRepository productRepository;
    
    @Autowired
    private CartRepository cartRepository;
    
    //...
    
    //paymentDTO에 담아 보낼 상품 엔티티
    private Product product;
    
    //주문 요청하는 사용자 엔티티
    private Member member;
    
    @BeforeEach
    void init() {
        //테스트에 필요한 데이터 생성 및 저장
        //여기에서 생성된 데이터로 product, member 할당
    }
    
    @Test
    void rabbitMQTest() {
        PaymentDTO paymentDTO = OrderFixture.createPaymentDTO(product);
        int productStockFixture = product.getStock() - paymentDTO.getOrderCount();
        
        assertDoesNotThrow(() -> testService.payemnt(paymentDTO));
        
        //저장된 order 데이터 조회
        List<Order> saveOrderList = orderRepository.findAll();
        assertFalse(saveOrderList.isEmpty());
        assertEquals(1, saveOrderList.size());
        
        //저장된 order 데이터 검증
        //...
        
        await()
            .atMost(10, TimeUnit.SECONDS) // 총 대기시간 ( 10초 )
            .pollInterval(200, TimeUnit.MILLISECONDS) // 재시도 수행간격 ( 200ms )
            .untilAsserted(() -> {
                //저장된 SalesSummary 조회
                List<SalesSummary> saveSummaryList = salesSummaryRepository.findAll();
                assertFalse(saveSummaryList.isEmpty());
        
                //저장된 SalesSummary 검증
                //...
        
                //수정된 Product 재고 검증
                Product patchProduct = productRepository.findById(productId).orElse(null);
                assertNotNull(patchProduct);
                assertEquals(productStockFixture, patchProduct.getStock());
        
                //Cart 삭제 검증
                Cart deleteCart = cartRepository.findCartByUserId(member.getUserId());
                assertNull(deleteCart);
            })
    }
}

위와 같이 작성하게 되면 최대 10초까지 대기하게 되며, 200ms마다 재시도를 수행하게 된다.

그리고 그 과정에서 untilAsserted 내부의 검증 로직이 모두 통과되는 순간 즉시 종료되게 된다.

그럼 인프라 상황이 좋아서 1초만에 마무리 되더라도 정적인 설정 시간을 대기할 필요 없이 1초로 마무리될 수 있고,

연결 이슈로 인해 지연시간이 발생하더라도 atMost에 설정해둔 시간 내에만 처리된다면 정상적으로 처리할 수 있기 때문에 문제가 없다.

그런 경우는 잘 없겠지만 정말 극단적으로 30초의 총 대기시간을 주고 처리하게 되면 왠만한 이슈에는 대응할 수 있으며, '테스트가 통과되는 시점' 이라는 조건에 의해 불필요한 예비 시간까지 대기하도록 할 필요가 없어진다.

그럼 여기까지 했을 때 1번 문제는 해결됐다.

문제 원인 2) Consumer가 테스트 코드에서 저장해둔 데이터에 접근하지 못하는 문제

하지만 그럼에도 위 테스트는 통과되지 않을 것이다.

예시 가정 상, SalesSummary는 독립적으로 다른 데이터를 참조하지 않는 테이블이기 때문에 SalesSummary 검증은 통과하게 될 것이고 insert 역시 잘 수행될 것이다.

하지만 기존 데이터가 필요한 Product, Cart는 절대 통과되지 않는다.

이유는 테스트 Transaction을 RabbitMQ가 공유하지 않기 때문이다.

테스트 클래스에는 @Transactional이 정의되어 있다.

각 테스트 메서드가 종료되었을 때마다 데이터가 롤백되기 때문에 독립성을 보장할 수 있다.

특히 BeforeEach와 함께 사용할 경우 매 테스트 메서드마다 일관된 상태로 테스트를 수행하기 때문에 유리하다.

하지만 RabbitMQ가 이 트랜잭션을 공유해야만 Product, Cart에 대한 조회가 정상적으로 이루어지고 처리할 수 있게 되는데, RabbitMQ는 Application 외부에서 비동기적으로 동작하는 Message Broker이므로 트랜잭션을 공유할 수 없다.

메시지 발행 시점까지는 같은 트랜잭션에 위치하지만, 메시지가 RabbitMQ에 전달되는 순간 외부로 이동하는 것이기 때문에 트랜잭션에서 벗어나게 되고 Consumer에 메시지를 전달하더라도 다시 트랜잭션에 들어가지 못하므로 별개의 트랜잭션에서 동작하게 된다.

즉, 서버 -> 외부(Message Broker(RabbitMQ)) -> 서버 (Consumer) 순서로 처리되기 때문에 외부를 거쳐 호출된 consumer는 다른 트랜잭션을 가질 수 밖에 없다.

@Transactional을 테스트 클래스에서 정의하는 경우 save()한 데이터가 commit 되지 않기 때문에 consumer에서는 데이터에 접근할 수 없게 된다.

좀 더 명확하게 정리해보자면 테스트 환경과 consumer의 트랜잭션이 분리되었기 때문에 Transaction Isolation에 의해 consumer가 아직 commit 되지 않은 데이터에 접근할 수 없는 것이다.

이 격리 수준을 READ_UNCOMMITED로 수정하는 방법도 있지만, 일반적으로 다른 트랜잭션의 접근을 허용하는 것은 데이터 정합성 문제를 발생시킬 수 있기 때문에 특수한 상황이 아니라면 좋은 방법은 아니다.

특히나, 테스트만을 위해 격리 수준을 수정하는 건 더욱 더.

그럼 결과적으로 원인은 다른 트랜잭션을 사용하는 것이 원인이므로 테스트 환경의 트랜잭션을 제거하면 된다.

해결 방안

정말 단순하게 테스트 클래스 단위에 정의되어 있는 @Transactional을 제거해주면 된다.

단, 이 경우 @BeforeEach에서 저장했던 데이터를 제거해주는 처리가 필요하다.

이 처리를 @AfterEach를 통해 삭제하도록 해주면 @Transactional과 동일한 효과를 볼 수 있다.

물론, 애초에 데이터가 아예 없는 데이터베이스를 활용하는 경우라는 전제하에.

데이터가 이미 존재하는 테이블을 대상으로 테스트 한다면 좀 더 신경써야 하는 부분이 많아진다.

import static org.awaitility.Awaitility.await;

@SpringBootTest(classes = TestApplication.class)
@ActiveProfiles("test")
@Transactional
public class TestServiceIntegrationTest {
    
    @Autowired
    private TestService testService;
    
    @Autowired
    private SalesSummaryRepository salesSummaryRepository;
    
    @Autowird
    private OrderRepository orderRepository;
    
    @Autowird
    private ProductRepository productRepository;
    
    @Autowired
    private CartRepository cartRepository;
    
    //...
    
    //paymentDTO에 담아 보낼 상품 엔티티
    private Product product;
    
    //주문 요청하는 사용자 엔티티
    private Member member;
    
    @BeforeEach
    void init() {
        //테스트에 필요한 데이터 생성 및 저장
        //여기에서 생성된 데이터로 product, member 할당
    }
    
    @AfterEach
    void cleanUp() {
        //Product, member 데이터 제거
    }
    
    @Test
    void rabbitMQTest() {
        PaymentDTO paymentDTO = OrderFixture.createPaymentDTO(product);
        int productStockFixture = product.getStock() - paymentDTO.getOrderCount();
        
        assertDoesNotThrow(() -> testService.payemnt(paymentDTO));
        
        //저장된 order 데이터 조회
        List<Order> saveOrderList = orderRepository.findAll();
        assertFalse(saveOrderList.isEmpty());
        assertEquals(1, saveOrderList.size());
        
        //저장된 order 데이터 검증
        //...
        
        await()
            .atMost(10, TimeUnit.SECONDS) // 총 대기시간 ( 10초 )
            .pollInterval(200, TimeUnit.MILLISECONDS) // 재시도 수행간격 ( 200ms )
            .untilAsserted(() -> {
                //저장된 SalesSummary 조회
                List<SalesSummary> saveSummaryList = salesSummaryRepository.findAll();
                assertFalse(saveSummaryList.isEmpty());
        
                //저장된 SalesSummary 검증
                //...
        
                //수정된 Product 재고 검증
                Product patchProduct = productRepository.findById(productId).orElse(null);
                assertNotNull(patchProduct);
                assertEquals(productStockFixture, patchProduct.getStock());
        
                //Cart 삭제 검증
                Cart deleteCart = cartRepository.findCartByUserId(member.getUserId());
                assertNull(deleteCart);
            })
    }
}

문제 원인 3) RabbitMQ를 통해 insert 된 데이터가 롤백되지 않는 문제

이건 '롤백'되지 않는 다는 말이 조금 애매하긴 하다.

애초에 테스트 클래스에서 @Transactional을 사용하지 않는다면 @AfterEach에서 애초에 제거하도록 해야 하기 때문이다.

insert 된 데이터라고 한다면 예시에서 SalesSummary 데이터가 있다.

이건 BeforeEach에서 저장하는 데이터가 아니기 때문에 애초에 AfterEach로 제거해야할 필요가 있다.

이 문제를 따로 정리하는 이유는 2번 문제 초반쪽에서 정리했다시피 consumer에서 다른 테이블 데이터를 조회한다거나 하는 참조가 필요없고 테이블 자체도 외래키가 없는 독립적인 테이블이라면, 데이터가 정상적으로 저장되기 때문이다.

즉, @Transactional이 테스트 클래스에 정의되어 있더라도 저장이 되는 케이스라는 것이다.

이런 경우 롤백이 되지 않고 데이터가 남아있기 때문에 따로 제거해줄 필요가 있다.

데이터가 남아있는건 2번까지 이해했다면 당연한 문제다.

간결하게 테스트 트랜잭션을 Transaction1이라고 하고, consumer의 트랜잭션을 Transaction2라고 해보자.

그럼 Transaction1은 자기가 담당한 데이터만 롤백한다.

하지만 SalesSummary는 Transaction2가 처리했고, 이 트랜잭션은 롤백을 수행해야 한다는걸 알지 못하기 때문에 롤백을 처리하지 않게 된다.

즉, 테스트 트랜잭션이 자기가 알고있는것만 롤백할 수 있기 때문에 consumer가 insert 한 데이터는 제거할 수 없다는 말이 된다.

해결방안

결국 이 문제의 해결은 매우 간단하다.

해당 테스트 메서드에서 검증이 끝난 이후 consumer가 insert 한 테이블의 데이터를 제거해주거나,

@AfterEach에서 해당 데이터의 제거까지 처리하도록 작성해주면 된다.

이건 간단하기 때문에 해결 코드는 생략.

'Project&문제해결' 카테고리의 다른 글

[ Spring ] 통합 테스트 중 JPA Lazy Loading 문제 (0)	2025.07.01
Oracle의 다중 insert 처리 시 Sequence 비정상 동작 문제 해결 (0)	2024.10.18
Kotlin build.gradle.kts 빨간 밑줄 문제 해결 (0)	2024.08.06
React 권한 페이지 접근 렌더링 문제 해결 (0)	2024.04.12
JWT&Redis 로그인에 대한 고민 (0)	2023.10.25

공부하는블로그