在批处理模式下如何处理反序列化异常

Question

如何在批处理模式的情况下处理反序列化异常？
我正在使用Spring boot-2.3.8版本的Spring kafka。

尝试过此选项：

@Bean
  public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
      ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
      ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory); 
    factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
    }

但它抛出了一个异常：由java引起。lang.IllegalStateException:错误处理程序必须是ErrorHandler，而不是org。springframework。Kafka。听众。请参阅OcuCurrentBatchErrorHandler

以下是我的应用程序属性：

spring.kafka.consumer.bootstrap-servers = localhost:9093
spring.kafka.consumer.enable-auto-commit = false
spring.kafka.consumer.auto-offset-reset = earliest
spring.kafka.consumer.max-poll-records = 10

spring.kafka.consumer.key-deserializer= org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer= org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.listener.type =  batch 
spring.kafka.ack.discard =  true
spring.kafka.listener.ack-mode = MANUAL
spring.kafka.listener.concurrency = 1
idle-between-polls  = 120000

Answer 1

这意味着侦听器类型设置不正确。

我刚刚将您的代码和配置复制到一个新应用程序，它按预期工作。