Kafka 对通过 topicPattern 订阅多个主题的消费者进行非阻塞重试答案

【问题标题】：Kafka non blocking retry on consumer subscribed to multiple topics by topicPatternKafka 对通过 topicPattern 订阅多个主题的消费者进行非阻塞重试
【发布时间】：2022-01-11 15:20:13
【问题描述】：

我正在尝试对通过 topicPattern 订阅多个主题的消费者实现 kafka 非阻塞重试。当我尝试创建重试 RetryTopicConfiguration 时出现此错误。

No topics were provided for RetryTopicConfiguration for method dynamicConsumer in class KafkaConsumerClass

这里的dynamicConsumer是带有topicPattern注解的函数。

示例配置

    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);

    props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StickyAssignor.class.getName() );

重试配置

 .newInstance()
 .listenerFactory(kafkaListenerContainerFactoryRetry)
 .exponentialBackoff(1000, 2, 5000)
 .maxAttempts(3)
 .includeTopics(Arrays.asList("postfixtest"))
 .create(template);

卡夫卡消费者

@KafkaListener( topicPattern = ".*postfixtest",groupId = "edsdefd",containerFactory = "kafkaListenerContainerFactory4")
public void dynamicConsumer(ConsumerRecord<String,String> message) throws Exception {
    System.out.println("got a message from  topic --> "+message.topic()+" - message is ->  "+message.value()+" partition" +message.partition());
}

其他假设：除非父主题在应用程序上下文中有其消费者，否则重试 bean 不会从主题中消费。

【问题讨论】：

请显示您如何配置RetryTopicConfiguration 以及如何使用它的代码。
我只使用上面的重试配置，我已经在构建器中添加了我喜欢的主题名称。重试配置会自动为我创建重试和 dlt 主题。
你需要解释一下你所说的“动态消费者”是什么意思；重试主题机制依赖于底层 Spring Framework 功能。提供更多信息。
动态消费者只是一个使用 topicPattern 而不是单个主题的消费者。我应该用正则表达式编写多个主题消费者，我的错。
@GaryRussell 还有一个问题。您如何建议在运行时制作非阻塞重试 bean，您认为它是否具有可扩展性。使用 topicPattern 可以通过增加消费者来扩展。我不认为自动创建的非阻塞重试主题是可扩展的，至少对于在运行时创建的主题而言。

标签： spring-boot apache-kafka spring-kafka

【解决方案1】：

您不能对topicPattern 使用非阻塞重试，因为框架需要提前知道主题名称，以便它可以设置基础架构并重试主题。

您可以使用原型 bean 在运行时创建新的侦听器；它也应该适用于非阻塞重试。

见Can i add topics to my @kafkalistener at runtime

【讨论】：

这适用于大多数情况，但我需要像可扩展解决方案这样的主题模式，通过手动创建侦听器，我必须手动将运行时创建的大量主题分发到每个服务器上。这是不可能的。
目前我正在为每个客户端创建主题，在运行时当客户端载入时，我需要随着负载的增加扩展服务器。我认为手动创建侦听器不适用于我的情况，然后我需要手动创建非阻塞重试模式。