How to handle Kafka connection refused Exception

[Xiaojing Gong]

Hi,

I want to save the data to Cassandra when Kafka can't publishing the data. So I wrote the following code , using Call back to handle the exceptions. However, call back doesn't have connection Exception. My question is where can I handle Kafka Connection Exception like connection get refused. 

public void publishToKafka(String topicName,String msg) throws Exception {

    producerRecord = new ProducerRecord(topicName,msg);

    final long startTime = System.currentTimeMillis();
    producer.send(producerRecord, new Callback() {
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if(e != null) {

                kafkaFailurePublishServiceImpl.saveToCassandra(producerRecord.topic(),producerRecord.value().toString());

            }
        }

    });
    final long stopTime = System.currentTimeMillis();

    System.out.println("Time Consumed = "+ (stopTime-startTime));
}

[gerard...@dizzit.com]

You can add a try/catch around the 'producer.send(producerRecord, callBack) and catch ge general runtime KafkaException, not very pretty, but it works. I ran into a similar problem when a had a serialisation problem. The callback is only used if the connection and serialisation work, and the bits are actually send.

[Xiaojing Gong]

Hi Thank you for replying me.

I tried , however the error not catch, it just throw out the connection refused error message :( , it not go that catch block. 

try {

    producer.send(producerRecord, new Callback() {
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {

                kafkaFailurePublishServiceImpl.saveToCassandra(producerRecord.topic(), producerRecord.value().toString());

            } else {
                System.out.print("testing");

            }
        }

    });
    final long stopTime = System.currentTimeMillis();

    System.out.println("Time Consumed = " + (stopTime - startTime));

} catch (KafkaException e) {
    kafkaFailurePublishServiceImpl.saveToCassandra(producerRecord.topic(), producerRecord.value().toString());
}

[2016-04-14 11:23:53.420] boot - 2767  WARN [kafka-producer-network-thread | ideal-kafka-adapter] --- Selector: Error in I/O with serena.cof.ds.capitalone.com/10.219.18.189

java.net.ConnectException: Connection refused

at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)

at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)

at org.apache.kafka.common.network.Selector.poll(Selector.java:238)

at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:192)

at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:191)

at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:122)

at java.lang.Thread.run(Thread.java:745) 

[Xiaojing Gong]

[2016-04-14 11:23:53.420] boot - 2767  WARN [kafka-producer-network-thread | ideal-kafka-adapter] --- Selector: Error in I/O with *****

java.net.ConnectException: Connection refused

at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)

at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)

at org.apache.kafka.common.network.Selector.poll(Selector.java:238)

at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:192)

at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:191)

at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:122)

at java.lang.Thread.run(Thread.java:745)

[gerard...@dizzit.com]

You can add it to the catch with (KafkaException | ConnectException e)  But not sure is this is supposed to happen.

[Xiaojing Gong]

:(

It is not allowed me to add connectionException here 

[HunJae Lee]

I am facing the same situation...
Is there anyway to catch the ConnectionException?

[Prasad Dls]

I am also facing same issue, not able get exception when my broker is down or not able to reach

[iba...@bbtv.com]

Same issue over here... isn't there any solution for it?

[ke...@databerries.com]

Same issue for me. Any update?