如何解决kafka-python连接MSK的KafkaTimeoutError元数据更新失败问题

你已通过命令行完成生产消费全流程验证，说明VPC连通性、安全组规则、集群状态、主题配置均正常，9098端口是MSK Serverless的标准接入端口，和端口配置无关。
报错核心原因是kafka-python连接的认证配置缺失：MSK Serverless不支持纯SSL无认证的接入方式，必须使用SASL_SSL安全协议搭配AWS IAM做身份校验。你当前代码仅配置了SSL，未携带合法身份凭证，集群会直接丢弃连接请求，最终触发元数据拉取超时。

• 安装必要依赖
  kafka-python本身未内置MSK IAM认证的实现，需要额外安装官方提供的IAM认证签名包：

pip install kafka-python aws-msk-iam-sasl-signer-python

• 修正生产者代码
  需要补全SASL认证配置，同时增加消息序列化逻辑，修正后的可直接运行代码如下：

from kafka import KafkaProducer
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider

# 替换为你的MSK集群实际所在AWS区域，例如us-east-1、cn-north-1
AWS_REGION = "你的集群所在区域"

# 定义Token提供器，自动读取当前环境的AWS凭证（EC2实例角色、本地~/.aws凭证等）
class MSKTokenProvider:
    def token(self):
        token, _ = MSKAuthTokenProvider.generate_auth_token(AWS_REGION)
        return token

# 替换为你控制台查到的集群bootstrap端点，注意核对拼写，不要出现类似reagion的笔误
BOOTSTRAP_SERVER = "boot-name.c1.kafka-serverless.region-name.amazonaws.com:9098"
TOPIC_NAME = "my-topic-name"

producer = KafkaProducer(
    bootstrap_servers=[BOOTSTRAP_SERVER],
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    sasl_oauth_token_provider=MSKTokenProvider(),
    # 自动将字符串消息序列化为字节，避免类型错误
    value_serializer=lambda msg: msg.encode("utf-8")
)

# 发送消息并等待结果确认
send_future = producer.send(TOPIC_NAME, value="test message")
# 阻塞等待发送结果，有异常会直接抛出
record_meta = send_future.get(timeout=10)
print(f"消息发送成功，所在分区：{record_meta.partition}，偏移量：{record_meta.offset}")

producer.close()

• 校验EC2实例的IAM权限
  确保运行代码的EC2实例绑定的IAM角色，拥有对应MSK集群的连接和写入权限，最小权限参考如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:WriteData"
            ],
            "Resource": "你的MSK集群ARN"
        }
    ]
}

• 核对bootstrap端点拼写，尤其是区域名部分，笔误会直接导致连接失败
• 不要尝试用纯SSL、PLAINTEXT等协议连接MSK Serverless，这些协议均不被支持
• 不要在代码里硬编码AK/SK，优先使用EC2实例角色、环境变量等标准方式传递AWS凭证
• 消息发送后需要调用get()等待结果或者调用producer.flush()，否则进程提前退出时，缓冲区的消息可能未实际发出

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