如何配置DeadLetterPublisherRecoverer以在Spring Cloud Stream批处理模式下向DLQ发送错误消息

Question

我已经使用Spring Cloud Stream和Spring Cloud函数创建了一个Kafka消费者，用于以批处理模式从Kafka主题消费消息。现在，我想将错误批处理发送到死信队列以进一步调试错误。

我正在使用Spring重试处理消费者方法中的重试。但对于不可重试的异常，我希望将整个批次发送到DLQ。

这是我的消费者的样子：

@Bean
public Consumer<List<GenericRecord>> consume() {
    return (message) -> {
        processMessage(message);  
    }
}

错误处理配置如下所示:

@Autowired
private DefaultErrorHandler errorHandler;

ListenerContainerCustomizer<AbstractMessageListenerContainer> c = new ListenerContainerCustomizer<AbstractMessageListenerContainer>() {
      @Override
      public void configure(AbstractMessageListenerContainer container, String destinationName, String group) {
        container.setCommonErrorHandler(errorHandler);
      }
}

使用DeadRec总计PublishinRecoverer启用错误处理程序以将失败的消息发送到DLQ：

@Bean
public DefaultErrorHandler errorHandler(KafkaOperations<String, Details> template) {
    return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template,
        (cr, e) -> new TopicPartition("error.topic.name", 0)),
        new FixedBackOff(0, 0));
}

但这并没有向error.topic发送任何消息，从错误日志中我可以看到它正在尝试连接到localhost：9092，而不是我在spring.cloud.stream.kafka.binder.brokers中提到的代理。

如何配置 DLQ 提供程序以从应用程序属性读取 Kafka 元数据？

还有没有办法配置供应商函数来创建DLQ提供程序？

Answer 1

您可能正在使用 Boot 自动配置的 KafkaTemplate。

请使用< code > spring . Kafka . bootstrap-servers ，如果没有< code > spring . cloud . stream . Kafka . binder . brokers ，绑定器将使用它；这样，绑定器和模板都将连接到同一个代理。

您必须抛出BatchListnerFailedExc的来指示批处理中的哪一条记录失败。