kafka-consumer-groups.sh消费者组管理

1.查看消费者列表 –list

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --list

在这里插入图片描述

先调用MetadataRequest拿到所有在线Broker列表再给每个Broker发送ListGroupsRequest请求获取消费者组数据。

2. 查看消费者组详情–describe

查看指定消费组详情–group

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --describe --group hy-group

在这里插入图片描述

查看所有消费组详情–all-groups

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --describe --all-groups

在这里插入图片描述

查询消费者成员信息–members

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --describe --all-groups  --members

在这里插入图片描述

查询消费者状态信息–state

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --describe --group hy-group --state

在这里插入图片描述

3. 删除消费者组–delete

删除指定消费组–group

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --delete --group test

删除所有消费组–all-groups

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --delete --all-groups

想要删除消费组前提是这个消费组的所有客户端都停止消费/不在线才能够成功删除;否则会报下面异常

* Group 'hy-group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

4.重置消费组的偏移量 –reset-offsets

重置指定消费组的偏移量 –group

重置指定消费组的所有Topic的偏移量–all-topic

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reset-offsets --to-earliest --group  hy-group --dry-run --all-topic

重置指定消费组的指定Topic的偏移量–topic

bin/kafka-consumer-groups.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --reset-offsets --to-earliest --group  hy-group --dry-run --topic hy-test-topic

重置所有消费组的偏移量 –all-group

重置所有消费组的所有Topic的偏移量–all-topic

bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --dry-run --all-topic

重置所有消费组中指定Topic的偏移量–topic

bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --dry-run --topic test2


参数	描述	例子
–to-earliest :	重置offset到最开始的那条offset(找到还未被删除最早的那个offset)
–to-current:	直接重置offset到当前的offset，也就是LOE
–to-latest：	重置到最后一个offset
–to-datetime:	重置到指定时间的offset;格式为:YYYY-MM-DDTHH:mm:SS.sss;	–to-datetime “2021-6-26T00:00:00.000”
–to-offset	重置到指定的offset,但是通常情况下,匹配到多个分区,这里是将匹配到的所有分区都重置到这一个值; 如果 1.目标最大offset–to-offset ，则重置为最小; 3.否则的话才会重置为–to-offset的目标值; 一般不用这个	–to-offset 3465
–shift-by	按照偏移量增加或者减少多少个offset；正的为往前增加;负的往后退；当然这里也是匹配所有的;	–shift-by 100 、–shift-by -100

–shift-by	按照偏移量增加或者减少多少个offset；正的为往前增加;负的往后退；当然这里也是匹配所有的;	–shift-by 100 、–shift-by -100
–from-file	根据CVS文档来重置