如何清空Kafka Topic中的记录并判断所有消息已消费完成以便开展维护

核心逻辑 #

Kafka 没有传统MQ那种直接的队列深度统计，但可以通过偏移量比对的方式实现相同的判断效果：Kafka消息按分区存储，每条消息对应唯一的递增偏移量，只要目标Topic所有分区的「消费者组已提交偏移量」等于「分区最新末尾偏移量」，且生产者已停止写入无新消息进入，就说明所有存量消息已经被消费完毕。

实操步骤 #

• 第一步：先停止所有向目标Topic写入数据的生产者，避免判断过程中有新消息进入导致结果不准。
• 第二步：查询目标Topic所有分区的最新末尾偏移量，使用Kafka自带工具脚本即可：

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker地址:端口> --topic <目标Topic名称> --time -1

其中--time -1代表取分区当前最新的末尾偏移量。

• 第三步：查询消费该Topic的消费者组已提交偏移量：

kafka-consumer-groups.sh --bootstrap-server <broker地址:端口> --group <消费者组名称> --describe

输出结果中CURRENT-OFFSET列为消费者组已提交的偏移量，LOG-END-OFFSET列即为分区当前的最新末尾偏移量。

• 第四步：比对所有分区的偏移量，如果该Topic所有分区的CURRENT-OFFSET都和LOG-END-OFFSET相等，就说明该消费者组已经消费完全部存量消息。

注意事项 #

• 若消费者配置的是自动提交偏移量，建议停止生产后等待至少auto.commit.interval.ms配置的时长再查询偏移量，避免最后一批消息已消费但偏移量还未提交导致判断错误。
• 若同一个Topic有多个消费者组消费，需要逐个校验所有消费者组的偏移量全部对齐后，才能确认全链路消费完毕。
• 后续要清空Topic时，确认生产消费全部停止后，可以先将Topic的消息保留时间临时调整为1秒，等待消息自动清理完成后再改回原有配置，比删除重建Topic的操作风险更低。

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