From b0456c18a26d1d51f80d0bf7941af0032a3ccb9c Mon Sep 17 00:00:00 2001 From: pricelees Date: Fri, 27 Jun 2025 14:13:33 +0900 Subject: [PATCH] =?UTF-8?q?test:=20Dead=20Letter=20Topic=20=ED=94=84?= =?UTF-8?q?=EB=A1=9C=EB=93=80=EC=84=9C=20=EC=9E=AC=EC=8B=9C=EB=8F=84=20?= =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ref: 1f60a0ff --- .../kafka/producer/DLTProducerTest.java | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 src/test/java/com/sangdol/consumer/infrastructure/kafka/producer/DLTProducerTest.java diff --git a/src/test/java/com/sangdol/consumer/infrastructure/kafka/producer/DLTProducerTest.java b/src/test/java/com/sangdol/consumer/infrastructure/kafka/producer/DLTProducerTest.java new file mode 100644 index 0000000..2ffd6e6 --- /dev/null +++ b/src/test/java/com/sangdol/consumer/infrastructure/kafka/producer/DLTProducerTest.java @@ -0,0 +1,78 @@ +package com.sangdol.consumer.infrastructure.kafka.producer; + +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +import java.time.Instant; +import java.util.concurrent.Future; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import com.sangdol.consumer.domain.TestRecord; +import com.sangdol.consumer.infrastructure.kafka.common.serialize.JsonObjectMapper; + +/** + * Spring Retryable 동작 및 로그 출력 확인용 테스트 + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +class DLTProducerTest { + + @MockitoBean + private KafkaProducer kafkaProducer; + + @Autowired + private DLTProducer dltProducer; + + @Autowired + private JsonObjectMapper jsonObjectMapper; + + @DisplayName("DLT 메시지 전송이 첫 번째 시도에서는 실패하고, 두 번째 시도에서 성공한다.") + @Test + void dltRetryTest_whenSucceedSecondTrial() throws Exception { + // given + String key = "key"; + TestRecord value = new TestRecord(1234152, "topic1", Instant.now().toString()); + RuntimeException origin = new RuntimeException("Intended Exception for Test"); + Future mockfuture = mock(Future.class); + when(kafkaProducer.send(any())).thenReturn(mockfuture); + when(mockfuture.get()) + .thenThrow(origin) + .thenReturn(mock(RecordMetadata.class)); + + // when + assertThatCode(() -> dltProducer.sendToDeadLetterTopic(key, value, origin)) + .doesNotThrowAnyException(); + } + + @DisplayName("DLT 메시지 전송에 실패한다.") + @Test + void dltRetryTest_whenFailedAllTries() throws Exception { + // given + String key = "key"; + TestRecord value = new TestRecord(1234152, "topic1", Instant.now().toString()); + RuntimeException origin = new RuntimeException("Intended Exception for Test"); + Future mockfuture = mock(Future.class); + when(kafkaProducer.send(any())).thenReturn(mockfuture); + when(mockfuture.get()).thenThrow(origin); + + // when + dltProducer.sendToDeadLetterTopic(key, value, origin); + + // then + verify(kafkaProducer, times(5)).send(any()); + + assertThat(dltProducer.getFailed()).hasSize(1); + + DLTProducer.FailedDltMessage failedDltMessage = dltProducer.getFailed().get(0); + assertThat(failedDltMessage.key()).isEqualTo(key); + assertThat(failedDltMessage.message().exception()).isEqualTo(origin.getMessage()); + assertThat(failedDltMessage.message().message()).isEqualTo(jsonObjectMapper.serialize(value)); + } +} \ No newline at end of file