使用Kafka JS连接Kafka集群时TLS连接失败问题排查

核心问题分析 #

错误 Client network socket disconnected before secure TLS connection was established 表明TLS握手过程中断，核心原因是SSL证书配置不匹配或缺失关键凭证。结合你的代码和凭证信息，以下是具体排查与修复步骤：

具体排查与修复步骤 #

1. 替换CA证书来源：用Truststore而非Keystore #

Truststore的作用是存储客户端信任的服务端根证书，Keystore则存储客户端自身的身份证书/密钥。你当前代码注释了Truststore的CA提取逻辑，误用了Keystore中的CA，这是关键错误。

修改证书提取逻辑：

// 从Truststore提取服务端信任的CA证书
const { caroot: { ca: trustCa } } = truststore;
// 从Keystore提取客户端身份的密钥和证书（注意确认别名是否正确）
const { localhost: { key, cert } } = keystore;

2. 确认JKS文件中的条目别名 #

jks-js解析JKS后返回的对象键是证书/密钥的别名，你当前使用的localhost和caroot可能与实际JKS中的别名不符。建议先打印完整解析结果确认真实别名：

console.log('Keystore 条目:', Object.keys(keystore));
console.log('Truststore 条目:', Object.keys(truststore));

如果实际别名是client-cert和kafka-ca，则调整解构逻辑：

const { 'client-cert': { key, cert } } = keystore;
const { 'kafka-ca': { ca: trustCa } } = truststore;

3. 完善SSL参数配置 #

KafkaJS的ssl参数有细节需要注意：

• ca参数接受字符串数组，需用数组包裹提取的CA证书
• 如果Keystore中的密钥设置了单独密码（与Keystore密码不同），需添加passphrase参数
• 生产环境建议启用rejectUnauthorized，避免绕过证书验证

调整后的SSL配置：

const kafka = new Kafka({
  clientId: 'qa-topic',
  brokers: ['xxxxx-xxxxx-x.cloudclusters.net:xxxxx'], // 替换为真实HOST:PORT
  ssl: {
    rejectUnauthorized: true,
    ca: [trustCa],
    key: key,
    cert: cert,
    // 如果密钥有单独密码，取消注释并填写
    // passphrase: 'your-key-passphrase'
  },
})

4. 验证Broker地址与端口 #

确保brokers数组中的地址是凭证提供的host:port，不要直接使用IP（除非证书SAN字段包含该IP），否则会触发证书域名不匹配错误。

5. 修复代码中的其他潜在问题 #

• consumer变量未定义，需添加实例创建代码：

  const consumer = kafka.consumer({ groupId: 'qa-consumer-group' });
  

• user_uuiD和i.supplier_id是未定义变量，需替换为实际值或临时注释，避免连接成功后发送消息时报错。

修复后的完整代码片段 #

import * as dotenv from 'dotenv' 
import express from 'express'
import { Kafka } from 'kafkajs';
import { Partitioners } from 'kafkajs';
import jks from 'jks-js';
import fs from 'fs';

dotenv.config();
const app = express();

// 读取并解析JKS文件
const keystore = jks.toPem(
    fs.readFileSync('./kafka.keystore.jks'),
    process.env.KEYSTORE_PASSWORD // 建议用环境变量存储密码
);

const truststore = jks.toPem(
    fs.readFileSync('./kafka.truststore.jks'),
    process.env.TRUSTSTORE_PASSWORD
);

// 打印条目别名确认正确性
console.log('Keystore 条目:', Object.keys(keystore));
console.log('Truststore 条目:', Object.keys(truststore));

// 提取证书与密钥（根据实际别名调整）
const { localhost: { key, cert } } = keystore;
const { caroot: { ca: trustCa } } = truststore;

// Kafka实例配置
const kafka = new Kafka({
  clientId: 'qa-topic',
  brokers: ['xxxxx-xxxxx-x.cloudclusters.net:xxxxx'],
  ssl: {
    rejectUnauthorized: true,
    ca: [trustCa],
    key: key,
    cert: cert,
    // passphrase: process.env.KEY_PASSWORD // 密钥单独密码（如有）
  },
})

const producer = kafka.producer({ createPartitioner: Partitioners.DefaultPartitioner })
const consumer = kafka.consumer({ groupId: 'qa-consumer-group' });

// 生产者事件监听
producer.on('producer.connect', () => console.log('KafkaProvider: 已连接'));
producer.on('producer.disconnect', () => console.log('KafkaProvider: 已断开'));
producer.on('producer.network.request_timeout', (payload) => 
  console.log(`KafkaProvider: 请求超时 ${payload.clientId}`)
);

const run = async () => {
  try {
    // 连接生产者
    await producer.connect()
    console.log('生产者连接成功');

    // 发送测试消息
    await producer.send({
      topic: 'supplier-ratings',
      messages: [
        {
          value: Buffer.from(JSON.stringify({
            "event_name": "QA",
            "external_id": "test-user-uuid",
            "payload": {
              "supplier_id": "test-supplier-id",
              "assessment": {
                "performance": 7,
                "quality": 7,
                "communication": 7,
                "flexibility": 7,
                "cost": 7,
                "delivery": 6
              }
            },
            "metadata": {
              "user_uuid": "5a12cba8-f4b5-495b-80ea-d0dd5d4ee17e"
            }
          }))
        },
      ],
    })
    console.log('消息发送成功');

    // 连接消费者
    await consumer.connect()
    await consumer.subscribe({ topic: 'test-topic', fromBeginning: true })

    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        console.log({
          partition,
          offset: message.offset,
          value: message.value.toString(),
        })
      },
    })
  } catch (error) {
    console.error('Kafka操作失败:', error);
    // 异常时断开连接
    await producer.disconnect();
    await consumer.disconnect();
  }
}

// 启动Express服务
const port = process.env.PORT || 5000;
app.listen(port, () => {
  console.log(`服务监听端口 ${port}`);
  run().catch(console.error);
});

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