带有Kafka偏移量管理的Apache Storm

Question

我已经用Storm构建了一个示例拓扑，使用Kafka作为源。这是一个我需要解决的问题。

每次我杀死一个拓扑并重新启动它时，该拓扑都从一开始就开始处理。

假设Topology处理了主题X中的消息A，然后我终止了该拓扑。

现在，当我再次提交拓扑时，消息A仍然存在，主题X再次被处理。

有没有一个解决方案，也许某种偏移管理来处理这种情况。

Answer 1

在创建spoutconfig时，请确保它有一个固定的spout id，在重新启动后可以通过它来标识自己。

来自官方风暴站点：

重要提示：在重新部署拓扑时，确保未修改spoutconfig.zkroot和spoutconfig.id的设置，否则spout将无法从ZooKeeper读取其先前的使用者状态信息（即偏移量）--这可能导致意外行为和/或数据丢失，这取决于您的用例。

Answer 2

对于新代码，您不应该使用storm-kafka，它是不推荐使用的，因为底层客户端API在Kafka中是不推荐使用的，并且从2.0.0开始被删除。相反，使用storm-kafka-client。

使用storm-kafka-client您希望设置组id和第一轮询偏移策略。

KafkaSpoutConfig.builder(bootstrapServers, "your-topic")
            .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup")
            .setFirstPollOffsetStrategy(UNCOMMITTED_EARLIEST)
            .build();

以上将使您的喷口开始在最早的偏移第一次你启动它，然后它将拾起它离开的地方，如果你重新启动它。组id被Kafka用来在喷口重新启动时识别，因此它可以拿回存储的偏移量检查点。其他偏移量策略的行为会有所不同，您可以检查javadoc中的FirstPollOffsetStrategy枚举。

spout会定期检查它到达多远，在config中也有一个设置来控制这一点。检查点由配置中的setprocessingguranure设置控制，可以设置为至少一次（只有检查点加密的偏移量）、最多一次（spout发出消息之前的检查点）和“任何时间”（定期检查点，忽略ACK）。

看看Storm https://github.com/apache/Storm/blob/dc56e32f3dcdd9396a827a85029d60ed97474786/examples/storm-kafka-client-examples/src/main/Java/org/apache/Storm/kafka/spout/kafkaPoutTopologyMainNamedTopics.Java#l93中包含的一个示例拓扑。