Kafka Producer : Handle Exception in Async Send with Callback

Question

I need to catch the exceptions in case of Async send to Kafka. The Kafka producer Api comes with a fuction send(ProducerRecord record, Callback callback). But when I tested this against following two scenarios :

Kafka Broker Down
Topic not pre created The callbacks are not getting called. Rather I am getting warning in the code for unsuccessful send (as shown below).

Questions :

So are the callbacks called only for specific exceptions ?
When does Kafka Client try to connect to Kafka broker while async send : on every batch send or periodically ?

Kafka Warning Image

Note : I am also using linger.ms setting of 25 sec to batch send my records.


public class ProducerDemo {

    static KafkaProducer<String, String> producer;

    public static void main(String[] args) throws IOException {

         final Logger logger = LoggerFactory.getLogger(ProducerDemo.class);
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "30000");

        producer = new KafkaProducer<String, String>(properties);
        String topic = "first_topic";

        for (int i = 0; i < 5; i++) {
            String value = "hello world " + Integer.toString(i);
            String key = "id_" + Integer.toString(i);

            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, value);

              producer.send(record, new Callback() {
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        //execute everytime a record is successfully sent or exception is thrown
                        if(e == null){
                           // No Exception
                        }else{
                            //Exception Handling
                        }
                    }
                });
        }
        producer.close();
    }

Raj Raj · Accepted Answer · 2020-07-07T21:42:22

For the first question, here is the answer. As per the apache kafka documentation, you can capture below exceptions using onCompletion method when you are implementing Callback interface

https://kafka.apache.org/25/javadoc/org/apache/kafka/clients/producer/Callback.html

For the second question, the combination of below properties control when to send the records and as far as i understand, it's same for synchronous or asynchronous call.

linger.ms max.block.ms

https://kafka.apache.org/documentation/#linger.ms

Kafka Producer : Handle Exception in Async Send with Callback

3 Answers