使用 EmbeddedKafka 进行测试 - 证明重试次数

Question

我有一个使用@EmbeddedKafka 编写的测试。 我使用 SeekToCurrentErrorHandler 配置了错误处理。

我有一个测试，我将一条消息推送到 Kafka，我可以看到它由 DeadLetterPublishingRecover 正确处理，它发布到 DLT。

我想在我的测试中添加一个进一步的断言，以证明 SeekToCurrentErrorHandler 重试了 X 次，但发生 MyCustomException 时只重试了 1 次。

我已经配置了错误处理程序

errorHandler.addNotRetryableExceptions(MyCustomException.class)

我正在努力如何在测试中获取重试信息。如果 DeadLetterPublishingRecoverer 添加一个标头即 kafka_deliveryAttempt 标头会很好。

我也试过这样做：

1. 在测试中创建 RetryTemplate

@Configuration
public class TestConfiguration {

    @Bean
    @Primary
    public RetryTemplate retryTemplate() {
        return new RetryTemplateBuilder().maxAttempts(3)
                                         .fixedBackoff(500)
                                         .build();
    }
}

1. 在重试模板上设置监听器。

@Import(TestConfiguration.class)
class MyTest {

@Autowired
private RetryTemplate retryTemplate;

private int retryCount = 0;

  @BeforeEach
  void setup () {
   retryTemplate.registerListener(new RetryListenerSupport() {
            @Override
            public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                retryCount = context.getRetryCount();
            }
        });
  }
}


// I then execute the test and I can see in the logs it had 3 goes at processing the message. However when I assert for the count to equal 3 it's always 0

Answer 1

SeekToCurrentErrorHandler 不使用RetryTemplate；它有自己的重试机制。

从 2.7 版本开始，您可以在错误处理程序中添加一个或多个 RetryListeners：

    /**
     * Set one or more {@link RetryListener} to receive notifications of retries and
     * recovery.
     * @param listeners the listeners.
     * @since 2.7
     */
    public void setRetryListeners(RetryListener... listeners) {
        this.failureTracker.setRetryListeners(listeners);
    }

/**
 * A listener for retry activity.
 *
 * @author Gary Russell
 * @since 2.7
 *
 */
@FunctionalInterface
public interface RetryListener {

    /**
     * Called after a delivery failed for a record.
     * @param record the failed record.
     * @param ex the exception.
     * @param deliveryAttempt the delivery attempt.
     */
    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);

    /**
     * Called after a failing record was successfully recovered.
     * @param record the record.
     * @param ex the exception.
     */
    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
    }

    /**
     * Called after a recovery attempt failed.
     * @param record the record.
     * @param original the original exception causing the recovery attempt.
     * @param failure the exception thrown by the recoverer.
     */
    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
    }

}