I am trying to convert single message processing to batch processing with manual acknowledgement . without batch the code works as expected. But somehow , it is not working with batch.
Here is my properties and code
spring.cloud.stream.kafka.bindings.inputChannel-in-0.batch-mode=true
@Bean
public Consumer<List<Message<?>>> inputChannel() {
return batch -> {
for ( message : batch) {
log.info("Message Processing Starts :: ");
Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
// process message and
// acknowledge manually
if(acknowledgement != null ){
ackwoledge...
}
};
}
But I am getting classCastException Caused by: java.lang.ClassCastException: class org.springframework.messaging.support.GenericMessage cannot be cast to class java.util.List (org.springframework.messaging.support.GenericMessage is in unnamed module of loader 'app'; java.util.List is in module java.base of loader 'bootstrap')
Also , how can I add retry feature here considering batch processing has maxAttempt = 1