如何检查Kafka主题是否已被清理？修改retention.ms后验证遇问题

首先，你的代码逻辑存在几个关键问题，导致它返回None而不是预期的True/False：

核心问题点 #

• 循环逻辑错误：你的for循环里第一次迭代就直接return，不管是收到消息还是没收到——这意味着如果主题里有消息，你只会检查第一条就返回False；如果没有消息，循环根本不会执行（因为消费者超时后会退出循环），函数没有任何return语句，自然返回None。
• 偏移量设置不合理：auto_offset_reset='latest'会让消费者从主题的最新偏移量开始消费，如果主题里还有旧消息但当前偏移量已经走到末尾，你会完全看不到这些剩余消息，导致误判主题为空。

修复后的代码 #

from kafka import KafkaConsumer

def is_topic_empty(topic, bootstrap_servers):
    consumer_kafka = KafkaConsumer(
        topic,
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest',  # 从最早的偏移量开始消费，确保能看到所有剩余消息
        consumer_timeout_ms=10000  # 10秒没收到消息就退出
    )
    has_messages = False
    for message in consumer_kafka:
        has_messages = True
        break  # 只要收到一条消息就可以确定主题非空，无需继续消费
    consumer_kafka.close()  # 关闭消费者，释放资源
    return not has_messages  # 没有消息返回True，有消息返回False

# 使用示例
if is_topic_empty("your_topic_name", ["your_bootstrap_server:9092"]):
    # 发送JSON文件的逻辑
    print("主题已空，发送JSON文件")
else:
    print("主题还有剩余消息")

额外验证方法 #

除了消费验证，你还可以用Kafka命令行工具直接检查主题状态，结果更准确：

• 查看主题的消息总数：

  kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list your_bootstrap_server:9092 --topic your_topic_name --time -1 | awk -F ":" '{sum += $3} END {print sum}'
  

• 检查消费者组的偏移量（确认是否所有消息都被消费）：

  kafka-consumer-groups.sh --bootstrap-server your_bootstrap_server:9092 --describe --group your_consumer_group_name
  

  如果CURRENT-OFFSET等于LOG-END-OFFSET，说明该消费者组已经消费完所有消息。

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