在实际生产环境中，我们可能遇到这种情况：
1、topic中有大量的消息需要消费；
2、消费服务由于负载问题不能在短时间内消费掉topic内的消息；
3、由此造成了消费服务负载过高，甚至服务进程阻塞；
在这种情况下，我们应该需要手动去调整某消费者组对应的topic的offset，使整个系统进入平稳。
前提条件
相关所有消费服务必须停止
只有相关所有的消费服务停止，才能修改对应topic的offset。
当前环境
当前kafka的安装是使用confluent kafka安装，所以在实际命令中根据不同环境使用对应的工具。
修改过程
执行如下命令即可：
查看消费者组1
2
3
4
5# kafka-consumer-groups --bootstrap-server localhost:9092 --list
1
_confluent-controlcenter-5-1-2-1-command
_confluent-controlcenter-5-1-2-1
此次我们要修改的消费者组就是1。
查看消费者组订阅的topic以及topic信息1
2
3
4
5
6
7# kafka-consumer-groups --bootstrap-server localhost:9092 --group 1 --describe
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
discovery1.test.goods_info 0 1250 1250 0 1-9b463cba-e36d-4d4c-ae2b-660831449796 /172.19.138.54 1
discovery1.test.risk_control_info 0 161 161 0 1-9b463cba-e36d-4d4c-ae2b-660831449796 /172.19.138.54 1
discovery1.test.calculate_user 0 56 56 0 1-9b463cba-e36d-4d4c-ae2b-660831449796 /172.19.138.54 1
discovery1.test.order_info 0 7322 7322 0 1-9b463cba-e36d-4d4c-ae2b-660831449796 /172.19.138.54 1
注意看current-offset和log-end-offset还有lag，分别为当前偏移量，结束的偏移量，落后的偏移量。
以topics：discovery1.test.order_info为例，当前已消费完成。此处修改current-offset，让消费服务再重新消费最后几条消息。1
2
3
4# kafka-consumer-groups --bootstrap-server localhost:9092 --group 1 --topic discovery1.test.order_info --execute --reset-offsets --to-offset 7312
TOPIC PARTITION NEW-OFFSET
discovery1.test.order_info 0 7312
修改成功，再查看topic信息1
2
3
4
5
6
7
8# kafka-consumer-groups --bootstrap-server localhost:9092 --group 1 --describe
Consumer group '1' has no active members.
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
discovery1.test.goods_info 0 1250 1250 0 - - -
discovery1.test.risk_control_info 0 161 161 0 - - -
discovery1.test.calculate_user 0 56 56 0 - - -
discovery1.test.order_info 0 7312 7322 10 - - -
现在明显是lag=10，current-offset=7312，log-end-offset=7322，告诉我们有10条未消费。
current-offset当前已经消费到偏移量为7312,可以理解为已经消费7312条。
log-end-offset可以理解为总共7322条记录。
lag可以理解为未消费记录条数。
现在可以重新启动消费服务，可以从日志中看到会重新消费最后10条消息。
