如何在NestJS Kafka客户端配置中指定分区数量？

问题场景 #

我正在使用NestJS原生Kafka客户端，当前生产者和消费者运行时会自动创建仅含1个分区的Kafka Topic，想通过客户端配置实现创建多分区Topic，相关代码如下：

main.ts #

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.connectMicroservice(
    {
      transport: Transport.KAFKA,
      options: {
        client: {
          clientId: 'app1',
          brokers: ['localhost:9092'],
          logLevel: logLevel.ERROR,
        },
        producer: {
          allowAutoTopicCreation: true,
          retry: {
            retries: 10,
            multiplier: 2,
          },
        },
        consumer: {
          groupId: 'app1',
        },
        run: {
          autoCommit: false,
        },
      },
    },
    {
      inheritAppConfig: true,
    },
  );
  await app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();

Controller代码 #

@MessagePattern('event1')
async event1Handler(@Payload() data: any, @Ctx() context: KafkaContext) {
  console.log(`receive ${data.arg} - part: ${context.getPartition()}`);
  await context.getConsumer().commitOffsets([
    {
      topic: context.getTopic(),
      partition: context.getPartition(),
      offset: context.getMessage().offset,
    },
  ]);
}

解决方案 #

NestJS的Kafka客户端基于kafkajs库，默认自动创建Topic时仅生成1个分区，可通过以下两种方式配置多分区：

方案一：通过生产者配置指定分区数 #

在producer配置项中添加createTopics参数，明确指定目标Topic的分区数和副本因子：

producer: {
  allowAutoTopicCreation: true,
  retry: {
    retries: 10,
    multiplier: 2,
  },
  createTopics: {
    topics: [
      {
        topic: 'event1', // 对应你的目标Topic名称
        numPartitions: 3, // 设置需要的分区数量
        replicationFactor: 1, // 单节点集群只能设为1，多节点可根据集群规模调整
      },
    ],
    waitForLeaders: true, // 等待Leader节点就绪后再返回
  },
},

当allowAutoTopicCreation为true时，发送消息到未创建的Topic时，会优先使用createTopics中的配置生成Topic；若未指定对应Topic的配置，则会使用Kafka集群默认的分区数。

方案二：预先通过Kafka命令创建Topic（生产环境推荐） #

生产环境中建议提前通过Kafka命令行工具创建Topic，这种方式更可控，避免客户端自动创建的配置不一致问题：

kafka-topics.sh --create --topic event1 --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

注意事项 #

• Topic创建完成后，分区数只能增加，无法减少；
• 副本因子的取值不能超过Kafka集群的Broker节点数量，否则会创建失败。

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