如何在Kafka Connector中添加自定义分区器（MongoDB源连接器场景）

核心思路 #

MongoDB源连接器本身没提供直接配置自定义分区器的入口，但可以通过Kafka Connect的原生机制来实现哈希分区的需求，下面是几种可行的方案：

方案1：利用默认分区器+指定消息键 #

Kafka默认的DefaultPartitioner会对消息键做哈希来分配分区，所以只要让连接器把你想用来哈希的字段（比如MongoDB文档的_id或者业务主键）设为Kafka消息的键就行：

• 给连接器加这些配置：

  mongo.fields.inkey=_id  # 替换成你要用来哈希的字段
  key.converter=org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable=false
  

  配置后，连接器会把指定字段作为消息键，默认分区器自动基于键的哈希值把消息分配到对应分区，这是最简单的实现方式。

方案2：自定义Producer分区器（进阶） #

如果默认哈希逻辑不够用，需要自己写哈希规则，可以直接给连接器指定自定义的Producer分区器：

1. 写一个继承自org.apache.kafka.clients.producer.internals.DefaultPartitioner的类，重写partition方法实现你的哈希逻辑（比如指定特定字段哈希、改用MD5/SHA哈希算法）
2. 把这个类打包成JAR，放到Kafka Connect的插件目录（就是plugin.path配置指向的文件夹）
3. 在连接器配置里加一行：

   producer.partitioner.class=com.yourteam.YourCustomHashPartitioner
   

   MongoDB源连接器会复用Kafka Producer的配置，所以这个参数会生效。

方案3：用SMT预处理消息键再哈希 #

如果需要先对字段做处理（比如提取嵌套字段、转换格式）再哈希，可以用Kafka Connect的SMT（Single Message Transform）来修改消息键：

• 配置示例：

  transforms=pickKey
  transforms.pickKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
  transforms.pickKey.field=user_id  # 提取user_id作为消息键
  

  这个SMT会把文档里的user_id字段抽出来当消息键，之后默认分区器就会基于这个键的哈希分配分区，适合需要对键做预处理的场景。

注意点 #

• 主题分区数一旦确定就别轻易改，不然哈希映射关系会变，导致消息分布混乱
• 自定义分区器要保证逻辑一致，同一键每次都要分到同一个分区，不然会出现重复消费或丢消息
• 测试的时候可以用kafka-console-consumer.sh --partition <分区号>来验证消息是不是分到了预期分区

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