java - 在 KafkaBolt 中暴露 Kafka 发布异常

Question

我使用KafkaBolt在 Storm 中向各种 Kafka 主题发布消息。我想在发布逻辑周围放置日志记录和指标，以便我可以围绕发布失败时可能引发的任何异常创建警报。公开这些异常是通过传递到 KafkaProducer.send() 的Callback 函数来完成的。 ，发布成功或失败后执行。

问题是 KafkaBolt 完全封装了它的 KafkaProducer，因此无法注入(inject)自定义 Callback，所以如果我想看到任何错误我必须在 Storm UI 中查看。我通过为 KafkaBolt 创建包装器来解决这个问题。反过来，该包装器会将传入 KafkaBolt.prepare() 的 OutputCollector 包装在自定义 OutputCollector 中，该自定义 OutputCollector 会覆盖 的行为OutputCollector.reportError()。然后我可以在那里添加我自己的日志记录和指标报告代码，然后让它调用原始方法。

这个解决方案似乎完全足以满足我的需要，但奇怪的是 KafkaBolt 使得以编程方式访问这些异常变得如此困难。我想知道我是否遗漏了一些明显的东西，以及是否有更好的方法来做到这一点。

Answer 1

我认为您没有遗漏任何东西，您可能只是第一个有这种需求的人。必须有人解决这个问题并决定解决它:)

如果您想对 Bolt 进行更改以支持自定义错误处理(例如，通过允许用户按照您的建议提供回调)，您可以在 https://issues.apache.org/jira/projects/STORM/issues 提出问题并针对 https://github.com/apache/storm/pulls 进行 PR 。当然，也欢迎您只提出问题，其他人可能会看到它并决定修复它，但自己贡献修复可能会更快。

编辑:您可以在 https://github.com/apache/storm/blob/master/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/bolt/KafkaBolt.java 找到 bolt 代码