Kafka的非阻塞重试是通过为主题配置重试主题来实现的。如果需要，还可以配置额外的死信主题。如果所有重试都耗尽，事件将被转发到DLT。在公共领域中有很多资源可用于了解技术细节。对于代码中的重试机制编写集成测试确实是一项具有挑战性的工作。以下是一些测试方法，可以用来验证重试机制的正确性：
- 验证事件已经按照所需的次数进行了重试：
在测试中，模拟一个会触发重试的事件，并设置重试次数为所需的次数。
使用断言来验证事件是否被重试了指定的次数。
- 验证只有在特定的异常发生时才进行重试，而不是其他异常：
在测试中，模拟不同的异常情况，包括需要重试的异常和不需要重试的异常。
使用断言来验证只有特定的异常触发了重试，而其他异常没有触发重试。
- 验证如果前一次重试已经解决了异常，不会进行另一次重试：
在测试中，模拟一个会触发重试的事件，并在每次重试之间解决异常。
使用断言来验证只有在异常没有被解决的情况下才进行重试。
- 验证在前面的 (n-1) 次重试失败后，第 n 次重试成功：
在测试中，模拟一个会触发重试的事件，并设置重试次数为 n。
使用断言来验证在前面的 (n-1) 次重试失败后，第 n 次重试成功。
- 验证如果所有的重试尝试都失败，事件是否已经发送到了死信队列：
- 在测试中，模拟一个会触发重试的事件，并设置重试次数为一个较小的值。
- 使用断言来验证当所有的重试尝试都失败后，事件是否已经发送到了死信队列。
设置可重试的消费者
@Slf4j
@Component
@RequiredArgsConstructor
public class CustomEventConsumer {
private final CustomEventHandler handler;
@RetryableTopic(attempts = "${retry.attempts}",
backoff = @Backoff(
delayExpression = "${retry.delay}",
multiplierExpression = "${retry.delay.multiplier}"
),
topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
dltStrategy = FAIL_ON_ERROR,
autoStartDltHandler = "true",
autoCreateTopics = "false",
include = {CustomRetryableException.class})
@KafkaListener(topics = "${topic}", id = "${default-consumer-group:default}")
public void consume(CustomEvent event, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
try {
log.info("Received event on topic {}", topic);
handler.handleEvent(event);
} catch (Exception e) {
log.error("Error occurred while processing event", e);
throw e;
}
}
@DltHandler
public void listenOnDlt(@Payload CustomEvent event) {
log.error("Received event on dlt.");
handler.handleEventFromDlt(event);
}
}
如果您注意上面的代码片段，参数@RetryableTopic中包含includes。这告诉消费者只在方法抛出CustomRetryableException时进行重试。您可以添加任意数量的异常类型。还有一个exclude参数，但一次只能使用其中一个。在将事件发布到死信队列之前，事件处理最多应重试指定的次数。
设置测试基础设施
为了编写集成测试，您需要确保拥有一个正常运行的Kafka代理（最好是嵌入式的）和一个完全运行的发布者。让我们设置我们的基础设施：
@EnableKafka
@SpringBootTest
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
@EmbeddedKafka(partitions = 1,
brokerProperties = {"listeners=" + "${kafka.broker.listeners}",
"port=" + "${kafka.broker.port}"},
controlledShutdown = true,
topics = {"test", "test-retry-0", "test-retry-1", "test-dlt"}
)
@ActiveProfiles("test")
class DocumentEventConsumerIntegrationTest {
@Autowired
private KafkaTemplate<String, CustomEvent> testKafkaTemplate;
// tests
}
配置从application-test.yml文件中导入。当使用嵌入式Kafka代理时，重要的是要提及要创建的主题。它们不会自动创建。在这种情况下，我们将创建四个主题，分别是：
"test", "test-retry-0", "test-retry-1", "test-dlt"
我们将最大重试次数设置为三次。每个主题对应于每次重试尝试。因此，如果三次重试都耗尽，事件应该被转发到DLT（死信队列）。
测试用例
如果在第一次尝试中成功消费，就不应该进行重试。可以通过方法只被调用一次来测试这一点。还可以添加对日志语句的进一步测试。
@Test
void test_should_not_retry_if_consumption_is_successful() throws ExecutionException, InterruptedException {
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));
}
如果引发了不可重试的异常，就不应该进行重试。在这种情况下，方法CustomEventHandler#handleEvent应该只被调用一次。
@Test void test_should_not_retry_if_non_retryable_exception_raised() throws ExecutionException, InterruptedException { CustomEvent event = new CustomEvent("Hello"); // GIVEN doThrow(CustomNonRetryableException.class).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)); }
如果抛出了RetryableException，则应该按照配置的最大重试次数进行重试，当重试次数耗尽时，事件应该被发布到死信主题。在这种情况下，方法CustomEventHandler#handleEvent应该被调用三次（maxRetries次），而方法CustomEventHandler#handleEventFromDlt应该只被调用一次。
@Test
void test_should_not_retry_if_non_retryable_exception_raised() throws ExecutionException, InterruptedException {
CustomEvent event = new CustomEvent("Hello");
// GIVEN
doThrow(CustomNonRetryableException.class).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));
}
在验证阶段添加了相当长的超时时间，以便在测试完成之前考虑指数退避延迟。这是很重要的，如果没有正确设置，可能会导致断言失败。应该重试直到RetryableException被解决，并且如果引发了不可重试的异常或者最终成功消费，就不应该继续重试。测试已经设置为首先抛出RetryableException，然后再抛出NonRetryableException，以便进行一次重试。
@Test
void test_should_retry_until_retryable_exception_is_resolved_by_non_retryable_exception() throws ExecutionException,
InterruptedException {
CustomEvent event = new CustomEvent("Hello");
// GIVEN
doThrow(CustomRetryableException.class).doThrow(CustomNonRetryableException.class).when(customEventHandler).handleEvent(any(CustomEvent.class));
// WHEN
testKafkaTemplate.send("test", event).get();
// THEN
verify(customEventHandler, timeout(10000).times(2)).handleEvent(any(CustomEvent.class));
verify(customEventHandler, timeout(2000).times(0)).handleEventFromDlt(any(CustomEvent.class));
}ndleEventFromDlt(any(CustomEvent.class)); }
@Test
void test_should_retry_until_retryable_exception_is_resolved_by_successful_consumption() throws ExecutionException,
InterruptedException {
CustomEvent event = new CustomEvent("Hello");
// GIVEN
doThrow(CustomRetryableException.class).doNothing().when(customEventHandler).handleEvent(any(CustomEvent.class));
// WHEN
testKafkaTemplate.send("test", event).get();
// THEN
verify(customEventHandler, timeout(10000).times(2)).handleEvent(any(CustomEvent.class));
verify(customEventHandler, timeout(2000).times(0)).handleEventFromDlt(any(CustomEvent.class));
}
结论
因此，您可以看到集成测试是一种混合和匹配的策略，超时时间，延迟和验证，以确保您的Kafka事件驱动架构的重试机制是可靠的。
作者： Mukut Bhattacharjee
更多技术干货尽在wx“云原生数据库”
