如何重置，更新或清除每個消費羣組在Kafka中的主題的偏移量？

Question

我有一個話題，我們假設爲test001，並假設主題中有10000條消息。我有兩個消費者小組的讓我們說test-group1和test-group2來消費上述話題的消息。

如果試驗1組消費者已經消耗4000信息和試驗組2消費者已經消耗了4500的消息使我該怎麼辦：

1. 復位偏移測試組1消費羣的0？
2. 更新test-group1消費者組偏移量爲4500？
3. 從主題中刪除消息並將所有消費者組的偏移重置爲0？

Answer 1

我不認爲您可以重置消費者組級別的偏移量。您可以使用seek方法（在Java客戶端API中）移至分區的開始位置（偏移量0），結束位置或您選擇的任何其他偏移量。嘗試探索一些CLI選項，如kafka-consumer-groups.sh，kafka-topics.sh

Answer 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不尋求一個給定的位置，並承諾補償。

Answer 3

爲每個主題+分區+ group.id存儲偏移量，而不是整個主題的總體偏移量。您不能刪除已提交的偏移量，只能提交更新的偏移量，或等待它們從_consumer-offsetsets主題（默認爲24小時）過期。

在0.11中會有一個偏移量管理工具，因此您可以獨立於消費應用程序更改來自CLI的偏移量。