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--grouptest

**删除所有消费组

--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, 这个时候重置为目标最大offset；2.目标最小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文档来重置