在 Spring Kafka 中结合阻塞和非阻塞重试

Question

我正在尝试使用单个主题固定后退来实现非阻塞重试。

感谢文档https://docs.spring.io/spring-kafka/reference/html/#single-topic-fixed-delay-retries，我能够做到这一点。

现在我还需要对主要主题执行一些阻塞/本地重试。我一直在尝试使用DefaultErrorHandler 来实现这一点，如下所示：

@Bean
public DefaultErrorHandler retryErrorHandler() {
        return new DefaultErrorHandler(new FixedBackOff(2000, 3));
}

这似乎不适用于RetryableTopic。

我也尝试了以下方法retry-topic-combine-blockinghttps://docs.spring.io/spring-kafka/reference/html/#retry-topic-combine-blocking使用ListenerContainerFactoryConfigurer 但我在这里面临的问题是创建 bean KafkaConsumerBackoffManager、DeadLetterPublishingRecovererFactory，尤其是 KafkaConsumerBackoffManager。

我需要知道这是否是使用 spring kafka 框架实现此目的的另一种方法，或者有没有办法在 bean 之上构建？

Answer 1

我们目前正在努力改进非阻塞重试组件的配置。

现在，正如 here 所记录的那样，您应该注入这些 bean，例如：

@Bean(name = RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME)
public ListenerContainerFactoryConfigurer lcfc(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
                                               DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory,
                                               @Qualifier(RetryTopicInternalBeanNames
                                                       .INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
    ListenerContainerFactoryConfigurer lcfc = new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager, deadLetterPublishingRecovererFactory, clock);
    lcfc.setBlockingRetryableExceptions(MyBlockingRetryException.class, MyOtherBlockingRetryException.class);
    lcfc.setBlockingRetriesBackOff(new FixedBackOff(500, 5)); // Optional
    return lcfc;
}}

此外，还有一个已知问题，如果您尝试在处理具有可重试主题的第一个 @KafkaListener bean 之前注入 bean，则该功能的组件的 bean 将不会出现在上下文中并且会引发错误。

这发生在你身上吗？

我们目前正在为此工作fix，但如果这是您的问题，我们应该能够解决这个问题。

编辑：由于问题是组件尚未实例化，因此最有保证的解决方法是自己提供组件。

这是有关如何执行此操作的示例。当然，如果您需要任何进一步的定制，请相应地进行调整。

    @Configuration
    public static class SO71705876Configuration {

        @Bean(name = RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME)
        public ListenerContainerFactoryConfigurer lcfc(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
                DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory) {
            ListenerContainerFactoryConfigurer lcfc = new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager, deadLetterPublishingRecovererFactory, Clock.systemUTC());
            lcfc.setBlockingRetryableExceptions(IllegalArgumentException.class, IllegalStateException.class);
            lcfc.setBlockingRetriesBackOff(new FixedBackOff(500, 5)); // Optional
            return lcfc;
        }

        @Bean(name = RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER)
        public KafkaConsumerBackoffManager backOffManager(ApplicationContext context) {
            PartitionPausingBackOffManagerFactory managerFactory =
                    new PartitionPausingBackOffManagerFactory();
            managerFactory.setApplicationContext(context);
            return managerFactory.create();
        }

        @Bean(name = RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_FACTORY_BEAN_NAME)
        public DeadLetterPublishingRecovererFactory dlprFactory(DestinationTopicResolver resolver) {
            return new DeadLetterPublishingRecovererFactory(resolver);
        }

        @Bean(name = RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME)
        public DestinationTopicResolver destinationTopicResolver(ApplicationContext context) {
            return new DefaultDestinationTopicResolver(Clock.systemUTC(), context);
        }

在下一个版本中，这应该不再是问题了。请让我知道这是否适合您，或者是否需要对此解决方法进行任何进一步的调整。

谢谢。