Spring云流kafka活页夹重试

Question

我有一个使用kafka活页夹的spring cloud stream应用程序，它可以消费和发送消息。在应用程序中，我使用重试策略配置自定义错误处理程序，并将不可重试的异常添加到处理程序中。配置示例：

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(
    SeekToCurrentErrorHandler customErrorHandler
) {
    return (((container, destinationName, group) -> {
        container.setErrorHandler(customErrorHandler);
    }));
}

@Bean
public SeekToCurrentErrorHandler customErrorHandler() {
    var errorHandler = new SeekToCurrentErrorHandler(
        (consumerRecord, e) -> log.error("Got exception skip record record: {}", consumerRecord, e),
        new FixedBackOff(1000L, 10)
    );
    errorHandler.addNotRetryableException(App.MyCustomException.class);
    return errorHandler;
}

但是我看到，如果异常抛出，比应用程序重试处理消息3次。预期行为-如果App. MyCustomException.class抛出，将不会重复消费消息。如何为Spring云流kafka绑定应用程序配置重试策略？

这里的代码示例：github

运行复制问题的测试。

Answer 1

您提供的自定义是针对容器级错误处理程序的。Binder有不同的重试机制。您可以将以下内容添加到您的配置中，以确保发生异常时不会重试记录。

spring.cloud.stream:
    bindings:
      processor-in-0:
        ...
        consumer:
          retryableExceptions:
            ru.vichukano.kafka.binder.retry.App.MyCustomException: false

当我尝试这样做时，我没有看到消息被重新传递。

以下是对此的一些解释。