如何使用SASL连接AWS MSK？本地Node.js应用连接失败求助

问题概述 #

在AWS创建MSK集群后，使用本地Node.js应用通过kafkajs连接，尝试SASL/PLAIN和IAM两种认证方式，均触发KafkaJSConnectionClosedError: Closed connection错误。

尝试过的SASL/PLAIN认证代码 #

this.kafka = new Kafka({
          clientId: 'user_service',
          brokers: ['b-1-public.xxxxxxxxxx.xxxxx.c3.kafka.xxxxxxx1.amazonaws.com:9196'],
          sasl: {
            mechanism: 'plain',
            username: 'bloom-msk',
            password: "bloom-msk-dev-secret"
          },
        });

尝试过的IAM认证代码 #

this.kafka = new Kafka({
      clientId: 'user_service',
      brokers: ['KAFKA_BROKER_ADDRESS'],
      sasl: {
        mechanism: 'aws',
        authorizationIdentity: 'kafka-user',
        accessKeyId: 'ACCESS_KEY',
        secretAccessKey: 'ACCESS_SECRET',
        // sessionToken: '' // Optional
      },
    });

排查与解决方案 #

1. 优先确认网络连通性 #

Closed Connection最常见的原因是网络不通，先做以下验证：

• 确认MSK集群已开启公网访问：AWS MSK默认仅允许私有网络访问，需在集群配置的「联网」选项中手动开启公网访问，并绑定公网安全组。
• 验证端口可达性：用telnet或nc测试本地到MSK公网broker的端口：

  # SASL/PLAIN用9196，IAM用9198
  telnet b-1-public.xxxxxxxxxx.xxxxx.c3.kafka.xxxxxxx1.amazonaws.com 9196
  

  如果连接超时，说明网络层面有问题，检查：
  • MSK集群安全组的入站规则：允许本地IP（或0.0.0.0/0用于测试）访问对应端口（9196/9198）
  • 本地防火墙/代理：确保没有阻断对MSK端口的出站请求
• 确认broker地址正确：从AWS MSK控制台的「客户端信息」中复制公网broker地址，不要手动输入。

2. SASL/PLAIN认证的额外排查点 #

• 确认集群已启用SASL/PLAIN：在MSK集群的「安全配置」中，检查是否开启了SASL/PLAIN认证，且用户名密码已正确配置（需通过AWS Secrets Manager存储并关联到集群）。
• 用户名密码准确性：确保代码中的用户名和密码与Secrets Manager中存储的完全一致，注意大小写和特殊字符。

3. IAM认证的额外排查点 #

• 确认集群已启用IAM认证：在MSK集群的「安全配置」中开启IAM认证。
• 修正authorizationIdentity：此处需填写IAM用户的用户ID（而非用户名），可在IAM控制台用户详情页的「用户ID」字段获取（一串由字母数字组成的字符串）。
• 使用正确的端口：IAM认证对应的公网端口是9198，需更新brokers数组中的端口号。
• IAM权限验证：确保IAM用户拥有MSK集群的连接权限，可使用以下最小权限策略（替换YOUR_CLUSTER_ARN）：

  {
    "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Action": [
          "kafka-cluster:Connect",
          "kafka-cluster:DescribeCluster"
        ],
        "Resource": "YOUR_CLUSTER_ARN"
      }
    ]
  }
  

4. 代码优化与调试 #

• 升级kafkajs版本：确保使用最新版kafkajs，旧版本可能存在IAM认证兼容性问题，执行npm install kafkajs@latest更新。
• 开启调试日志：添加日志级别配置，获取更详细的错误信息：

  const { Kafka, logLevel } = require('kafkajs');
  
  this.kafka = new Kafka({
    clientId: 'user_service',
    brokers: ['b-1-public.xxxxxxxxxx.xxxxx.c3.kafka.xxxxxxx1.amazonaws.com:9198'],
    sasl: {
      mechanism: 'aws',
      authorizationIdentity: 'IAM_USER_ID',
      accessKeyId: 'ACCESS_KEY',
      secretAccessKey: 'ACCESS_SECRET',
    },
    logLevel: logLevel.DEBUG // 开启调试日志
  });
  

修正后的IAM认证代码示例 #

const { Kafka, logLevel } = require('kafkajs');

this.kafka = new Kafka({
  clientId: 'user_service',
  // 替换为实际的公网broker地址+9198端口
  brokers: ['b-1-public.xxxxxxxxxx.xxxxx.c3.kafka.xxxxxxx1.amazonaws.com:9198'],
  sasl: {
    mechanism: 'aws',
    // 替换为IAM用户的用户ID
    authorizationIdentity: 'AIDXXXXXXXXXXXXXX',
    accessKeyId: 'AKIAXXXXXXXXXXXXXX',
    secretAccessKey: 'your-secret-access-key',
  },
  logLevel: logLevel.DEBUG
});

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