区分如何在异步Kafka producer中处理异常

Question

当生成消息到Kafka时，您可以得到两种错误：可检索和不可检索。在处理它们时，你应该如何区分它们？

我希望异步生成记录，将回调对象接收到不可重试异常的记录保存在另一个主题（或HBase）中，并让生产者为我处理所有接收到可重试异常的记录（最多尝试次数，当它最终到达时，会成为第一批异常之一）。

我的问题是：尽管有回调对象，但生产者是否仍会自行处理可检索的异常？因为在接口回调中说：

可重试异常（暂时的，可通过增加#.重试次数来覆盖）

代码可能是这样的吗？

producer.send(record, callback)

def callback: Callback = new Callback {
    override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {
      if(null != e) {
         if (e == RecordTooLargeException || e == UnknownServerException || ..) {
            log.error("Winter is comming")
            writeDiscardRecordsToSomewhereElse
         } else {
            log.warn("It's no that cold") //it's retriable. The producer will keep trying by itself?
         }
      } else {
        log.debug("It's summer. Everything is fine")
      }
    }
}

Kafka版本：0.10.0

任何光将被赞赏！：）

Answer 1

正如《Kafka圣经》（又名《Kafka权威指南》）所说：

缺点是，虽然commitSync（）将重试提交，直到提交成功或遇到不可重试的失败，但commitSync（）不会重试。

原因是：

“不重试”是指，当commitAsync（）从服务器接收到响应时，可能已经有一个已成功的后续提交。

假设我们发送了一个提交偏移量2000的请求。存在临时通信问题，因此代理永远不会收到请求，因此永远不会响应。同时，我们处理了另一个批次并成功提交了偏移量3000。如果委员会同步（）现在重试以前失败的提交，它可能会在已经处理和提交了偏移量3000之后成功提交偏移量2000。在再平衡的情况下，这将导致更多的重复。

除此之外，您仍然可以创建一个递增的序列号，每次提交时都可以增加该序列号，并将该序列号添加到回调对象中。当重试时间到来时，只需检查Acc的当前值是否等于您给回调的数字。如果是这样，它是安全的，您可以执行提交。否则，将出现新的提交，您不应重试此偏移量的提交。

这似乎有很多麻烦，这是因为如果你正在考虑这个问题，你应该改变你的策略。