Kafka Consumer Invalid Payload Error Handler

Viewed 295

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");
    }
1 Answers

I presume your emai code is in dbService.saveErrorMsg.

Spring Boot should automatically detect the ErrorHandler @Bean and wire it into the container factory.

See Boot's KafkaAnnotationDrivenConfiguration class and ConcurrentKafkaListenerContainerFactoryConfigurer.

Related