Kafka——》消费组管理kafka-consumer-groups_kafka-consumer-groups命令

推荐链接：
    总结——》【Java】
    总结——》【Mysql】
    总结——》【Redis】
    总结——》【Kafka】
    总结——》【Spring】
    总结——》【SpringBoot】
    总结——》【MyBatis、MyBatis-Plus】

KAFKA_HOME=/opt/app/install/kafka

list：查看消费者组列表

# 查看消费者组列表
bin/kafka-consumer-groups.sh --bootstrap-server x.x.x.x:9092 --list
1
2

describe：查看消费者组详情

# 查看所有消费者组详情
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups
    
# 查看指定消费者组详情
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1
1
2
3
4
5

state：查看消费者组状态

# 查看所有消费者组状态 
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups --state

# 查看指定消费者组状态
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1 --state
1
2
3
4
5

members：查看消费者组成员

# 查看所有消费者组成员
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --all-groups --members

# 查看指定消费者组成员    
bin/kafka-consumer-groups.sh --bootstrap-server 10.116.192.27:9092 --describe --group listenerForSyncEsfCommunity1 --members
1
2
3
4
5

delete：删除消费者组

# 删除所有消费者组
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete --all-groups

# 删除指定消费者组
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete --group listenerForSyncEsfCommunity1

    

# 只有这个消费组的所有客户端都停止消费/不在线才能够成功删除，否则会报下面异常
Error: Deletion of some consumer groups failed:
* Group 'listenerForSyncEsfCommunity1' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.
1
2
3
4
5
6
7
8
9
10
11

reset-offsets：重置消费组的偏移量

# 重置指定消费组的所有Topic的偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --group listenerForSyncEsfCommunity1 --all-topic --dry-run

# 重置指定消费组的指定Topic的偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --group listenerForSyncEsfCommunity1 --topic test_topic --dry-run

# 重置所有消费组的所有Topic的偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --all-group --all-topic --dry-run 

# 重置所有消费组的指定Topic的偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --reset-offsets --to-earliest --all-group --topic test_topic --dry-run



# 只有这个消费组不可用状态才能重置成功，否则会报下面异常：
Error: Assignments can only be reset if the group 'listenerForSyncEsfCommunity1' is inactive, but the current state is Stable.

TOPIC                          PARTITION  NEW-OFFSET     

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

delete-offsets：删除消费组的偏移量

# 删除指定消费组的指定Topic的偏移量（下一次从头消费）
bin/kafka-consumer-groups.sh --bootstrap-server 10.122.139.40:9092 --delete-offsets --group listenerForSyncEsfCommunity1 --topic test_topic

# 只有这个消费组不可用状态才能删除成功，否则会报异常
1
2
3
4