用于设置 SeekToCurrentBatchErrorHandler 的 Spring Kafka 属性答案

【问题标题】：Spring Kafka Property for setting SeekToCurrentBatchErrorHandler用于设置 SeekToCurrentBatchErrorHandler 的 Spring Kafka 属性
【发布时间】：2019-04-09 03:29:00
【问题描述】：

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

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

与 SeekToCurrentBatchErrorHandler ？提前致谢。

【问题讨论】：

标签： spring apache-kafka spring-kafka

【解决方案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;
}

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

【讨论】：

谢谢加里 - 这真的很有帮助。以后有计划添加吗？
我添加了一个new feature issue 来启动。它不是一个属性，但如果在应用程序上下文中恰好找到一个属性，它可以在@Bean 中自动装配（就像当前对非批处理ErrorHandler 所做的那样）。
这也有帮助。 :) 我注意到自动装配 ErrorHandler Bean 并不完全适用于批处理。以为我做错了什么。 :)
不确定我是否应该将问题放在这里 - 是否有原因未实现“spring.kafka.listener.ack-on-error=false/true”功能？
你真的不应该；他们不喜欢这里的扩展评论。只是有太多属性，只选择了几个，您总是可以在我的回答中使用该技术进行“高级”配置。您可以打开另一个问题来请求它反对启动，但他们可能不会在 2.2 之前添加它（它使用 S-K 2.3，我们已将默认设置切换为 false 那里）。