Python与NestJS使用相同Kafka分区键无法路由至同一分区的问题

你的问题核心在于NestJS与Python服务对Kafka消息分区键的处理逻辑不一致：

• Python代码通过confluent_kafka.Producer.produce()的key参数，直接将partitioning_key设置为Kafka消息的原生分区键，Kafka会基于这个key的哈希值计算分区。
• NestJS代码中，client.emit(topic, { key: partitionKey, value: message })的写法，实际上是把包含key和value的对象作为Kafka消息的value内容发送，而非将partitionKey设置为Kafka消息的原生分区键。@nestjs/microservices默认的序列化器会把整个传入对象序列化为消息体，此时Kafka消息的原生key为null，分区计算会采用轮询或其他逻辑，自然和Python服务的分区结果不匹配。

要让两边的分区键逻辑对齐，需要在NestJS中正确设置Kafka消息的原生分区键，有两种可靠的实现方式：

方式1：直接使用底层kafka-js Producer发送 #

@nestjs/microservices的ClientKafka基于kafka-js封装，你可以直接获取底层Producer，按照kafka-js的标准方式发送消息，确保分区键被正确设置：

private client: ClientKafka;
...
async sendMessage(topic: string, message: any, partitionKey: string): Promise<void> {
  // 获取底层kafka-js Producer实例
  const producer = await this.client.getProducer();
  await producer.send({
    topic,
    messages: [
      {
        key: partitionKey, // 这里设置的是Kafka消息的原生分区键
        value: JSON.stringify(message), // 消息体序列化为JSON字符串
      },
    ],
  });
}

方式2：自定义序列化器适配消息结构 #

如果希望继续使用emit方法，可以自定义序列化器，让它识别你传入的{ key, value }结构，将key提取为Kafka消息的原生分区键：

1. 创建自定义序列化器类：

import { KafkaSerializer } from '@nestjs/microservices';

export class CustomKafkaSerializer extends KafkaSerializer {
  serialize(value: any): { key?: Buffer; value: Buffer } {
    // 识别包含key和value的结构
    if (typeof value === 'object' && 'key' in value && 'value' in value) {
      return {
        key: Buffer.from(value.key, 'utf-8'), // 将key转为Buffer（和Python的utf-8编码对齐）
        value: Buffer.from(JSON.stringify(value.value), 'utf-8'), // 序列化消息体
      };
    }
    // 兼容其他消息格式
    return super.serialize(value);
  }
}

2. 在ClientKafka配置中指定该序列化器：

@Module({
  providers: [
    {
      provide: 'KAFKA_CLIENT',
      useFactory: () => {
        return ClientKafka.register({
          client: {
            brokers: ['你的Kafka集群地址'],
          },
          serializer: new CustomKafkaSerializer(), // 应用自定义序列化器
        });
      },
    },
  ],
})
export class YourModule {}

• 确保两边的分区键编码一致：Python中用encode("utf-8")，NestJS中也用UTF-8编码（上述两种方式默认都是UTF-8）。
• 测试时可以打印Kafka消息的原生key：比如用kafka-console-consumer.sh加上--property print.key=true参数，确认NestJS发送的消息key和Python的一致。

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