DLT消息消耗

Question

我们使用的是Spring kafka 2.7非阻塞重试机制。在Spring Kafka重试机制中，Kafka listenser使用来自main topic、Retry topic和DLT topic的消息，我们希望侦听器仅使用来自main和Retry topic的消息。

有没有简单的方法来进行设置？

因为我们不希望同一个消费者处理DLT消息。DLT还将被另一个进程使用，以发送请求通知。

// our configuration

 @Bean
  public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory<String, Object> factory
        = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
  }


@Bean
  public RetryTopicConfiguration retryTopicConfiguration(KafkaTemplate<String, Object> template) {

    List<Class<? extends Throwable>> throwableList = Arrays.asList(IllegalArgumentException.class,
        IllegalAccessException.class);

    return RetryTopicConfigurationBuilder
        .newInstance()
        .dltHandlerMethod(XYZ.class, "xyz")
        .exponentialBackoff(delayMs, backoffMultiplier, maxIntervalInMs)
        .maxAttempts(retryAttempt)
        .notRetryOn(throwableList)
        .doNotAutoCreateRetryTopics()
        .listenerFactory(kafkaListenerContainerFactory())
        .setTopicSuffixingStrategy(TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
        .create(template);
  }

Answer 1

只需删除。dltHandlerMethod（XYZ.class，“XYZ”）。

编辑

这仍然会创建一个默认的DLT处理程序，它只会记录记录。

您可以在“其他”使用者中使用其他组，也可以手动启动容器（DLT容器除外）。

在KafkaListener上设置自动启动；然后添加此。。。

@Bean
ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {
    return args -> {
        registry.getListenerContainerIds().forEach(id -> {
            if (!id.endsWith("-dlt")) {
                registry.getListenerContainer(id).start();
            }
        });
    };
}