Kafka + Redis + MySQL 환경을 Testcontainers로 통합 테스트하기

목차

1. 왜 통합 테스트가 필요했는가?

2. Testcontainers 기반 인프라

3. 테스트

4. 회고

 

1. 왜 통합 테스트가 필요했는가?

이번 일정 관리 프로젝트는 기획 단계부터 Kafka, Redis, RDS(MySQL) 등 다양한 외부 시스템과의 연동을 기반으로 설계되었습니다. 특히 Kafka를 통한 이벤트 기반 구조, Redis를 활용한 캐시 및 분산 락 처리, RDS와의 스케줄러 기반 트랜잭션 흐름 등은 단순한 로직 검증을 넘어서, 실제 환경과 유사한 상황에서 전체 동작 흐름을 검증할 필요가 있었습니다.

 

처음부터 mock/stub 기반 단위 테스트 대신, "외부 시스템을 포함한 통합적인 테스트 환경을 어떻게 구성할 것인가?"가 주요 고민이었습니다. 실제 테스트 중 직면했던 이슈는 다음과 같습니다:

• Kafka → Consumer → DB 저장까지 흐름을 단위로 쪼개서 보기 어려움
• DLQ로 전송된 메시지를 실제로 재처리하는지 확인하기 위해선 Kafka + 스케줄러 + DB 상태를 함께 확인해야 했음
• Redis의 TTL, 캐시 만료 시점, 분산락 충돌 등을 정확히 재현할 수 있는 테스트 환경이 필요
• ShedLock을 사용한 스케줄러가 멀티 인스턴스 환경에서 중복 실행을 막는지 실제로 검증하고 싶었음

이런 복잡한 테스트 시나리오를 커버하기 위해, Testcontainers를 활용해 통합 테스트 환경을 구성하게 되었습니다. 이를 통해 Kafka, MySQL, Redis를 모두 컨테이너로 띄워, CI/CD에서도 실제 인프라와 유사한 조건에서 테스트를 수행할 수 있게 되었습니다.

2. Testcontainers 기반 인프라

Testcontainers를 사용하여 Kafka, Redis, MySQL 컨테이너를 통합 테스트 환경에 구성했습니다. @Container 어노테이션을 통해 각 서비스를 선언하고, @DynamicPropertySource를 활용하여 Spring Boot 애플리케이션이 이 컨테이너들에 연결되도록 설정했습니다.

 

Kafka 설정

Kafka 컨테이너는 confluentinc/cp-kafka:7.3.0 이미지를 사용했습니다. bootstrap-servers 속성을 동적으로 설정하여 Spring이 Kafka에 연결되도록 합니다. 또한, 테스트 실행 전 AdminClient를 사용하여 필요한 토픽들을 미리 생성했습니다. 이는 Kafka 컨테이너가 완전히 준비되기 전에 토픽 생성이 시도될 경우 발생할 수 있는 지연 문제를 방지하기 위함입니다.

@Container
static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.3.0"));

@DynamicPropertySource
static void setKafkaProps(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", () -> kafka.getBootstrapServers()); [cite: 131, 177]
}

@BeforeAll
static void setUp() throws Exception {
    Map<String, Object> config = new HashMap<>(); [cite: 153]
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); [cite: 154]
    try (AdminClient adminClient = AdminClient.create(config)) { [cite: 154]
        List<NewTopic> topics = List.of(
                new NewTopic("member-signup-events", 1, (short) 1), [cite: 154]
                new NewTopic("member-signup-events.DLQ", 1, (short) 1), [cite: 154]
                new NewTopic("member-signup.retry.5s", 1, (short) 1), [cite: 154]
                new NewTopic("notification-events.retry.final", 1, (short) 1) [cite: 155]
        );
        adminClient.createTopics(topics).all().get(); [cite: 156]
    }
}

 

Redis 설정

Redis 컨테이너는 redis:7.0 이미지를 사용하며, 포트 6379를 노출합니다. Spring Data Redis가 Redis 컨테이너에 연결되도록 host와 port 속성을 동적으로 설정합니다. ShedLock을 위한 LockProvider도 Redis를 사용하여 분산 락을 구현합니다.

@Container
static final RedisContainer redis = new RedisContainer(DockerImageName.parse("redis:7.0"))
        .withExposedPorts(6379)
        .waitingFor(Wait.forListeningPort())
        .withStartupAttempts(3); [cite: 129]

@DynamicPropertySource
static void setRedisProps(DynamicPropertyRegistry registry) {
    registry.add("spring.data.redis.host", redis::getHost); [cite: 132]
    registry.add("spring.data.redis.port", () -> redis.getMappedPort(6379)); [cite: 132]
}

@TestConfiguration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "PT30S")
static class ShedLockTestConfig {
    @Bean
    public LockProvider lockProvider(RedisConnectionFactory redisConnectionFactory) {
        return new RedisLockProvider(redisConnectionFactory,"shedlock"); [cite: 152]
    }
}

 

