Как я могу обрабатывать IOException, когда Kafka не работает?

Я пытаюсь опубликовать сообщение, и Apache Kafka не работает. Как мне обращаться с такой чрезвычайной ситуацией?

Метод KafkaProducer:: send() не генерирует никаких исключений, которые могут быть обработаны. Провайдер проглатывает их и записывает ошибки вместо этого, поэтому я получаю сообщения с такими сообщениями, и все зависает, пока Кафка не подключится снова.

2014-03-31 09:38:23.752 ERROR o.a.kafka.common.network.Selector - Error in I/O: 
java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.7.0_51]
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739) ~[na:1.7.0_51]
    at org.apache.kafka.common.network.Selector.poll(Selector.java:205) ~[kafka-clients-0.8.1.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:212) [kafka-clients-0.8.1.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:150) [kafka-clients-0.8.1.jar:na]
    at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]

Ответ 1

Вызовите get() в будущем, возвращенном Producer.send(...), или если вы не хотите блокировать свои код, передайте callback.

try { producer.send(new ProducerRecord("mytopic", key, value)) .get(); // block until acknowledged } catch(Exception e) { // handle message wasn't acknowledged }

Ответ 2

Я думаю, что одна вещь, которую вы могли бы сделать, это использовать try/catch для регистрации всех Throwable в вашем коде производителя.