Опция Spring Kafka Batch: ClassCastException с использованием SeekToCurrentBatchErrorHandler

Question

Trying to configure the Factory as below:
private ConcurrentKafkaListenerContainerFactory<Integer, String> getKafkaContainerFactory()
        throws IOException {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(this.getConsumerFactory());//DefaultKafkaConsumerFactory
    factory.setAutoStartup(true);
    ContainerProperties props = factory.getContainerProperties();
    props.setAckMode(AbstractMessageListenerContainer.AckMode.BATCH);
    props.setAckOnError(false);
    props.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
    return factory;
}

с:

Kafka version : 1.0.2, spring-kafka:2.1.10.RELEASE & springBootVersion = '2.0.5.RELEASE'

получает следующее исключение:

Исключительная ситуация при инициализации контекста - отмена попытки обновления: org.springframework.context.ApplicationContextException: не удалось запустить bean-компонент 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; вложенным исключением является java.lang.ClassCastException: org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler не может быть приведен к org.springframework.kafka.listener.ErrorHandler

Ценю вашу помощь.

Answer 1

Для создания контейнеров пакетного прослушивателя добавьте factory.setBatchListener(true);