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.