AWS MSK集群Node.js Kafka生产者报KafkaJSProtocolError求助

问题描述 #

我有一个AWS MSK集群，在EC2服务器上通过本地安装的Kafka客户端可以正常创建主题、生产及消费消息。但在Node.js应用中使用KafkaJS生产者时，执行npm start出现如下错误：

{"level":"ERROR","timestamp":"2023-07-29T06:18:34.532Z","logger":"kafkajs","message":"[Connection] Response SaslHandshake(key: 17, version: 1)","broker":"b-1.ashixxxxxx4.c3.kafka.ap-southeast-1.amazonaws.com:9094","clientId":"cab-allocation","error":"Request is not valid given the current SASL state","correlationId":1,"size":10}
{"level":"ERROR","timestamp":"2023-07-29T06:18:34.533Z","logger":"kafkajs","message":"[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Request is not valid given the current SASL state","retryCount":0,"retryTime":273}
KafkaJSProtocolError: Request is not valid given the current SASL state

我的kafkaproducer.js代码如下：

const { Kafka, AwsSasl } = require('kafkajs');// Define the Kafka client configuration
const BROKER_1 = process.env.KAFKA_BROKER_1 as string
const BROKER_2 = process.env.KAFKA_BROKER_2 as string
const AWS_REGION = process.env.AWS_DEFAULT_REGION as string
const ACCESS_KEY_ID = process.env.AWS_ACCESS_KEY_ID as string
const SECRET_ACCESS_KEY = process.env.AWS_SECRET_ACCESS_KEY as string

const kafka = new Kafka({
  clientId: 'cab-allocation',
  brokers: ["b-1.axxxxxxx3.kafka.ap-southeast-1.amazonaws.com:9094,b-2.asxxxxx74.c3.kafka.ap-southeast-1.amazonaws.com:9094"], // Replace with your AWS MSK broker endpoints
  ssl: true,
  sasl: {
    mechanism: 'aws',
    authenticationProvider: AwsSasl,
   aws: {
     region: "ap-southeast-1",
     //authorizationIdentity:"Geolah",       
     secretAccessKey: "q7SI3JyeOxxxxxyYpyMhY0ciAou6TDXWdyR6h",
     accessKeyId: "AKIAxxxxS25GD37NEJ",
   },
 },
});

const producer = kafka.producer();
producer.connect();

export async function sendKafkaMessage(topic: string, message: any) {
  const result = await producer.send({
    topic,
    messages: [
      { value: message }
    ]
  });
  console.log(result);
  return result;
}

producer.on('producer.connect', () => {
  console.log('Kafka producer connected');
});

注：我已确认使用了正确的AWS凭证和MSK broker地址。

解决方案 #

1. 修复Brokers配置错误 #

你的brokers字段配置是一个包含逗号分隔字符串的数组，这是错误的。KafkaJS要求brokers是一个字符串数组，每个元素对应一个broker地址：

brokers: [
  "b-1.axxxxxxx3.kafka.ap-southeast-1.amazonaws.com:9094",
  "b-2.asxxxxx74.c3.kafka.ap-southeast-1.amazonaws.com:9094"
],

2. 异步处理Producer连接 #

producer.connect()是异步操作，直接调用会导致后续sendKafkaMessage执行时生产者还未完成连接。建议改为：

const producer = kafka.producer();

// 用立即执行函数处理异步连接
(async () => {
  try {
    await producer.connect();
    console.log('Kafka producer connected');
  } catch (err) {
    console.error('Failed to connect producer:', err);
  }
})();

3. 确认AWS SASL配置的正确性 #

• 确保你的AWS凭证拥有kafka-cluster:Connect权限，且MSK集群的IAM策略允许该凭证访问
• 若在EC2上运行Node.js应用，建议使用IAM角色而非硬编码凭证，KafkaJS会自动从EC2元数据获取凭证，同时避免密钥泄露
• 检查region配置是否与MSK集群所在区域完全一致

4. 升级KafkaJS版本 #

旧版本的KafkaJS可能存在AWS SASL握手的兼容性问题，尝试升级到最新稳定版：

npm install kafkajs@latest

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