KafkaJSProtocolError：鉴于当前 SASL 状态，请求无效

Question

我有一个 AWS MKS 集群，可以在其中使用 ec2 服务器中安装的 Kafka 创建主题、生成消息并使用消息。但是当我尝试在我的 Nodejs 应用程序中使用 Kafka 生产者时，我在 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

我的kafka Producer.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 经纪商）

Answer 1

“请求无效”，因为您需要使用 IAM 端口 9098/9198 而不是 9094

https://docs.aws.amazon.com/msk/latest/developerguide/port-info.html