Batch Processing with Manual Acknowledgment using Spring Cloud Stream Kafka

Viewed 383

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

0 Answers
Related