Kafka Topic收不到消息：程序poll()返回None但控制台消费者可正常消费

最高优先级修复：线程池提交逻辑错误 #

你main.py中的线程池提交代码存在语法错误，直接导致消费逻辑未按预期运行：

# 错误写法：加了()会直接同步执行run()方法，死循环会卡住主线程，不会提交到线程池
executor.submit(kafka_message_consumer.run())
# 正确写法：传入函数引用，不需要加括号，才会提交到线程池异步执行
executor.submit(kafka_message_consumer.run)

同理，第二行提交kafka_discovery_executor.run()也需要去掉后面的括号。修改后优先测试是否正常消费。

其他常规排查点 #

• 验证消费者组配置
  你代码中固定了group.id，如果该消费者组之前已经消费完了Topic的所有存量消息且自动提交了offset，就算设置auto.offset.reset = 'earliest'也不会重复消费历史消息。而你用kafka-console-consumer.bat时如果没有指定消费者组，默认会生成随机的新消费者组，就可以拉取到历史消息。测试时可以换一个从未使用过的group.id值，确认是否能拉取到消息。
• 校验配置参数正确性
  打印加载后的kafka_properties字典，确认bootstrap.servers的值和你调用kafka-console-consumer.bat时指定的--bootstrap-server参数完全一致，没有出现配置文件key拼写错误、值为空的情况。
• 单独测试最简消费逻辑
  剥离线程池、队列等上层封装逻辑，直接运行最简消费脚本，确认底层消费能力正常：

  from confluent_kafka import Consumer
  import json
  
  with open('kafka_properties.json', 'r', encoding='utf-8') as f:
      kafka_properties = json.load(f)
  
  consumer_config = {
      'bootstrap.servers': kafka_properties.get('bootstrap.servers'),
      'group.id': 'test_new_group_001', # 用全新的消费者组
      'enable.auto.commit': True,
      'auto.offset.reset': 'earliest'
  }
  consumer = Consumer(consumer_config)
  consumer.subscribe(['mytopic'])
  
  while True:
      msg = consumer.poll(1.0)
      if msg is None:
          print("等待消息")
          continue
      if msg.error():
          print(f"消费报错: {msg.error()}")
      else:
          print(f"收到消息: {msg.value().decode('utf-8')}")
  

• 依赖版本兼容校验
  升级confluent_kafka到最新稳定版，避免库版本和Kafka集群版本不兼容的问题：
  pip install --upgrade confluent_kafka
• 网络与权限校验
  确认运行Python代码的设备和Kafka集群的网络连通正常，没有ACL权限限制、防火墙拦截端口的问题，如果你是在同一台设备运行控制台消费者和Python代码，该步骤可跳过。

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