结论：你"跨分区重投"的假设大概率不成立——标准 Kafka 的分区分配是 murmur2(key) % numPartitions，纯函数、与重试无关，同一 key 在分区数不变的前提下永远落同一分区。你真正要查的是同分区内的 at-least-once 重复 + version 守卫的并发竞态。下面拆成"定位 → 根治 → 8000 msg/s 推荐组合"三步。
一、先证伪"跨分区"假设（30 分钟可验证）
Kafka producer 的 partitioner 对非空 key 走 murmur2(key.hashCode()) & Integer.MAX_VALUE % partitionCount，这是确定性映射。重试（retries）只影响"同一条消息是否重发到同一 broker 的同一分区"，不会改变分区号。所以：
验证方法（在 consumer 侧加日志，跑 10 分钟压测）：
// consumer 侧，打印每条消息的 (key, partition, offset, version)
log.info("key={} partition={} offset={} version={}", key, record.partition(), record.offset(), version);
然后 grep 同一 userId 的消息，确认 partition 是否一致。如果 100% 同分区，"跨分区"假设直接排除。
真正会跨分区的场景只有两个：
- 消息 key 为 null → 走 StickyPartitioner → 轮询分区（检查你的 producer 代码，是否有某些分支没 set key）；
- 运维期间做了
kafka-topics --alter --partitions扩分区 → 同一 key 的 hash 结果 mod 新分区数后落点变了（检查变更日志）。
如果确认是同分区，问题就变成：同分区内 at-least-once 重复 + version 守卫的并发窗口。
二、真正的根因：version 守卫的 TOCTOU 竞态
你当前的 upsert 守卫大概是：
UPDATE t_user_profile SET payload = ?, version = ?
WHERE user_id = ? AND version < ?;
竞态场景（at-least-once 下同一分区内可重现）：
- 消息 A（version=10）和消息 B（version=11）先后到达 consumer；
- consumer-1 正在处理 A（version=10），尚未 commit offset；
- consumer-1 处理完 A、commit 前进程 crash；
- rebalance 后 consumer-2 从上一个 committed offset 重新消费 → 再次处理 A（version=10）和 B（version=11）；
- 但此时 consumer-2 处理 B（version=11）的 UPDATE 先执行成功；
- 接着 consumer-2 处理 A（version=10）的 UPDATE：
WHERE version < 10→ 当前 version 已是 11，应该被拦住。
看起来没问题？问题出在两个 consumer 实例并发消费同一分区（rebalance 窗口期，旧 consumer 还没 stop、新 consumer 已经 poll）：
- consumer-1 执行
UPDATE ... WHERE user_id=? AND version < 10（当前 version=9）→ 成功，version 变 10； - consumer-2 同时执行
UPDATE ... WHERE user_id=? AND version < 11（当前 version 还是 9，因为 consumer-1 的 commit 还没可见）→ 成功，version 变 11； - 但 consumer-1 随后又处理了 version=11 的消息 →
WHERE version < 11→ 当前 version=11，被拦住。
真正导致"旧值覆盖新值"的场景是：你的 SQL 里 version < ? 用的是消息里的 version，而不是数据库里的当前 version。如果两条消息 version 相同（重试产生的完全相同的消息），两条 UPDATE 都会命中 WHERE version < 10（当前 version=9），两条都成功，最后一条的值取决于执行顺序——如果后到的是"旧 payload"，就覆盖了。
三、根治方案（按改动面从小到大）
方案 A：(userId, version) 唯一键 + INSERT ON DUPLICATE（推荐，改动最小）
把"UPDATE 守卫"改成"幂等 INSERT"：
-- 表结构：UNIQUE KEY uk_user_version (user_id, version)
INSERT INTO t_user_profile (user_id, version, payload, updated_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
payload = IF(VALUES(version) > version, VALUES(payload), payload),
updated_at = IF(VALUES(version) > version, NOW(), updated_at);
关键变化：
- 唯一键
(user_id, version)保证同一版本只写一次，天然去重； ON DUPLICATE KEY UPDATE里的IF守卫保证"旧版本不会覆盖新版本"；- 即使两条完全相同的消息并发执行，唯一键冲突 → 第二条直接走 DUPLICATE 分支 →
IF(10 > 10)为 false → 不更新。
取舍：如果同 userId 的 version 是单调递增的（你的场景应该是），这个方案完美。如果 version 可能回退（比如上游 bug），需要额外加 version >= 0 的硬下限。
方案 B：Kafka Idempotent Producer（消除 producer 侧重复）
# producer 配置
enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5
retries=Integer.MAX_VALUE
作用：Kafka broker 侧对同一 producer 的同一 sequence number 去重，即使 producer 重试（网络抖动、broker 切换），broker 只接受一次。这消除了"producer 重投"导致的重复。
边界：
- 只保证同一 producer 实例的幂等，跨 producer 实例不保证；
- 不解决 consumer 侧重复（at-least-once 的 consumer 重复是语义本身）；
- 所以方案 B 必须配合方案 A，单独用 B 不够。
方案 C：Consumer 侧 offset commit 原子化（消除 consumer 侧重复）
// 正确做法：处理完一批消息后，一次性 commitSync
consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, byte[]> record : records) {
// 方案 A 的幂等 INSERT
executeUpsert(record);
}
consumer.commitSync(); // 同步提交，确保 crash 后从正确 offset 恢复
反模式（你当前可能在做）：
// 错误：先 commit 再处理，或异步 commit
consumer.commitAsync();
executeUpsert(record); // crash 在这里 → 消息丢失或重复
四、8000 msg/s 推荐组合
| 层 | 配置 | 作用 |
|---|---|---|
| Producer | enable.idempotence=true, acks=all, retries=MAX | 消除 producer 侧重复 |
| Consumer | max.poll.records=200, commitSync() | 减少 consumer 侧重复窗口 |
| DB | (user_id, version) 唯一键 + ON DUPLICATE KEY UPDATE | 最终兜底，即使前两层都漏了也不覆盖 |
| 监控 | performance_schema.events_statements_summary_by_digest 监控 t_user_profile 的 INSERT vs UPDATE 比例 | 如果 DUPLICATE 命中率 > 5%，说明上游重复率异常，需排查 |
量化预期：
- 方案 A 单独：0.3% 的"旧值覆盖"→ 0（唯一键硬保证）；
- 方案 A + B：producer 侧重复率从 ~0.3% 降到 <0.01%（broker 去重），DB 写入量减少 ~30%；
- 方案 A + B + C：端到端重复率 < 0.001%，且即使出现也是"重复写入相同值"，不会"旧值覆盖新值"。
五、一个容易忽略的坑
你的 12 个 consumer 实例消费 32 个分区，峰值 8000 msg/s → 每实例 ~667 msg/s。如果 max.poll.records 设得太大（比如 500），单次 poll 返回 500 条但处理不完（超时），Kafka 会触发 rebalance → 同一分区被两个 consumer 短暂消费 → 重复。建议：
max.poll.records=100
session.timeout.ms=45000
max.poll.interval.ms=300000
处理 100 条消息的时间应 < 30s（你的 upsert 是单行操作，100 条 < 1s，安全）。
引用
可复用清单：
- 唯一键 +
INSERT ON DUPLICATE是最简单的幂等方案； - Kafka Idempotent Producer 可消减 producer 侧重复；
commitSync()原子化 commit 保证 consumer 侧重复可控；- 监控
INSERTvsUPDATE比例，及时发现重复源； max.poll.records控制每次拉取量，减少 rebalance 频率。
最终建议：在 Kafka 与数据库层面同时做幂等，避免依赖上游的「不重复」承诺。