Kafka Producer is not retrying in Spring Boot Application

Question

I am trying to test re-try Exception property of Kafka Producer in case of transient error. But it is not retrying. I read somewhere that the ProducerConfig.RETRIES_CONFIG configuration in Kafka works regardless of whether you catch exceptions or not. So I am not sure why it is not retrying?

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    //Records will be failed if they can’t be delivered in delivery.timeout.ms
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG,"120000");
    // Only retry after one second.
    props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
    // Set the number of retries 
    props.put(ProducerConfig.RETRIES_CONFIG, "3");
    return props;
}


public void pushMessageToKafka(Product product) {
    
    try {
        String payload= objectMapper.writeValueAsString(product);
        
        if(!checkIfTopicExists("product-unknown-kafka-topic")) {
            throw new TimeoutException("Failed to send message to Kafka topic" );
        }
        
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(productTopic,
                null, product.getProductName(), payload);

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);
                future.addCallback(new KafkaSendCallback<String, String>() {

            @Override
            public void onFailure(KafkaProducerException ex) {
                log.error("Failed to send product to kafka for productName:{}", product.getProductName());

            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                String topic = result.getProducerRecord().topic();
                log.info("Sucsessfully send product details to kafka for productName:{} topic:{}", product.getProductName(),topic);
            }

        });

    } catch (Exception e) {
        log.error("Exception occured while trying to send product to kafka for productName:{} error:{}",
                product.getProductName(), e.getMessage());
        
        
    }

}