使用启用批处理模式的Spring Cloud Stream在Kafka中实现DLQ

Question

我正在尝试使用启用批处理模式的spring cloud stream实现DLQ

    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(BatchErrorHandler handler) {
        return ((container, destinationName, group) -> {
              if(dlqEnabledTopic.contains(destinationName))
                                    container.setBatchErrorHandler(handler);});
    }

    @Bean
    public BatchErrorHandler batchErrorHandler(KafkaOperations<String, byte[]> kafkaOperations) {
        CustomDeadLetterPublishingRecoverer recoverer = new CustomDeadLetterPublishingRecoverer(kafkaOperations,
                (cr, e) -> new TopicPartition(cr.topic()+"_dlq", cr.partition()));
        return new RecoveringBatchErrorHandler(recoverer, new FixedBackOff(1000, 1));
    }

但有一些疑问：

>

如何使用属性配置键/值序列化程序-我的消息是String类型，但KafkaOperations使用的是ByteArraySerializer

在批处理中，有多个消息，但如果第一条消息失败，它会转到DLQ，但看不到下一条消息的处理。

要求-如果批处理失败，我只需要将该消息发送到DLQ，并且应该再次处理消息的其余部分。

Answer 1

>

spring.kafka.producer.*属性-但是，DLT发布应该使用与主流应用相同的序列化程序。ByteArraySeriezer通常是正确的。

恢复的批处理错误处理程序将对未处理的记录执行查找，它们将被返回。调试日志记录应该可以帮助您找出问题所在。如果您无法弄清楚，请提供一个显示您正在看到的行为的MCRE。

不活页夹不支持批量模式的DLQ；配置错误处理程序是正确的方法。