Spring Kafka Manual Ack and retry happens after exception : unexpected behaviour

Viewed 30

I have a Kafka consumer annotated with @KakaListner, with in consumer i have requirement like to acknowledge kafka message first and then execute rest api. So while executing rest api i am getting exception which is expected behaviour and consumer reties happening even i ack. same message before rest call.

     @KafkaListener(topics = "${spring.kafka.topic.name}", groupId = "${consumer.topicGroupId}")
        public void listenEvent(ConsumerRecord<String, Event> consumerRecord, Acknowledgment acknowledgment) throws IOException {

             acknowledgment.acknowledge();
             patchService.patch(arguments...);
}

PatchService code :

    public class PatchService {

         public String patch(arguments....) {

       try {
            ResponseEntity<String> response = restTemplate.exchange(uri, HttpMethod.PATCH, request, String.class);
            return response.getBody();
        } catch (HttpClientErrorException ex) {
            log.error("Error updating API having error response {}", ex.getResponseBodyAsString());
            throw new APIException(ex.getStatusCode(), ex.getResponseBodyAsString());
        }

     }
}

ConsumerConfiguration Code :

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() throws FileNotFoundException {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<String, String>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

With this property set to false in consumerFactory() :

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

What i am doing wrong here ? i don't want retries when exception comes as part of rest api i just want to log that message which i am doing.

1 Answers

Catch the exception and don't throw it.

Kafka maintains two pointers for each group/partition - the current position and the committed offset.

The default error handler resets the current position so the failed record will be redelivered, regardless of the committed offset.

Related