Kafka 生产者：如何处理“java.net.ConnectException：连接被拒绝”答案

【问题标题】：Kafka producer: how to handle "java.net.ConnectException: Connection refused"Kafka 生产者：如何处理“java.net.ConnectException：连接被拒绝”
【发布时间】：2017-04-13 11:50:41
【问题描述】：

我使用 Kafka 0.10.1.0。

这是我的制作人：

val props: Properties = ...
val producer = new KafkaProducer[String, AnyRef](props)
val callback = new Callback {
   override def onCompletion(md: RecordMetadata, e: Exception): Unit = ...
}
producer.send(new ProducerRecord[String, AnyRef]("topic", "hello"), callback)

但是上面的回调不能处理java.net.ConnectException: Connection refused，以防kafka-server宕机。

UPD

ConnectionException 在另一个线程中引发（进入Sender 类，该类用于KafkaProducer）。因此我们不能使用try {} catch。另外我不需要重试机制，我需要一种处理这种情况的方法（例如，如果 Kafka 宕机并且生产者无法发送消息，那么我将使用其他队列 API）。

有没有办法处理这个异常？

【问题讨论】：

标签： java scala apache-kafka kafka-producer-api

【解决方案1】：

您有几个选择。 Scala 提供了一种捕获异常的方法，其形式如下：

   try { 
   // ... 
   } 
   catch {
     case ioe: IOException => ... // more specific cases first !
     case e: Exception => ...
   }

所以最简单的方法是：

  try { 
     producer.send(new ProducerRecord[String, AnyRef]("topic", "hello"), callback)
   } 
   catch {
     case ce: ConnectionException => // handle exception
   }

更复杂但更强大的是重试机制：

What's the Scala way to implement a retry-able call like this one?

还要注意，Kafka Producer 内置了重试机制，这也可能有帮助：

设置大于零的值将导致客户端重新发送发送失败的任何记录潜在的瞬态错误。请注意，此重试与如果客户端在收到错误后重新发送记录。允许重试而不将 max.in.flight.requests.per.connection 设置为 1 可能会改变记录的顺序，因为如果两个批次被发送到单个分区，第一个分区失败并重试，但是第二个成功，那么可能会出现第二批的记录首先。

【讨论】：

嗨@crypto，感谢您的回答。但是这种方式行不通，因为ConnectionException 是在另一个线程中引发的（进入Sender 类，用于KafkaProducer）。因此我们不能使用try {} catch。
是的，这是一个单独的线程，在 producer.send 方法中创建
所以在调用你的回调之前抛出异常？
当异常抛出时，不会调用回调。此异常仅出现在日志中。而且我们抓不到。
我无法重现该错误。 metadata.fetch.timeout.ms = 60000 在我得到 ConnectionException 之前发生。我尝试了无效的主机：端口组合，并且元数据获取在我看到任何 ConnectionException 之前超时