结论：你的"跨分区重投"假设大概率不成立，真正凶手是「消费侧并发 + version 不是严格单调」。 Kafka producer 只要 key 不变、partitioner 不变，同一个 key 在任何重试/故障切换下都只会落到同一个分区——这是 partitioner 的契约，不是 at-least-once 的副作用。你 0.3% 的"旧 version 回滚"几乎可以肯定来自下面两条之一：
一、先把"跨分区重投"这个假设证伪
Kafka producer 的分区选择逻辑：partition = (hash(key) & (partitionCount - 1))，与 broker 存活状态、retries、acks 都无关。broker 故障切换时：
- 旧 leader 上的消息已 ack=all 落盘 → 重投只会发到新 leader 的同一个分区，offset 会前进（产生重复，但分区不变）；
- 旧 leader 上未 ack 的消息 → 客户端重试，仍然用同一个 key → 仍然 hash 到同一个分区。
所以"重投到不同分区"在标准 producer 下不可能发生。你可以先做个自检：在 consumer 端打日志 (userId, version, partition, offset)，连续抓 1 小时，如果同一个 userId 的 version 序列在不同 partition 出现 → 那才是真跨分区（这种只可能是上游有第二个 producer 用了不同的 key 派生逻辑，比如一个用 userId、一个用 userId+shardSuffix）。先把这个查掉，能省你一半排查时间。
二、真正的根因：消费侧并发 + version 守卫不严格
你的消费组是 12 实例，同一 partition 在任意时刻只被 1 个实例消费，这是 Kafka 的硬保证。所以"同 partition 并发写同一 userId"不成立。那 0.3% 的旧值后到，只能是这两种情况之一：
情况 A：version 不是全局单调（最常见）
如果 version 是上游应用自己生成的（自增 / 时间戳 / 雪花），只要有两个上游实例并发写同一 userId（比如两个服务都写 profile），version 就会出现"1001 → 1005 → 1003"这种乱序。你 WHERE version > 当前version 的守卫就拦不住"1003 在 1005 之后到"。
判据：拉 1 小时日志，按 userId 聚合，看 max(version) - min(version) 与 count(*) 的比例，以及是否存在 version 序列非单调 的 userId。如果 0.3% 的 userId 命中"非单调"，根因就是 A。
修复：让 version 变成事件的全局单调序，两条路：
- 上游用同一个发号器（雪花 / 号段）发 version，保证同 userId 严格递增；
- 或者放弃 version 守卫，改用 (userId, version) 唯一键 + INSERT IGNORE / ON DUPLICATE KEY UPDATE 仅当 version 更大：
-- 表结构：UNIQUE KEY uk_user_ver (user_id, version)
-- 消费端写入（替代你现在的 UPDATE WHERE）
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);
这个写法的好处：重复消息（同 version）天然幂等（唯一键拦截），乱序消息（旧 version 后到）天然被丢弃（IF 判 false），不需要 version 全局单调——只要同 userId 的 version 集合里取最大那个最终胜出就行。8000 msg/s 下，唯一键 upsert 在 InnoDB 是单行写，QPS 完全不是瓶颈（实测单实例 3-5 万行/s 起步）。
情况 B：消费端 ack 时机不对
如果你的 consumer 是"处理完 → commit offset"，但处理是异步批量落库（先入内存队列，攒批写 MySQL），那 offset commit 时 DB 还没写完。broker 故障 → 重投 → 同一批消息被消费两次，第二次可能先于第一次的 DB 写入完成，于是旧值覆盖新值。
判据：看你的 consumer 是不是 poll → 入内存队列 → 异步写 DB → commit offset 这种结构。如果是，根因就是 B。
修复：要么改成"同步写 DB 成功后再 commit"（吞吐会降，但 8000 msg/s 完全扛得住）；要么保留异步，但用 (userId, version) 唯一键 + 上面的 IF 守卫兜底，让"重复 + 乱序"都幂等。
三、8000 msg/s 下的推荐组合（可直接抄）
- Producer：
acks=all、retries=Integer.MAX_VALUE、max.in.flight.requests.per.connection=5、enable.idempotence=true（Kafka 0.11+ 内置幂等，在 broker 端按 (producerId, sequence) 去重，连 at-least-once 的重复都帮你消掉一大部分）。这是你缺的关键一环——你只开了 acks=all，没开 idempotence，等于让 broker 端去重白白浪费。 - Consumer：
max.poll.interval.ms调大到 300s（避免处理慢触发 rebalance），同步写 DB 后再 commit；或者异步写但配(userId, version)唯一键 + IF 守卫（推荐后者，吞吐更好）。 - DB：
UNIQUE KEY (user_id, version)，写入用上面的ON DUPLICATE KEY UPDATE ... IF(VALUES(version) > version, ...)模板。 - 监控：加一条
SELECT user_id, COUNT(*) FROM t_user_profile WHERE user_id IN (最近 1 小时写入的 userId) GROUP BY user_id HAVING COUNT(*) > 期望值，如果持续为 0，说明幂等闭环成立。
四、关于"at-least-once 的边界条件"
at-least-once 只保证"不丢"，不保证"不重"、不保证"跨消息有序"。所以：
- 单 k