EC2部署Kafka：Python生产者消息无法被控制台消费者接收

1. 确认控制台消费者的启动参数 #

控制台消费者默认只会消费启动后新产生的消息，且必须指定正确的主题名：

• 确保主题名拼写完全一致（区分大小写），使用以下命令重新启动消费者：

  bin/kafka-console-consumer.sh --bootstrap-server <EC2公网IP>:9092 --topic demo_test --from-beginning
  

  --from-beginning 参数会让消费者拉取主题中所有历史消息，避免遗漏已发送的内容。

2. 验证消息是否真正被Broker接收 #

Python的flush()仅确保消息被写入客户端缓冲区，不代表Broker已确认接收。修改代码添加确认逻辑：

import pandas as pd
from kafka import KafkaProducer
from time import sleep
from json import dumps
import json

try:
    producer = KafkaProducer(
        bootstrap_servers=['<My-EC2-public-IP>:9092'],
        value_serializer=lambda x: dumps(x).encode('utf-8'),
        acks='all'  # 要求所有同步副本确认消息写入
    )

    # 发送消息并等待Broker确认
    future = producer.send('demo_test', value={'surname': 'parameter'})
    result = future.get(timeout=10)  # 超时时间10秒
    print(f"消息已确认：分区{result.partition}，偏移量{result.offset}")
    
    producer.flush()
except Exception as e:
    print(f"错误：{str(e)}")
finally:
    producer.close()

如果执行后抛出超时或其他异常，说明消息并未真正到达Broker，需进一步排查网络或Broker配置。

3. 检查主题的分区与副本状态 #

使用命令查看demo_test主题的详细配置：

bin/kafka-topics.sh --describe --topic demo_test --bootstrap-server <EC2公网IP>:9092

重点关注：

• Isr（同步副本集）是否包含至少一个副本，若为空则消息无法被持久化
• min.insync.replicas配置是否小于等于实际同步副本数，否则会导致消息确认失败

4. 确认序列化与反序列化的一致性 #

Python生产者使用JSON序列化消息，控制台消费者默认按字节输出，可显式指定字符串反序列化器确保正常显示：

bin/kafka-console-consumer.sh --bootstrap-server <EC2公网IP>:9092 --topic demo_test --from-beginning --value-deserializer org.apache.kafka.common.serialization.StringDeserializer

5. 查看Broker日志排查静默异常 #

检查EC2上Kafka的logs/server.log文件，搜索demo_test相关日志，确认是否存在：

• 权限拒绝（如主题写入权限限制）
• 磁盘空间不足导致消息无法持久化
• 副本同步异常等问题

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