ps：命令在kafka/bin下执行

查看全部消费者组

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --list

查看groupid对应的消费情况

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --describe

重置offset 到最新

bin/kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --reset-offsets --topic icp_data_quality -to-latest --execute

取指定分区 指定topic的 最新多少条

./kafka-console-consumer.sh --bootstrap-server kafka集群地址 --topic 主题名 --partition 分区号 --max-messages 取出数据数量

手动设置offset

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --topic 主题名:分区号 --reset-offsets --to-offset offset位置数 –execute

手动重置offset到最早

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --topic 主题名 --reset-offsets --to-earliest --execute

手动重置offset到最新

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --topic 主题名 --reset-offsets --to-latest --execute

手动重置offset到指定时间

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --reset-offsets --topic 主题名 --to-datetime 2021-10-10T12:30:00.000

手动重置offset到距离当前时间多久以前 PT xxH xxM xxS

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --group 消费者组ID --reset-offsets --topic 主题名 --by-duration PT1H0M0S

支持的 重置offset参数

--to-earliest：把位移调整到分区当前最小位移
--to-latest：把位移调整到分区当前最新位移
--to-current：把位移调整到分区当前位移
--to-offset ： 把位移调整到指定位移处 --to-offset 100
--shift-by N： 把位移调整到当前位移 + N处，注意N可以是负数，表示向前移动
--to-datetime ：把位移调整到大于给定时间的最早位移处，格式是yyyy-MM-ddTHH:mm:ss.xxx
--by-duration ：把位移调整到距离当前时间指定间隔的位移处，格式是PnDTnHnMnS
--from-file ：从CSV文件中读取调整策略

支持的 topic参数

--all-topics（为consumer group下所有topic的所有分区调整位移）
--topic topic名称（为指定的topic的所有分区调整位移）
--topic topic名称:0,1,2(为指定topic的012分区 处理)

自建kafka启动

./kafka-server-start.sh -daemon ../config/server.properties

创建topic（2个分区，1个副本）

./kafka-topics.sh --create --zookeeper zk地址 --replication-factor 1 --partitions 2 --topic topic名称

删除topic

./kafka-topics.sh --zookeeper zk地址 --delete --topic topic名称

查询集群描述（列出kafka topic 及分区）

./kafka-topics.sh --describe --zookeeper zk地址

topic列表查询（列出topic名）

./kafka-topics.sh --zookeeper zk地址 --list

查看指定 Topic 明细

./kafka-topics.sh --describe --bootstrap-server 集群地址 --topic topic名称
PartitionCount：partition 个数
ReplicationFactor：副本个数
Partition：partition 编号，从 0 开始递增。
Leader：当前 partition 起作用的 breaker.id
Replicas: 当前副本数据所在的 breaker.id，列表，排在最前面的有效
Isr：当前 kakfa 集群中可用的 breaker.id 列表

查看topic内消息

./kafka-console-consumer.sh --bootstrap-server kafka集群地址 --topic topic名称 --from-beginning

console消费者 （控制台看kafka消息）

./kafka-console-consumer.sh --bootstrap-server kafka集群地址 --topic topic名称

查看现存消费者

./kafka-consumer-groups.sh --bootstrap-server lkafka集群地址 --list

消费者详情查看

./kafka-consumer-groups.sh --bootstrap-server kafka集群地址 --describe --group 消费者组名称（现存消费者列出的）