Apache storm拓扑中的Apache kafka偏移量提交

Question

我正在设计一个ApacheStorm拓扑（使用streamparse），它由一个喷口（ApacheKafka喷口）和一个具有并行性的螺栓构建

螺栓分批读取信息。如果批量成功完成，我手动提交apache kafka偏移。

当mysql上的螺栓插入失败时，我不会在Kafka中提交偏移量，但是一些消息已经在喷口发送到螺栓的消息队列中。

应该删除队列中已经存在的消息，因为我无法在不丢失先前失败消息的情况下推进kafka偏移量。

streamparse中是否有方法在启动时清除队列中已经存在的所有消息或使其失败？

Answer 1

我不知道streamparse，但我得到的印象是，您希望将元组捆绑起来，并将它们作为批处理编写。假设您已写入偏移量10。现在，您的螺栓收到偏移量11-15，批处理写入失败。偏移量15-20已排队，您不希望立即处理它们，因为这样会无序处理批次。

这种理解正确吗？

首先，我将放弃手动提交偏移量。你应该让壶嘴来处理。假设您使用的是storm kafka client，则可以将其配置为在确认相应的元组和所有前面的元组后仅提交偏移量。

您可能应该做的是在螺栓中跟踪（或者更好的是，在您的数据库中）失败批次中的最高偏移量。然后，当您的Bolt未能写入偏移量11-15时，您可以使用偏移量使Bolt失败每个元组

此解决方案假设您不在喷口和写入器螺栓之间对消息流进行重新排序，因此消息按照它们发出的顺序到达螺栓。