spring - 用于设置 SeekToCurrentBatchErrorHandler 的 Spring Kafka 属性

Question

是否有spring.kafka属性错误处理批次

spring.kafka.listener.type=BATCH和 spring.kafka.listener.ack-mode=BATCH

与SeekToCurrentBatchErrorHandler？提前致谢。

标签： springapache-kafkaspring-kafka

Answer 1

你不能用一个属性来设置它，但是你可以像这样覆盖 Boot 的自动配置容器工厂：

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory) {

    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);

    factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());

    return factory;
}

它将获取所有启动属性，然后您可以根据需要进一步配置工厂。