如何更改 Kafka 拓扑的消费者偏移量？

Question

现在，我正在开发一个使用 Kafka Streams 处理消息的项目。我们的消息由两个标识符组成，一个是用户的标识符，第二个是用户所属的消息列表的标识符。

应用程序的故事是，客户希望通过我的应用程序向他们的用户列表发送推送消息。因此，客户端在向用户发送推送消息之前从外部源加载用户列表，之后开始发送。但客户端也希望随时停止正在进行的发送。

此时我的问题出现了。我想停止发送推送消息，而无需通过其流 api 消耗来自 kafka 的所有消息。我用谷歌搜索将流源的偏移量更改为最新，但找不到合适的方法来完成它。

我怎样才能完成它？ 谢谢。

Answer 1

Kafka Streams 不支持这一点。如果您确实需要这样做，您可以停止 Kafka Streams 并使用bin/kafka-consumer-groups.sh 手动为 Kafka Streams 应用程序寻找结束（注意：application.id == group.id）。之后您可以重新启动 Kafka Streams。

顺便说一句：在即将发布的 1.1 版本中，bin/kafka-streams-application-reset.sh 也将允许操作偏移量。我希望我们还设法使内部使用的类 StreamsResetter 成为公共 API——这将允许您在我们的应用程序中以编程方式执行此操作，而无需回退到命令行。