如何重置、更新或清除 Kafka 中某个主题的每个消费者组的偏移量？答案

【问题标题】：How do i reset, update or clear the offset of each consumer group for a topic in Kafka?如何重置、更新或清除 Kafka 中某个主题的每个消费者组的偏移量？
【发布时间】：2017-10-20 09:29:17
【问题描述】：

我有一个主题，假设为 test001，假设主题中有 10000 条消息。我有两个消费者组，比如说 test-group1 和 test-group2 来消费来自上述主题的消息。

如果 test-group1 消费者已消费 4000 条消息，而 test-group2 消费者已消费 4500 条消息，我该怎么办：

将 test-group1 消费者组的偏移量重置为 0？
将 test-group1 消费者组的偏移量更新为 4500？
从主题中删除消息并将所有消费者组的偏移量重置为0？

【问题讨论】：

您在 Kafka 的文档中发现了哪些关于这些问题的信息？
apache kafka 的官方文档中没有正确指定任何内容。

标签： apache-kafka apache-zookeeper

【解决方案1】：

我认为您不能在消费者组级别重置偏移量。您可以使用 seek 方法（在 Java 客户端 API 中）移动到分区的开头（偏移量 0）、结尾或您选择的任何其他偏移量。尝试探索一些 CLI 选项，例如 kafka-consumer-groups.sh、kafka-topics.sh

【讨论】：

您有任何 CLI 选项吗？
据我回忆，我不这么认为。

【解决方案2】：

这张票表明可以直接生产到 __consumer_offsets 主题以覆盖偏移量，使用特殊的“__admin_client”id：

https://issues.apache.org/jira/browse/KAFKA-5246

我不熟悉 __consumer_offsets 主题消息的格式。这篇文章可能会有所帮助，但您需要自己进行更多挖掘：

http://dayooliyide.com/post/kafka-consumer-offsets-topic/

编写一个给定组 id 的应用程序可能会更简单，它会搜索给定位置并提交偏移量。

【讨论】：

【解决方案3】：

为每个主题+分区+group.id存储偏移量，而不是整个主题的整体。您不能删除已提交的偏移量，只能提交较新的偏移量，或等待它们从 _consumer-offsets 主题到期（默认为 24 小时）。

在 0.11 中将有一个偏移管理工具，因此您可以独立于使用应用程序从 CLI 更改偏移。

【讨论】：