I have the below configuration. When the message is invalid I want to send an email and for errors I want to save it in database. How can I handle this in errorHandler() ?
@Configuration
@EnableKafka
public class KafkaConsumerConfig implements KafkaListenerConfigurer{
@Bean
ErrorHandler errorHandler() {
return new SeekToCurrentErrorHandler((rec, ex) ->
{ dbService.saveErrorMsg(rec); }
,new FixedBackOff(5000, 3)) ;
}
@Override
public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
registrar.setValidator(this.validator);
}
@KafkaListener(topics = "mytopic", concurrency = "3", groupId = "mytopic-1-groupid")
public void consumeFromTopic1(@Payload @Valid ValidatedClass val, ConsumerRecordMetadata meta) throws Exception
{
dbservice.callDB(val,"t");
}