MySQL 설정

MySQL 컨테이너는 mysql:8.0 이미지를 사용합니다. 데이터베이스 이름, 사용자 이름, 비밀번호를 설정하며, Spring Data JPA가 MySQL 컨테이너에 연결되도록 JDBC URL, username, password 속성을 동적으로 설정합니다.

@Container
static MySQLContainer<?> mysql = new MySQLContainer<>(DockerImageName.parse("mysql:8.0"))
        .withDatabaseName("testdb")
        .withUsername("test")
        .withPassword("test"); [cite: 176]

@DynamicPropertySource
static void mysqlProps(DynamicPropertyRegistry registry) {
    registry.add("spring.datasource.url", mysql::getJdbcUrl); [cite: 178]
    registry.add("spring.datasource.username", mysql::getUsername); [cite: 178]
    registry.add("spring.datasource.password", mysql::getPassword); [cite: 178]
}

3. 테스트

통합 테스트 환경에서 다음 시나리오들을 검증했습니다.

회원가입 및 알림 성공 테스트

회원가입 이벤트가 Kafka를 통해 전송되고, Consumer에서 이를 처리하여 알림이 정상적으로 DB에 저장되는지 확인합니다.

@Test
@Disabled
@DisplayName("회원가입후 정상적으로 카프카에 송신이 되고 알림내역이 정상적으로 저장이 되는가?")
void memberSignUpNotificationSuccessTest(){
    MemberModel member = MemberModel.builder()
            .userId("testUser")
            .userEmail("test@test.com")
            .userPhone("010-0000-0000")
            .userName("tester")
            .password("12345")
            .roles(Roles.ROLE_USER)
            .build(); [cite: 205]
    MemberModel saved = memberService.createMember(member); [cite: 206]

    MemberSignUpKafkaEvent event = MemberSignUpKafkaEvent.builder()
            .receiverId(123L)
            .username("testuser")
            .email("test@test.com")
            .build(); [cite: 206]
    kafkaTemplate.send("member-signup-events", event); [cite: 207]

    await().atMost(10, TimeUnit.SECONDS)
            .untilAsserted(() -> {
                var notiList = notificationService.getNotificationsByUserId(saved.getId());
                assertThat(notiList).isNotEmpty(); [cite: 208]
                assertThat(notiList.get(0).getMessage()).contains("환영합니다"); [cite: 208]
            });
}

 

Kafka DLQ (Dead Letter Queue) 처리 테스트

Consumer에서 메시지 처리 실패 시, 해당 메시지가 DLQ로 올바르게 전송되는지 확인합니다.

@Test
@DisplayName("회원 컨슈머에서 실패를 했을경우에 DLQ로 진행이 되는 경우")
public void MemberSignUpDLQTest1() {
    dlqTestConsumer.clear(); [cite: 210]
    MemberSignUpKafkaEvent event = MemberSignUpKafkaEvent.builder()
            .receiverId(999L)
            .message("강제실패 알림")
            .username("dlquser")
            .email("fail@test.com")
            .build(); [cite: 211]
    kafkaTemplate.send("member-signup-events", event); [cite: 212]

    await().atMost(10, TimeUnit.SECONDS)
            .untilAsserted(() -> {
                List<MemberSignUpKafkaEvent> messages = dlqTestConsumer.getMemberDlqMessages();
                assertThat(messages).isNotEmpty(); [cite: 212]
                assertThat(messages.get(0).getEmail()).isEqualTo("fail@test.com"); [cite: 212]
            });
}

 

DLQ 재처리 및 실패 메시지 복구 테스트

DLQ로 전송된 실패 메시지가 FailedMessageService에 저장되고, 스케줄러에 의해 성공적으로 재처리되어 원래의 알림 기능이 작동하는지 확인합니다.

 

@Test
@DisplayName("회원 DLQ로 보내진 후 실패이력을 저장을 하고 재처리 하기")
public void MemberSignUpDLQTest2(){
    MemberSignUpKafkaEvent event = MemberSignUpKafkaEvent.builder()
            .receiverId(999L)
            .message("강제실패 알림")
            .username("retryUser")
            .email("fail@test.com")
            .build(); [cite: 213, 214]
    kafkaTemplate.send("member-signup-events", event); [cite: 215]

    await().atMost(10, TimeUnit.SECONDS)
            .untilAsserted(() -> {
                var failList = failedMessageService.findByResolvedFalse();
                assertThat(failList).isNotEmpty(); [cite: 215]
            });
    retryScheduler.retryMemberSignUps(); [cite: 216]
    await().atMost(10, TimeUnit.SECONDS)
            .untilAsserted(() -> {
                List<MemberSignUpKafkaEvent> messages = dlqTestConsumer.getMemberDlqMessages();
                assertThat(messages).isNotEmpty(); [cite: 217]
                assertThat(messages.get(0).getEmail()).isEqualTo("fail@test.com"); [cite: 217]
            });
}

 

