시스템 장애를 처리하기 위해 재처리 및 DLQ 적용하기

@Configuration
public class KafkaConsumerConfig {
    // ...

    @Bean
    public DefaultErrorHandler couponIssueErrorHandler(final KafkaTemplate<String, Object> kafkaTemplate) {
        // 모든 재시도가 실패했을 때 메시지를 DLQ 토픽으로 전송하는 Recoverer 정의
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, exception) -> {
                    log.error("[COUPON_ISSUE_DLQ] 쿠폰 발급에 대한 재시도 횟수를 초과하여 메시지를 DLQ로 전송합니다. " +
                                    "topic: {}, partition: {},offset: {}, error: {}",
                            record.topic(), record.partition(), record.offset(), exception.getMessage());

                    return new org.apache.kafka.common.TopicPartition(
                            record.topic() + "-dlq",
                            record.partition()
                    );
                });

        // 지수 백오프 전략 설정: 초기 2초 간격, 2.0배수로 최대 5회 재시도
        ExponentialBackOff backOff = new ExponentialBackOff(2000L, 2.0);
        backOff.setMaxAttempts(5);

        return new DefaultErrorHandler(recoverer, backOff);
    }
}

@Component
public class CouponIssueConsumer {
    // ...
    @KafkaListener(
            id = "dlq-consumer",
            topics = "${kafka.topic.coupon-issue.name}-dlq",
            groupId = "coupon-issue-dlq-group",
            autoStartup = "true"
    )
    public void listenDlq(final CouponIssueMessage message, final Acknowledgment ack) {
        log.info("[COUPON_ISSUE_DLQ_RETRY] 쿠폰 발급 실패 건에 대한 재처리를 시작합니다. userId={}", message.userId());
        try {
            // DLQ 메시지 재처리 시도
            couponIssueService.issue(message.userId(), message.couponId());
            ack.acknowledge();
            log.info("[COUPON_ISSUE_DLQ_SUCCESS] DLQ 재처리가 성공적으로 완료되었습니다. userId={}", message.userId());
        } catch (Exception e) {
            // DLQ에서도 실패 시 최종 실패 이력 저장
            log.error("[COUPON_ISSUE_DLQ_FAIL] DLQ 재처리가 최종적으로 실패했습니다. 실패 이력을 DB에 저장합니다. userId={}, error={}",
                    message.userId(), e.getMessage());
            String payload = convertToJson(message);
            couponIssueService.saveFailedEvent(message.userId(), message.couponId(), e.getMessage(), payload);
            ack.acknowledge();
        }
    }
    // ...
}

@ActiveProfiles("test")
@SpringBootTest(properties = {
        //...
})
@EmbeddedKafka(
        //...
)
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
class CouponIssueConsumerTest {
    // ...

    // Case 1: 일시적 장애 (재시도 성공)
    @Test
    @DisplayName("일시적인 장애로 처음에 실패하더라도, 재시도 횟수 내에 성공하면 쿠폰이 정상 발급되고 DLQ로 가지 않는다.")
    void should_issue_successfully_after_retries_on_temporary_failure() {
      // given
      final Coupon coupon = createCoupon();
      final UUID userId = UUID.randomUUID();
      final CouponIssueMessage message = new CouponIssueMessage(userId, coupon.getId());

      // NOTE: 메시지 전송 전에 미리 정의함
      doThrow(new RuntimeException("Temporary DB Error 1")).doThrow(new RuntimeException("Temporary DB Error 2")).doCallRealMethod().when(couponIssueService).issue(any(UUID.class), any(UUID.class));

      // when
      kafkaTemplate.send(issueTopic, message);

      // NOTE: 재시도 횟수와 시간 간격에 따라 timeout 조정 필요
      await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
        boolean exists = issuedCouponRepository.existsByUserIdAndCouponId(userId, coupon.getId());
        assertThat(exists).isTrue();
      });
      verify(couponIssueService, times(3)).issue(any(UUID.class), any(UUID.class));
    }

    // Case 2: 지속적 장애 (DLQ 전송)
    @Test
    @DisplayName("지속적인 시스템 장애로 모든 재시도가 실패하면, 메시지는 최종적으로 DLQ로 전송된다.")
    void should_send_to_dlq_when_all_retries_fail() {
        // given
        final Coupon coupon = createCoupon();
        final UUID userId = UUID.randomUUID();
        final CouponIssueMessage message = new CouponIssueMessage(userId, coupon.getId());

        doThrow(new RuntimeException("Main Consumer Error"))
                .doThrow(new RuntimeException("Main Consumer Error"))
                .doThrow(new RuntimeException("Main Consumer Error"))
                .doThrow(new RuntimeException("Main Consumer Error"))
                .doThrow(new RuntimeException("Main Consumer Error"))
                .doCallRealMethod() // 6번째 호출(DLQ)에서는 정상 실행
                .when(couponIssueService).issue(any(UUID.class), any(UUID.class));

        // when
        kafkaTemplate.send(issueTopic, message);

        // then
        await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
            boolean exists = issuedCouponRepository.existsByUserIdAndCouponId(userId, coupon.getId());
            assertThat(exists).isTrue();
        });

        // NOTE: 최초 실행 1회 + 재시도 5회 = 총 6회
        verify(couponIssueService, times(6)).issue(any(UUID.class), any(UUID.class));
    }

    // Case 3: 영구 장애 (DB 기록)
    @Test
    @DisplayName("DLQ 재처리마저 실패하면 실패 이력이 DB에 저장된다.")
    void should_save_failed_event_when_dlq_retry_fails() {
        // given
        final Coupon coupon = createCoupon();
        final UUID userId = UUID.randomUUID();
        final CouponIssueMessage message = new CouponIssueMessage(userId, coupon.getId());

        // 시나리오: 메인 리스너(5회) + DLQ 리스너(1회) 모두 실패
        doThrow(new RuntimeException("Persistent Error"))
                .when(couponIssueService).issue(any(UUID.class), any(UUID.class));

        // when
        kafkaTemplate.send(issueTopic, message);

        // then
        await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> {
            long count = failedEventRepository.count();
            assertThat(count).isEqualTo(1);
        });

        // 1. 발급 시도 횟수 검증 - 메인 횟수(초기 시도 + 재시도 횟수) + DLQ 1회
        verify(couponIssueService, atLeast(6)).issue(any(UUID.class), any(UUID.class));
        verify(couponIssueService, times(1))
                .saveFailedEvent(eq(userId), eq(coupon.getId()), anyString(), anyString());
    }
    // ...
}