KafkaProducer连接MSK报错：60秒未更新元数据问题求助

bootstrap_servers配置说明 #

不存在“主Bootstrap Server”的概念，生产环境建议配置2个Broker的全部接入地址。
Bootstrap地址列表仅用于客户端初始接入集群，客户端成功连上任意一个可用节点后，会自动拉取全集群元数据（包含所有Broker地址、Topic分区分布等信息）。配置多地址的作用是做接入点冗余，避免单个节点故障时客户端无法完成初始连接，不需要区分主从节点。

故障根因排查（按优先级排序） #

从日志看TCP连接可以建立但立即断开，结合你使用AWS MSK服务的场景，故障基本是以下原因导致：

1. 接入配置与MSK集群要求不匹配
   你代码里初始写的bootstrap_servers=['localhost:9092']是本地Kafka默认地址，完全无法连接MSK。即便你后续替换了地址，也要注意MSK不同接入方式对应不同端口：
   • 明文接入：9092端口
   • TLS加密接入：9094端口
   • IAM鉴权接入：9098端口
     必须使用MSK控制台提供的对应接入模式的地址串，同时如果集群开启了TLS加密、SASL/IAM鉴权，必须在KafkaProducer参数中补充对应security.protocol、SASL机制、凭证加载配置，否则客户端发起明文请求后会被服务端直接断开连接。另外你代码里写的api_version=(0, 10, 1)版本过旧，和当前MSK运行的Kafka版本差距过大可能导致协议不兼容，建议改成和你MSK集群大版本一致的版本号。
2. 网络访问策略拦截
   确认运行Airflow任务的节点和MSK集群网络连通：要么同VPC，要么已经配置VPC对等连接/私网访问通道；同时检查MSK集群绑定的安全组入站规则，必须放通客户端节点所属IP段对应接入端口的访问，网络ACL没有对对应端口做拦截。
   连接建立后立即断开是安全组/ACL拦截的典型表现——TCP三次握手可以完成，但后续Kafka协议报文被拦截，或者服务端返回的报文客户端收不到，最终连接超时断开。

现有代码的逻辑错误 #

• 重复查询数据库：你已经通过pd.read_sql把数据读入DataFrame，后续发送消息时又执行了一次pd.read_sql(query, conn)，浪费数据库资源；且直接遍历pd.read_sql返回的是表的列名，不是数据行，逻辑完全错误。
• 序列化逻辑冲突：你已经在KafkaProducer初始化时配置了json序列化器，发送消息时又对数据调用encode('utf-8')，会导致序列化报错。
• 缺少消息刷新逻辑：发送完消息没有调用producer.flush()，进程退出时缓冲区未发送的消息会直接丢失，也可能导致连接异常断开。

修正后参考代码 #

import json
import pandas as pd
from kafka import KafkaProducer
from airflow.providers.oracle.hooks.oracle import OracleHook

try:
    conn = OracleHook(oracle_conn_id=oracle_conn_id).get_conn()
    query = "Select * from sales"
    df = pd.read_sql(query, conn)
    conn.close()

    topic = 'my-topic'
    producer = KafkaProducer(
        # 替换为MSK控制台获取的真实Broker地址列表
        bootstrap_servers=['<MSK-broker1-address>:<port>','<MSK-broker2-address>:<port>'],
        value_serializer=lambda x: json.dumps(x, default=str).encode('utf-8'),
        # 开启鉴权/加密时补充对应安全配置
        api_version=(2, 8, 1) # 替换为实际MSK集群对应的Kafka大版本
    )

    for _, row in df.iterrows():
        producer.send(topic, value=row.to_dict())
    
    # 等待所有缓冲区消息发送完成
    producer.flush()
    producer.close()
    print(f"成功发送{len(df)}条记录")

except Exception as error:
    raise error

连通性验证方法 #

先在运行Airflow任务的节点上用命令行工具测试连通性，不要直接调度任务：
kafkacat -b <MSK-bootstrap地址串> -L
如果命令能正常返回集群元数据、Topic列表，说明网络、鉴权配置正常，再调试Python代码；如果命令执行失败，优先排查网络和安全组配置，无需调整代码。

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