CentOS7双节点Kafka仅单节点无法消费数据问题求助

针对你遇到的单台CentOS服务器上Kafka仅能发送消息、消费者无输出无报错的问题，可按以下步骤排查：

1. 用Kafka原生工具验证基础功能 #

先排除Python脚本的问题，用Kafka自带的控制台工具测试：

测试控制台消费者 #

执行以下命令，尝试消费logs主题的所有历史消息：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic logs --from-beginning

再开一个终端，用控制台生产者发送消息：

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic logs

输入任意文本，观察控制台消费者是否能接收：

• 如果能接收：问题出在Python消费者脚本或依赖库上
• 如果不能接收：问题出在Kafka Broker或主题配置上

2. 检查消费者组偏移量状态 #

你的Python消费者指定了group_id="my-group"，如果该组之前已经消费过logs主题，且偏移量已经追赶到最新位置，即使设置了auto_offset_reset="earliest"也不会重新消费历史消息。

执行以下命令查看消费者组的偏移量详情：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

重点关注CURRENT-OFFSET和LOG-END-OFFSET：

• 如果两者相等：说明该组已经消费完所有消息，新发送的消息需等待生产者发送后才能接收
• 如果CURRENT-OFFSET为空：说明该组未初始化偏移量，可尝试重置偏移量：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --to-earliest --group my-group --topic logs --execute

3. 验证主题配置与状态 #

检查logs主题的分区、副本及Leader状态，确保主题正常可用：

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic logs

确认：

• 主题存在，分区数量符合预期
• 每个分区的Leader为当前Broker的ID（无Leader: -1的异常情况）
• ISR列表包含当前Broker节点

4. 排查Python消费者脚本问题 #

检查依赖库版本 #

对比两台服务器的kafka-python版本，版本不一致可能导致兼容性问题：

pip show kafka-python

若版本差异较大，将问题服务器的库版本更新为正常服务器的版本：

pip install kafka-python==<正常服务器的版本号>

更换订阅方式测试 #

将subscribe改为手动指定分区的assign方式，避免自动订阅的潜在问题：

from kafka import KafkaConsumer, TopicPartition

broker_url = "localhost:9092"
topic_name = "logs"

consumer = KafkaConsumer(bootstrap_servers=broker_url, 
                         group_id="my-group",
                         auto_offset_reset="earliest",
                         value_deserializer=lambda x: x.decode("utf-8"))

# 手动绑定主题的所有分区
partitions = [TopicPartition(topic_name, p) for p in consumer.partitions_for_topic(topic_name)]
consumer.assign(partitions)

for message in consumer:
    print(f"收到消息: {message.value}")

增加调试日志 #

在Python消费者中添加调试日志，查看是否有隐藏的连接或订阅问题：

import logging
logging.basicConfig(level=logging.DEBUG)

# 后续保持原消费者代码不变

5. 检查Kafka Broker监听配置 #

确认server.properties中的监听配置正确，避免本地连接异常：

grep -E 'listeners|advertised.listeners' /etc/kafka/server.properties

确保配置包含：

listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092

若配置有误，修改后重启Kafka服务。

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