ShedLock 기반 스케줄러 중복 실행 방지 테스트

@EnableSchedulerLock과 Redis LockProvider를 사용하여 스케줄러가 여러 번 호출되어도 실제 로직은 한 번만 실행되는지 검증합니다.

@Test
void shedlock_MemberSchedulerTest() {
    Object bean = context.getBean(MemberSignUpDlqRetryScheduler.class); [cite: 161]
    System.out.println(bean.getClass().getName()); // 프록시 클래스명 출력 

    MemberSignUpDlqRetryScheduler.EXECUTION_COUNT = 0; [cite: 162]
    ((MemberSignUpDlqRetryScheduler)bean).retryMemberSignUps(); [cite: 163]
    ((MemberSignUpDlqRetryScheduler)bean).retryMemberSignUps(); [cite: 163]

    System.out.println("EXECUTION_COUNT = " + MemberSignUpDlqRetryScheduler.EXECUTION_COUNT); [cite: 164]
    assertEquals(1, MemberSignUpDlqRetryScheduler.EXECUTION_COUNT); [cite: 164]
}

@Test
public void shedlock_scheduleSchedulerTest(){
    Object bean = context.getBean(NotificationDlqRetryScheduler.class); [cite: 165]
    System.out.println(bean.getClass().getName()); // 프록시 클래스명 출력 

    NotificationDlqRetryScheduler.EXECUTION_COUNT = 0; [cite: 166]
    ((NotificationDlqRetryScheduler)bean).retryNotifications(); [cite: 167]
    ((NotificationDlqRetryScheduler)bean).retryNotifications(); [cite: 167]

    System.out.println("EXECUTION_COUNT = " + NotificationDlqRetryScheduler.EXECUTION_COUNT); [cite: 168]
    assertEquals(1, NotificationDlqRetryScheduler.EXECUTION_COUNT); [cite: 168]
}

Outbox 패턴을 통한 이벤트 발행 검증

Outbox 패턴을 사용하여 이벤트가 DB에 먼저 저장된 후 Kafka로 발행되고, 최종적으로 Consumer를 통해 알림이 저장되는 전체 흐름을 검증합니다.

@Test
@DisplayName("Outbox → Kafka → Consumer: 회원가입 알림 전파 검증")
void memberOutboxToKafkaIntegrationTest() {
    MemberSignUpKafkaEvent event = MemberSignUpKafkaEvent
            .of(555L,"outboxUser","outbox@test.com"); [cite: 233]
    outboxEventService.saveEvent(
            event,
            AggregateType.MEMBER.name(),
            "555",
            EventType.SIGNED_UP_WELCOME.name()
    ); [cite: 234]
    outboxEventPublisher.publishOutboxEvents(); [cite: 235]
    var events = outboxEventRepository.findAll();
    assertThat(events).isNotEmpty(); [cite: 236]
    assertThat(events.get(0).getSent()).isTrue(); [cite: 236]
    assertThat(events.get(0).getSentAt()).isNotNull(); [cite: 236]
    await().atMost(15, TimeUnit.SECONDS)
            .pollInterval(100, TimeUnit.MILLISECONDS)
            .untilAsserted(() -> {
                var notiList = notificationService.getNotificationsByUserId(555L);
                assertThat(notiList).isNotEmpty(); [cite: 237]
                assertThat(notiList.get(0).getMessage()).contains("🎉 환영합니다, outboxUser님! 회원가입이 완료되었습니다."); [cite: 238]
            });
}

4. 회고

이번 일정 관리 프로젝트에서 Kafka, Redis, MySQL을 포함한 통합 테스트 환경을 Testcontainers 기반으로 구축하면서 분산 시스템의 실제 환경을 효과적으로 시뮬레이션할 수 있었습니다.

도입 배경 및 효과

Kafka 기반 이벤트 처리, Redis 분산 락, MySQL 트랜잭션 흐름 등은 단위 테스트만으로 검증이 어려웠습니다. 특히 DLQ 처리, Redis TTL, ShedLock을 통한 중복 실행 방지 같은 시나리오는 실제 인프라 환경이 필요했습니다.

Testcontainers를 도입함으로써 Kafka, Redis, MySQL 컨테이너를 활용한 통합 테스트가 가능해졌고, CI/CD 환경에서도 실제와 유사한 조건으로 테스트 신뢰도를 높일 수 있었습니다.

주요 이슈 및 해결

• Kafka 토픽 지연 생성 문제: AdminClient로 사전 생성하여 해결.
• Spring과 컨테이너 연동: @DynamicPropertySource로 동적 포트 매핑 처리.
• DLQ 및 재처리 로직 검증: 메시지 실패 → DLQ 전송 → 스케줄러 기반 재처리 흐름까지 테스트.

개선점 및 과제

• 테스트 속도 최적화: 컨테이너 공유 전략을 더 고도화해 수행 시간 단축 필요.