一、现象与背景
去年 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 指标，运维侧决定是否缩容。
5. key 基数不足时的 salt 策略（原文遗漏，补充）
当业务 key 天然只有几个值（如 5 个租户），且必须保证"同租户同分区"（下游按租户做流式处理），salt 是唯一的折中。但 salt 有一个核心矛盾：加 salt 打破了"同租户同分区"的约束，下游必须改为"同租户可能跨分区"。具体实现：
// salt 方案：tenantId + 序号取模，把 5 个租户的事件打散到所有 32 个分区
// 代价：同一租户的事件不再保证落在同一分区
// 适用：下游能容忍"同租户跨分区"（如按 tenantId 做幂等、按 tenantId 做聚合）
int salt = Math.abs(Objects.hash(valueBytes)) % numPartitions;
return (murmur3_32(tenantId.getBytes()) * 31 + salt) % numPartitions;
边界条件（踩过的坑）：
- 如果下游依赖"同租户同分区"做顺序保证（如状态机、事件溯源），salt 方案不可用，只能接受 5 个分区在干活 + 缩容消费组到 5 实例；
- salt 的取值必须确定性（不能每次 send 都不同），否则同一条消息重试时会路由到不同分区，破坏 exactly-once 语义（Kafka 事务的 PID 绑定分区）；
- 如果下游是 Flink 按 key 做 keyBy，salt 后 key 变成
tenantId:partition，Flink 的 state 会膨胀 32 倍——上线前必须和下游确认 state 容量。
四、踩过的坑与反直觉
坑一：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，分区均匀度反而变好了——但这不是我们想要的"确定性路由"。
坑四（新增）：salt 方案与 Kafka 事务的冲突。
我们最初用 ThreadLocalRandom 做 salt，结果发现：同一条消息在 producer.send() 超时重试时，salt 值变了，路由到不同分区，Kafka 事务的 lastOffset 校验失败，整个事务 abort。修复：salt 必须从消息体里确定性派生（如 hash(valueBytes) % numPartitions），保证同一条消息无论重试多少次都路由到同一分区。
五、结论与可复用清单
- 自定义 Partitioner 必须用
ClusterMetadata.partitionCountForTopic()拿真实分区数，不要硬编码； - key 基数 < 分区数时，任何哈希函数都会倾斜——这是数学事实。如果业务 key 只有 5 个值，要么接受 5 个分区在干活 + 缩容消费组，要么在 key 里加确定性 salt（注意与下游顺序保证和 Kafka 事务的兼容性）；
- 上线前必须跑"分区覆盖度"预检，断言命中分区数 ≥ 50%；
- 生产端加 per-partition 写入监控 + 空分区告警；
close()方法里不要做重量级资源释放，Kafka 客户端会在 rebalance 时频繁调用；- salt 必须确定性派生（从消息体 hash），不能每次 send 随机——否则重试时路由漂移，破坏事务语义；
- 留一个开放问题：当业务 key 天然只有几个值（比如 5 个租户），"按 key 路由到分区"这个假设本身是否成立？是不是应该放弃"同租户同分区"的约束，改用"同租户同 broker"来降低跨 rack 流量，同时把分区数压到和 key 基数同量级？欢迎大家拍砖。