Kafka 生产者自定义 Partitioner 后 32 分区只有 5 个在干活：路由倾斜的排查与修复

一、现象与背景

去年 9 月，我们做租户隔离改造，把 event-bus topic（Kafka 3.5，KRaft 模式，3 broker，32 分区）的生产者从默认 round-robin 换成了自定义 Partitioner——按 tenantId 做一致性哈希，让同一租户的事件尽量落到同一分区，方便下游按租户维度做流式处理。

上线当晚就出了问题：

• 消费组 16 个实例，监控上 11 个实例 lag 始终为 0（根本没分到数据），另外 5 个实例 lag 从 0 飙到 47 万；
• 生产端 TPS 从改造前的 9.4 万 msg/s 掉到 1.6 万 msg/s，业务侧大量事件排队；
• broker 侧 kafka-server:type=replicator-metrics,name=MessagesInPerSec 显示 32 个分区里只有 5 个有写入，其余 27 个分区的 MessagesInPerSec 为 0。

影响面：租户事件流式处理延迟从 <1s 涨到 120s+，3 个租户的实时风控规则失效，持续约 2 小时 17 分钟才回滚。

二、原理与根因：key 基数太低 + 位运算取模的隐蔽陷阱

自定义 Partitioner 的核心代码（简化）：

public class TenantPartitioner implements Partitioner {
    private int numPartitions = 32; // 硬编码

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, ClusterMetadata metadata) {
        if (key == null) return 0;
        String tenantId = (String) key;
        int hash = murmur3_32(tenantId.getBytes());
        return hash & 0x1F;  // 0x1F = 31，取低 5 位 → 0~31
    }
}

乍看没问题：32 个分区，取低 5 位，0~31。但实际生产中 tenantId 只有 5 个值（T-1001、T-2001、T-3001、T-4001、T-5001）——SaaS 平台当时只签了 5 个大客户。

5 个 key 的 murmur3 低 5 位恰好落在 {3, 7, 14, 22, 29} 这 5 个分区上。

根因不是哈希函数不好，而是 key 基数（5）远低于分区数（32），任何哈希函数都救不了。

更隐蔽的是：代码里 numPartitions = 32 是硬编码的。如果运维把分区数从 32 扩到 64，hash & 0x1F 最大返回 31，永远不会路由到 32 以上的分区，扩容等于白扩，且没有任何报错。

三、具体做法

1. 修复 Partitioner：用 ClusterMetadata 拿真实分区数，换取模

@Override
public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, ClusterMetadata metadata) {
    if (key == null) {
        int n = Math.max(metadata.partitionCountForTopic(topic), 1);
        return Math.abs(ThreadLocalRandom.current().nextInt()) % n;
    }
    String tenantId = (String) key;
    int numPartitions = Math.max(metadata.partitionCountForTopic(topic), 1);
    int hash = murmur3_32(tenantId.getBytes());
    return (hash & 0x7FFFFFFF) % numPartitions;  // 全 32 位取模，不用位运算
}

2. 上线前加"分区覆盖度"预检

在 CI 里用生产环境真实的 5 个 tenantId 对 32 个分区做一遍路由，断言 命中分区数 ≥ 分区数的 50%（即 ≥16 个）。

# 快速验证：发 100 条到不同 tenantId，然后看各分区 offset
bin/kafka-console-producer.sh --bootstrap-server b1:9092 --topic event-bus \
  --property parse.key=true --property key.separator="|"
# 对比 32 个分区的 offset，如果 27 个是 0 就完蛋了
bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list b1:9092 --topic event-bus --time -1

3. 生产端加 per-partition 写入监控

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        METRICS.counter("producer.partition.write", "topic", record.topic(),
                        "partition", String.valueOf(metadata.partition())).increment();
    }
});

告警规则：producer.partition.write 按 partition 聚合，任意 10 分钟内 >50% 的分区写入量为 0 即 P1 告警。

4. 消费端兜底：空分区检测

16 个实例分 32 个分区，每个实例 2 个。如果只有 5 个分区有数据，11 个实例空转。应用层加"空分区回收"：某实例连续 3 个 poll 周期（约 45s）拉到的 records 全为空，上报 idle_partition 指标，运维侧决定是否缩容。

四、踩过的坑与反直觉

坑一：hash & 0x1F 和 hash % 32 在数学上等价，但工程上不等价。

等价的前提是分区数恰好是 2 的幂。Kafka 分区数不一定是 2 的幂（比如 24、40、48），用位运算取模会系统性地偏向低编号分区。我们这次 32 是 2 的幂所以没踩，但同事的另一个 topic 是 24 分区，用 hash & 0x17（23）路由，结果 8 个分区永远收不到消息。教训：永远用 % numPartitions，不要位运算。

坑二：ClusterMetadata 在启动瞬间可能不完整。

Kafka 客户端的 ClusterMetadata 是异步刷新的，启动瞬间 partitionCountForTopic(topic) 可能返回 0，直接 % 0 就 ArithmeticException。必须加 fallback 到保守默认值。

坑三：close() 方法里做了 cache.invalidateAll()，但 Kafka 客户端在 K8s 滚动发布时频繁 close/reopen。

我们的 Partitioner 里持有一个 LoadingCache<String, Integer>，close() 里清空 cache。高频 rebalance 下 cache 被反复重建，预热时间从 200ms 涨到 1.2s，期间所有路由 fallback 到 ThreadLocalRandom，分区均匀度反而变好了——但这不是我们想要的"确定性路由"。

五、结论与可复用清单

1. 自定义 Partitioner 必须用 ClusterMetadata.partitionCountForTopic() 拿真实分区数，不要硬编码；
2. key 基数 < 分区数时，任何哈希函数都会倾斜——这是数学事实。如果业务 key 只有 5 个值，要么接受 5 个分区在干活，要么在 key 里加 salt（如 tenantId + ":" + (seq % 32)）人为增加基数；
3. 上线前必须跑"分区覆盖度"预检，断言命中分区数 ≥ 50%；
4. 生产端加 per-partition 写入监控 + 空分区告警；
5. close() 方法里不要做重量级资源释放，Kafka 客户端会在 rebalance 时频繁调用；
6. 留一个开放问题：当业务 key 天然只有几个值（比如 5 个租户），"按 key 路由到分区"这个假设本身是否成立？是不是应该放弃"同租户同分区"的约束，改用"同租户同 broker"来降低跨 rack 流量，同时把分区数压到和 key 基数同量级？欢迎大家拍砖。