KafkaConsumer重置Offset

在Kafka Version爲0.11.0.0以後，Consumer的Offset信息再也不默認保存在Zookeeper上，而是選擇用Topic的形式保存下來。在命令行中能夠使用kafka-consumer-groups的腳本實現Offset的相關操做。

更新Offset的三個維度：

• Topic的做用域
• 重置策略
• 執行方案

Topic的做用域

• --all-topics：爲consumer group下全部topic的全部分區調整位移）
• --topic t1 --topic t2：爲指定的若干個topic的全部分區調整位移
• --topic t1:0,1,2：爲指定的topic分區調整位移

重置策略

• --to-earliest：把位移調整到分區當前最小位移
• --to-latest：把位移調整到分區當前最新位移
• --to-current：把位移調整到分區當前位移
• --to-offset <offset>： 把位移調整到指定位移處
• --shift-by N： 把位移調整到當前位移 + N處，注意N能夠是負數，表示向前移動
• --to-datetime <datetime>：把位移調整到大於給定時間的最先位移處，datetime格式是yyyy-MM-ddTHH:mm:ss.xxx，好比2017-08-04T00:00:00.000
• --by-duration <duration>：把位移調整到距離當前時間指定間隔的位移處，duration格式是PnDTnHnMnS，好比PT0H5M0S
• --from-file <file>：從CSV文件中讀取調整策略

執行方案

什麼參數都不加：只是打印出位移調整方案，不具體執行

• --execute：執行真正的位移調整
• --export：把位移調整方案按照CSV格式打印，方便用戶成csv文件，供後續直接使用

注意事項

1. consumer group狀態必須是inactive的，即不能是處於正在工做中的狀態
2. 不加執行方案，默認是隻作打印操做

經常使用示例

1. 更新到當前group最初的offset位置

   bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-earliest --execute

2. 更新到指定的offset位置

   bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-offset 500000 --execute

3. 更新到當前offset位置（解決offset的異常）

   bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-current --execute

4. offset位置按設置的值進行位移

   bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --shift-by -100000 --execute

5. offset設置到指定時刻開始

   bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-datetime 2017-08-04T14:30:00.000

   參考文檔：https://zoeminghong.github.io/2018/07/20/Kafka%20ConsumerOffset20180720/