NestJS API网关向Kafka发消息时遇分区Leader选举错误求助

核心问题分析 #

错误提示There is no leader for this topic-partition as we are in the middle of a leadership election说明Kafka集群尚未完成主题分区的leader选举，或者客户端无法正确连接集群获取元数据。结合Bitnami Docker镜像特性，重点排查网络配置、主题自动创建开关和集群初始化状态三个方向。

具体解决方案 #

1. 修正Kafka监听地址配置 #

Bitnami Kafka需要区分容器内部和外部访问地址，否则外部客户端无法正确获取集群元数据。修改Docker Compose配置（或容器启动参数）：

services:
  zookeeper:
    image: bitnami/zookeeper:latest
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
    ports:
      - "2181:2181"

  kafka:
    image: bitnami/kafka:latest
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 容器内部监听9092，外部监听29092
      - KAFKA_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:29092
      # 声明外部可访问地址，对应宿主机端口
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
    ports:
      - "9092:9092"
      - "29092:29092"
    depends_on:
      - zookeeper

重启Kafka容器后，将NestJS的Kafka客户端broker地址改为localhost:29092。

2. 开启主题自动创建功能 #

若bank_create主题未创建且Kafka未开启自动创建，会导致分区无leader。在Kafka环境变量中添加：

- KAFKA_AUTO_CREATE_TOPICS_ENABLE="true"

也可手动进入Kafka容器创建主题：

docker exec -it <kafka-container-name> kafka-topics.sh --create --topic bank_create --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

3. 确认Kafka集群初始化完成 #

容器启动成功不代表Kafka已完成ZK同步和leader选举。查看Kafka容器日志，确认出现以下日志后再测试：

[KafkaServer id=1] started (kafka.server.KafkaServer)

若日志存在ZK连接失败信息，检查ZK容器状态及KAFKA_ZOOKEEPER_CONNECT配置（应为zookeeper:2181，对应Docker Compose服务名）。

4. 优化NestJS客户端配置 #

确保客户端配置正确，示例如下：

@Module({
  providers: [
    BankService,
    {
      provide: 'BANK KAFKA SERVICE',
      useFactory: () => {
        return new ClientKafka({
          client: {
            clientId: 'bank-client',
            brokers: ['localhost:29092'], // 对应外部监听端口
          },
          consumer: {
            groupId: 'bank-consumer-client',
          },
        });
      },
    },
  ],
})
export class BankModule {}

同时在onModuleInit中等待客户端初始化完成：

async onModuleInit() {
  await this._bankKafka.connect();
  this._bankKafka.subscribeToResponseOf('bank_create');
}

额外提示 #

关于KafkaJS分区器警告，可添加环境变量KAFKAJS_NO_PARTITIONER_WARNING=1关闭，或按提示指定LegacyPartitioner，不影响核心功能。

内容的提问来源于stack exchange，提问作者V. Benavides