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;
}
with:
Kafka version : 1.0.2, spring-kafka:2.1.10.RELEASE & springBootVersion = '2.0.5.RELEASE'
getting the following exception:
Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is java.lang.ClassCastException: org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler cannot be cast to org.springframework.kafka.listener.ErrorHandler
Appreciate your help.
To create batch listener containers, add factory.setBatchListener(true);
Collected from the Internet
Please contact [email protected] to delete if infringement.
Comments