SpringKafka Consumer @RetryableTopic Anotasyonu -Integration Test

Örnek

Test iskeleti şöyle

import org.springframework.kafka.test.utils.ContainerTestUtils;

@Slf4j
@EnableKafka
@SpringBootTest
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
@EmbeddedKafka(partitions = 1,
        brokerProperties = {"listeners=" + "${kafka.broker-properties.listeners}",
                "port=" + "${kafka.broker-properties.port}"},
        controlledShutdown = true,
        topics = {"test", "test-retry-0", "test-retry-1", "test-retry-2", "test-dlt"}
)
@ActiveProfiles("test")
class CustomEventConsumerIntegrationTest {

  @Autowired
  private KafkaListenerEndpointRegistry registry;

  @Autowired
  private EmbeddedKafkaBroker embeddedKafkaBroker;

  @Autowired
  private KafkaTemplate<String, CustomEvent> testKafkaTemplate;

  @Value("${retry.attempts}")
  private int maxRetries;
  @MockBean
  CustomEventHandler customEventHandler;
  ...
}

Açıklaması şöyle

When using an embedded Kafka broker, it is important to mention the topics to be created. They will not be created automatically. In this case, we are creating 4 topics, namely:

"test", "test-retry-0", "test-retry-1", "test-dlt"

Her test için şöyle yaparız

import org.springframework.kafka.test.utils.ContainerTestUtils;

@BeforeEach
void setUp() {
  //wait for partitions to be assigned
  registry.getListenerContainers().forEach(container ->
    ContainerTestUtils.waitForAssignment(container, 
      embeddedKafkaBroker.getPartitionsPerTopic()));
  log.info("Partitioned Assigned. Starting tests");
}

@AfterEach
void tearDown() {
  embeddedKafkaBroker.destroy();
}

Daha sonra çeşitli testler burada anlatılıyor. Kod ise burada. Retry olmadığını test eden bir test şöyle

@Test
void test_should_not_retry_if_consumption_is_successful()  {
  CustomEvent event = new CustomEvent("Hello");
  // GIVEN
  doNothing().when(customEventHandler).handleEvent(any(CustomEvent.class));

  // WHEN
  testKafkaTemplate.send("test", event).get();

  // THEN
  verify(customEventHandler, timeout(2000).times(1)).handleEvent(any(CustomEvent.class));
  verify(customEventHandler, timeout(2000).times(0))
    .handleEventFromDlt(any(CustomEvent.class));
}