DLT 消息消费

Question

我们使用的是 Spring kafka 2.7 非阻塞重试机制。在 Spring Kafka 重试机制期间，kafka 监听器消费来自 main topic、retry topic 和 DLT 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上设置autoStartup="false"；然后添加这个...

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