如何列出Kafka集群中有数据流入的所有主题？

方法1：通过消费者组偏移量判断 #

如果集群内有长期运行的消费者组，可通过偏移量变化识别活跃主题：

• 列出所有消费者组：

  kafka-consumer-groups --list --bootstrap-server <你的Kafka地址:端口>
  

• 查看目标消费者组的偏移量详情，重点关注LOG-END-OFFSET（分区最新偏移量）和CURRENT-OFFSET（消费者当前偏移量）：

  kafka-consumer-groups --describe --group <消费者组名> --bootstrap-server <你的Kafka地址:端口>
  

  若LOG-END-OFFSET在短时间内持续增长，或LAST-LAG（未消费消息数）从0重新变为非零，说明对应主题有新数据流入。

方法2：定期对比主题分区偏移量 #

无长期消费者时，可通过两次获取分区最新偏移量的差值判断：

• 编写bash脚本定期检测：

  # 首次获取所有主题的最新偏移量并保存
  kafka-run-class kafka.tools.GetOffsetShell --bootstrap-server <你的Kafka地址:端口> --topic '*' --time -1 > offsets_first.txt
  
  # 等待30分钟（可按需调整）后再次获取
  sleep 1800
  kafka-run-class kafka.tools.GetOffsetShell --bootstrap-server <你的Kafka地址:端口> --topic '*' --time -1 > offsets_second.txt
  
  # 对比结果，输出偏移量增长的主题
  diff offsets_first.txt offsets_second.txt | grep '>' | awk -F ':' '{print $1}' | sort | uniq
  

  注：--time -1表示获取分区最新偏移量，--topic '*'匹配所有主题。

方法3：通过Confluent Control Center可视化查看 #

若部署了Confluent Control Center，直接在监控界面查看主题的Messages In Per Second指标，筛选出近期有消息流入的主题，这种方式更直观高效。

注意事项 #

• 由于主题数据最长保留7天，上述方法仅能检测近7天内有数据流入的主题，超过保留期的历史数据偏移量已被清理，无法回溯。
• 多分区主题需至少有一个分区偏移量增长，才算活跃主题。

内容的提问来源于stack exchange，提问作者CMPE