如何实现Kafka（MSK）生产者就近写入同可用区的分区Leader？

一、让指定可用区拿下所有分区的Leader #

要把某个AZ的Broker设为所有分区的Leader，核心靠Kafka的优先副本选举和已配置的机架感知，具体步骤：

1. 调整主题的优先副本指向目标AZ

   • 先查看当前主题的分区副本分布：

     kafka-topics.sh --describe --topic <你的主题名> --bootstrap-server <MSK集群地址>
     

   • 输出里Replicas列表第一个就是优先副本。如果它不在目标AZ，就得重新分配副本，让每个分区的优先副本都属于目标AZ的Broker。比如目标AZ是us-east-1a，对应Broker ID是1、2、3，就把每个分区的Replicas列表第一个换成这些ID。
   • 编写副本分配JSON文件（比如reassignment.json），示例格式：

     {
       "version": 1,
       "partitions": [
         {"topic": "my-topic", "partition": 0, "replicas": [1,4,5]},
         {"topic": "my-topic", "partition": 1, "replicas": [2,4,5]},
         ...
       ]
     }
     

     第一个元素是目标AZ的Broker ID，后面跟着其他AZ的Broker。

2. 执行副本分配+优先副本选举

   • 先应用副本分配：

     kafka-reassign-partitions.sh --bootstrap-server <MSK集群地址> --reassignment-json-file reassignment.json --execute
     

   • 然后触发优先副本选举，让所有分区的Leader切换为目标AZ的Broker：

     kafka-preferred-replica-election.sh --bootstrap-server <MSK集群地址>
     

   • 验证结果：再次执行kafka-topics.sh --describe，所有分区的Leader列都应该是目标AZ的Broker ID。

3. 锁死Leader防止自动切换（可选）

   • 在MSK的Broker配置里，把leader.imbalance.check.interval.seconds设为较大值（比如86400，即一天），减少自动检查Leader平衡的频率；或者把leader.imbalance.per.broker.percentage设为0，直接关闭自动Leader平衡，避免Kafka自动将Leader切换到其他AZ。

二、只写当前可用区有Leader的分区 #

如果不想把所有Leader集中到一个AZ，只想让生产者只写入当前AZ内有Leader的分区，有两种实现方式：

1. 自定义生产者分区器

   • 自己实现分区器逻辑：先查询每个分区的Leader所在AZ（通过Kafka AdminClient获取Leader的Broker信息，再结合Broker的broker.rack配置即AZ标识），然后只选择Leader AZ与生产者Pod所在AZ一致的分区。
   • 伪代码思路（Java）：

     public class AzAwarePartitioner implements Partitioner {
         private AdminClient adminClient;
         private String currentAz;
     
         @Override
         public void configure(Map<String, ?> configs) {
             currentAz = System.getenv("POD_AZ"); // 从K8s环境变量获取Pod所在AZ
             adminClient = AdminClient.create(configs);
         }
     
         @Override
         public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
             List<PartitionInfo> allPartitions = cluster.partitionsForTopic(topic);
             List<Integer> eligiblePartitions = new ArrayList<>();
             for (PartitionInfo p : allPartitions) {
                 Node leaderNode = p.leader();
                 String leaderAz = cluster.nodeById(leaderNode.id()).rack();
                 if (currentAz.equals(leaderAz)) {
                     eligiblePartitions.add(p.partition());
                 }
             }
             // 从符合条件的分区中选择，比如按key哈希
             if (eligiblePartitions.isEmpty()) {
                 // 无符合条件分区时的降级逻辑，比如随机选一个
                 return Utils.toPositive(Utils.murmur2(keyBytes)) % allPartitions.size();
             }
             return eligiblePartitions.get(Utils.toPositive(Utils.murmur2(keyBytes)) % eligiblePartitions.size());
         }
     
         @Override
         public void close() {
             adminClient.close();
         }
     }
     

   • 在生产者配置中指定该分区器：partitioner.class=com.yourcompany.AzAwarePartitioner，同时通过K8s的downwardAPI给Pod注入自身所在AZ的环境变量（比如spec.containers[].env[].fieldRef.fieldPath=metadata.labels['topology.kubernetes.io/zone']）。

2. 主题分区按AZ分组创建

   • 创建主题时，将分区按AZ分组分配，比如3个分区的Leader放在AZ A，3个放在AZ B，3个放在AZ C。然后让K8s中部署在AZ A的Pod只写入AZ A的Leader分区，其他AZ同理。
   • 示例创建主题命令：

     kafka-topics.sh --create --topic my-topic --partitions 9 --replication-factor 3 --bootstrap-server <MSK地址> --replica-assignment "1,4,5:1,4,5:1,4,5:2,4,5:2,4,5:2,4,5:3,4,5:3,4,5:3,4,5"
     

     前3个分区的Leader是AZ A的Broker 1，中间3个是AZ B的Broker 2，最后3个是AZ C的Broker 3。后续给不同AZ的Pod配置对应的分区选择逻辑即可。

注意事项 #

• 集中所有Leader到单个AZ存在单点风险：如果该AZ故障，所有分区会自动切换Leader到其他AZ，故障恢复后需手动执行优先副本选举恢复原有配置。
• 自定义分区器会增加生产者复杂度，需处理Leader动态切换的场景（比如Leader突然切换到其他AZ时，分区器要能动态感知并调整）。
• 必须确保MSK每个Broker的broker.rack参数正确配置为所在AZ，否则机架感知和AZ判断逻辑会失效。

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