Python Kafka Consumer无法接收Debezium投递的Kafka消息

核心故障点是Kafka容器的监听器配置错误：从watcher输出的日志可以看到，当前Kafka的ADVERTISED_LISTENERS配置为Docker内网地址172.17.0.6:9092，该地址仅对加入同一Docker网桥的容器可见。
你在宿主机运行的Python消费者虽然能通过映射的localhost:9092端口和Kafka建立初始连接，但Kafka会把自身的内网通告地址返回给客户端，后续消费者拉取消息的请求会发往这个宿主机无法访问的内网IP，最终表现为连接状态正常、订阅关系正常，但永远拉取不到消息。
另外你贴的Kafka Connect启动命令存在语法错误，镜像名前多写了一个--，会导致容器启动参数解析失败。

• 停掉当前运行的Kafka容器，使用修正后的命令重新启动，显式配置正确的对外通告地址：

docker run -it --rm --name kafka -p 9092:9092 \
--link zookeeper:zookeeper \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
quay.io/debezium/kafka:1.9

• 停掉当前运行的Kafka Connect容器，删掉命令里镜像名前多余的--后重新启动：

docker run -it --rm --name connect -p 8083:8083 \
-e GROUP_ID=1 \
-e CONFIG_STORAGE_TOPIC=my_connect_configs \
-e OFFSET_STORAGE_TOPIC=my_connect_offsets \
-e STATUS_STORAGE_TOPIC=my_connect_statuses \
--link zookeeper:zookeeper \
--link kafka:kafka \
quay.io/debezium/connect:1.9

• 等Kafka、Zookeeper、Connect服务全部启动完成后，重新提交Debezium连接器创建请求，等待连接器完成初始快照同步。
• 调整Python消费者代码，增加显式版本声明和超时配置避免无意义卡住，参考代码如下：

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test_server.testDB.test_table',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    api_version=(2,8,1), # 匹配Debezium 1.9内置的Kafka版本，跳过版本自动探测
    consumer_timeout_ms=15000 # 15秒拉取不到消息直接抛出超时异常，方便排查
)

print(consumer.bootstrap_connected())
print(consumer.subscription())
for message in consumer:
    print(message.value.decode('utf-8'))

• 容器内部的watcher工具之所以能正常消费，是因为它和Kafka在同一个Docker内网中，可以直接访问Kafka通告的内网IP，不受宿主机网络限制。
• 配置修复后，宿主机上的所有客户端（包括Python消费者、本地命令行工具）都可以通过localhost:9092正常完成元数据拉取和消息消费流程。

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