Kafkajs连接AWS MSK报超时错误但消息仍可写入，求优化方案

问题场景 #

使用Lambda通过KafkaJS向AWS MSK集群写入数据，消息能成功写入，但CloudWatch持续记录[Connection] Connection timeout错误日志，希望通过代码调整或配置优化消除这类日志。

生产者代码示例 #

const client = new Kafka({ 
    clientId: "client-id", 
    brokers: ["broker1:9092", "broker2:9092"],  // 示例broker地址
});

const producer = client.producer({
    idempotent: true
});

const record = {
    topic: "topic1",
    messages: [
        { value: JSON.stringify("message") }
    ]
};

await producer
    .connect()
    .then(async () => await producer.send(record))
    .then(async () => await producer.disconnect())
    .catch(err => throw new Error(JSON.stringify(err)));

错误日志示例 #

{
    "level": "ERROR",
    "timestamp": "2022-12-05T20:44:06.637Z",
    "logger": "kafkajs",
    "message": "[Connection] Connection timeout",
    "broker": "[some-broker]:9092",
    "clientId": "[some-client-id]"
}

解决方法 #

1. 调整Kafka客户端超时参数 #

延长连接和请求超时时间，适配Lambda冷启动或MSK集群的网络延迟，同时可降低日志级别过滤非关键信息：

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

const client = new Kafka({ 
    clientId: "client-id", 
    brokers: ["broker1:9092", "broker2:9092"],
    connectionTimeout: 30000, // 连接超时设为30秒（默认10秒）
    requestTimeout: 30000,     // 请求超时设为30秒
    logLevel: logLevel.WARN    // 只记录WARN及以上级别日志
});

2. 优化生产者实例复用 #

Lambda每次调用都创建、连接、销毁生产者会增加连接开销，建议将生产者初始化放在handler外部复用：

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

// 初始化放在handler外，复用实例
const client = new Kafka({ 
    clientId: "client-id", 
    brokers: ["broker1:9092", "broker2:9092"],
    connectionTimeout: 30000,
    requestTimeout: 30000
});
const producer = client.producer({
    idempotent: true,
    retry: {
        initialRetryTime: 100,
        retries: 5 // 增加重试次数，应对临时网络波动
    }
});
let isProducerConnected = false;

exports.handler = async (event) => {
    try {
        if (!isProducerConnected) {
            await producer.connect();
            isProducerConnected = true;
        }
        
        const record = {
            topic: "topic1",
            messages: [{ value: JSON.stringify("message") }]
        };
        await producer.send(record);
        
        return { statusCode: 200, body: "Message sent successfully" };
    } catch (err) {
        // 连接断开时重置状态，下次调用重新连接
        isProducerConnected = false;
        throw new Error(JSON.stringify(err));
    }
    // 不要每次调用都disconnect，保持连接复用
};

3. 检查MSK网络配置 #

• 确认Lambda所在VPC与MSK集群VPC的网络连通性，安全组需允许对应端口（明文9092、TLS9094）的双向流量。
• 若使用MSK私有链接，确保VPC端点配置正确，Lambda能访问到MSK的私有DNS。

4. 错误日志说明 #

部分情况下，这类超时是KafkaJS尝试连接备用broker时的临时错误，只要消息能成功发送，属于客户端重试机制的正常表现。若不想看到这类日志，可通过logLevel参数调整日志级别过滤非关键信息。

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