Scala：如何异步捕获通过 Kafka 发送的所有多条消息的状态？答案

【问题标题】：Scala: How to capture status of all the multiple messages sent over Kafka asynchronously?Scala：如何异步捕获通过 Kafka 发送的所有多条消息的状态？
【发布时间】：2019-08-27 20:14:30
【问题描述】：

我正在使用 Scala 将消息发布到 Kafka。我有要发布到 kafka 的消息列表，可能是异步的。但是，最后，如果在发布至少一条消息时出现任何错误，我必须抓住它并采取行动。我知道回调方法，但这对每条消息都有效。我想积累或弄清楚所有这些状态的方式，并决定是否没有任何失败。类似于批处理状态。

for (message <- messageList) {
    // Create a message
    // producer is created
    // I have a separate util package, having the MyCallback class with overridden onCompletion method.
    producer.send(new ProducerRecord[String, MyMessage](config("topic"), message ), new MyCallback(config))

}
// here, I need help, to see if any of the messages sent is failed or not.
// How to capture the statuses of callback here for entire batch?


// Separate class file
class MyCallback (config: Map[String, String]) extends Callback {
    override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
     // if successful meta data, do something
     // if exception, I can log the error.

    }
}

基本上，我可以在我的回调类方法中检查单个消息的状态。

我如何在我的 for 循环完成后跟踪这些状态并据此采取进一步行动？

【问题讨论】：

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

【解决方案1】：

如果您在 for 循环之前移动 new MyCallback(config)，那么这可能会有所帮助。
然后你就有了一个可以收集异常的实例。
加上回调对象将在循环后的范围内，您可以检索状态列表

我的 scala 有点生锈，但我认为它显示了这个想法

class MyCallback (config: Map[String, String]) extends Callback {

    private val exceptions = _ // some mutable List

    def getExceptions() { return exceptions }

    override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
     if (exception != null) {
         exceptions.add(exception)
     }
    }
}

然后

val cb = new MyCallback(config)
for (message <- messageList) {
    producer.send(new ProducerRecord[String, MyMessage](config("topic"), message ), cb)
}

// TODO: check cb.getExceptions().size > 0

或者您可以通过在循环内检查来尝试提前失败，但这将是异步的，并且可能无法正常工作

【讨论】：