发生错误时如何关闭KafkaListener

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    DefaultKafkaListenerContainerFactory<String, String> factory = new DefaultKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setErrorHandler(new SeekToCurrentErrorHandler());
    return factory;
}

@KafkaListener(topics = "topicName", containerFactory = "kafkaListenerContainerFactory")
public void listen(ConsumerRecord<String, String> record) {
    // 处理消息
}

@Component
public class CustomKafkaListenerErrorHandler implements ErrorHandler {
    @Override
    public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
        // 处理错误
    }
}

@KafkaListener(topics = "topicName", containerFactory = "kafkaListenerContainerFactory", errorHandler = "customKafkaListenerErrorHandler")
public void listen(ConsumerRecord<String, String> record) {
    // 处理消息
}