毒消息不是「失败」，是「每次都一样地失败」
普通失败和毒消息的区别只有一个判据：同样的输入再处理一次，结果会不会不同。
- 下游超时：这次失败，下次可能成功，是普通失败。
- JSON 少一个必填字段：跑一万次都是同一个解析异常，是毒消息。
- 会员记录还没同步过来：等几秒可能就有了，是普通失败。
memberId指向一个永远不存在的会员：是毒消息。
毒消息最麻烦的地方不是它自己失败，而是它会拖住同一个分区里排在它后面的所有消息。Kafka 的消费位移是分区级的连续值，处理不了第 2 条，就不能把位移提交到第 3 条；提交了就等于宣布第 2 条已经被处理，消息就丢了。
常见的四类毒消息：
| 类型 | 例子 | 特征 |
|---|---|---|
| 反序列化失败 | schema 变更后旧格式、字符编码错误、字段类型不匹配 | 在进入业务代码之前就抛异常 |
| 业务前置条件不可能满足 | 引用的会员/图书不存在、金额为负、状态不可迁移 | 业务规则确定性地拒绝 |
| 触发代码缺陷 | 空指针、数组越界、并发条件下的死锁 | 与数据内容强相关，换一条就好 |
| 资源型 | 单条消息触发超大内存分配、超长循环 | 不只是卡分区，可能拖垮进程 |
第四类已经超出「隔离消息」的范畴，需要限额与熔断；本篇讨论前三类。
一条毒消息如何卡住整个分区
消费者拉取一批记录后，应用逐条处理，处理成功才提交位移。假设分区里有 offsets 0 到 5，第 2 条是毒消息：
- 0、1 处理成功，提交位移到 2；
- 拉到 2，业务抛异常；
- 代码决定原地重试，不提交位移；
- 3、4、5 在日志里等着，但永远轮不到。
即使 max.poll.records 大于 1，后面的记录已经被拉进内存也没有用：位移提交是连续的，提交 3 就意味着 2 被跳过。除非显式 seek 跳过，否则框架只会提交到 2 之前。
这里有一个容易误读的细节：应用内的 position() 已经走到 3 了。 position 是「下一次拉取的位置」，记录被 poll 返回时它就前进了；committed 才是「已经确认处理完的位置」。毒消息被应用攥在手里重试时，两个值会分叉：
| 口径 | 停摆时的值 | 含义 | 用途 |
|---|---|---|---|
committed | 2 | 已确认处理完，重启从这里开始 | 监控 lag 的默认口径 |
position | 3 | 下一条要拉的位移 | 应用内观察，不能当进度 |
log-end | 6 | 分区末尾（又投了 5 条后是 11） | 计算积压 |
| CLI 的 LAG | 4（后为 9） | log-end - committed | 告警与容量 |
position 口径的 lag | 3（后为 8） | log-end - position | 会少算正在处理的那条 |
两个 lag 差 1，差值就是「已被拉取但未提交」的记录数。毒消息长期停摆时，这个差值会稳定存在，看板上的 lag 和代码里算出来的永远对不上。
实测：停摆前后
本机实测（Temurin 25.0.5+8 / kafka-clients 4.3.0 / apache/kafka:4.3.0 单节点 KRaft / 2026-10-29）：主题 library.loan.returned 1 分区，先投 6 条消息，offset 2 的 value 是 payload=corrupt，消费端稳定抛异常并原地重试。消费组 libraryd-poison-demo，手动提交，max.poll.records=1。
消费端输出（前两条正常，第三条开始停摆）：
[13:41:20.959] ok partition=0 offset=0 committed=1 log-end=6
[13:41:20.976] ok partition=0 offset=1 committed=2 log-end=6
[13:41:20.985] STALL attempt=1 partition=0 offset=2 position=3 committed=2 log-end=6 lag(committed)=4 value=returned loan=loan-2 payload=corrupt
[13:41:21.986] STALL attempt=2 partition=0 offset=2 position=3 committed=2 log-end=6 lag(committed)=4 value=returned loan=loan-2 payload=corrupt
停摆 8 秒时，从另一侧用命令行看消费组：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
libraryd-poison-demo library.loan.returned 0 2 6 4
这时再投 5 条健康消息（offset 6 到 10），再看：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
libraryd-poison-demo library.loan.returned 0 2 11 9
CURRENT-OFFSET 一直是 2，LAG 从 4 涨到 9。30 秒窗口内，同一条消息被重试了 27 次，位移一次都没动：
[13:41:46.114] STALL attempt=26 partition=0 offset=2 position=3 committed=2 log-end=6 lag(committed)=4 value=returned loan=loan-2 payload=corrupt
[13:41:47.122] STALL attempt=27 partition=0 offset=2 position=3 committed=2 log-end=6 lag(committed)=4 value=returned loan=loan-2 payload=corrupt
stall window over, record still unprocessed offset=2
这不是「消息处理慢」，是「这个分区从 offset 2 起完全没有进展」。 后面 9 条健康消息（原来的 3、4、5 和新投的 6 到 10）全部被堵住，业务上表现为退货处理整体延迟，而其他分区如果存在则不受影响——阻塞范围是分区，不是主题，更不是整个集群。
隔离手段对照
发现毒消息之后，有四条路可走。没有一条是免费的：
| 手段 | 操作 | 恢复速度 | 证据 | 主要风险 |
|---|---|---|---|---|
| 原地死等 | 不提交、持续重试 | 永不恢复 | 完整 | 分区永久停摆，重试风暴打下游 |
| 跳过位移 | seek 到下一 offset 并提交 | 秒级 | 丢失 | 消息永久消失，事后无法补 |
| 转隔离主题 | 写 quarantine，确认后提交位移 | 分钟级 | 完整 | 需要人工处理，隔离区也会堆积 |
| 进 DLT | 写 DLT，提交位移 | 秒级 | 完整 | 若配置了自动回灌，会形成循环 |
选择依据是三件事：
- 这条消息还有没有业务价值？ 有，就必须留证据，走隔离或 DLT。
- 能不能自动判断它不可恢复？ 反序列化失败可以；业务规则拒绝需要人确认。
- 等待人工处理的时间，分区能不能承受？ 不能承受就先跳过并留副本，能承受就停住等人。
最危险的是第二种和第四种的组合：先 seek 跳过、又把 DLT 配成自动回灌。消息在「跳过 → 回灌 → 再失败 → 再跳过」之间循环，看板上一切正常，实际上这条消息永远处理不了。
隔离实现：从已提交位移读，不是当前位置
隔离动作本身必须小心，因为它要同时做三件事：读出一条消息、写到隔离主题、提交位移。顺序错了就会丢消息。
static RecordHeader h(String key, Object value) {
return new RecordHeader(key,
String.valueOf(value).getBytes(StandardCharsets.UTF_8));
}
TopicPartition tp = new TopicPartition(topic, 0);
consumer.assign(List.of(tp));
long committed = committedOf(consumer, tp); // 从 committed 开始
consumer.seek(tp, committed); // 不是 position()
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
for (ConsumerRecord<String, String> r : records) {
if (!isPoison(r)) continue; // 只处理确认有毒的那条
ProducerRecord<String, String> q =
new ProducerRecord<>(QUARANTINE, r.key(), r.value());
q.headers()
.add(h("original-topic", r.topic()))
.add(h("original-partition", r.partition()))
.add(h("original-offset", r.offset()))
.add(h("quarantined-at", System.currentTimeMillis()))
.add(h("reason", "payload=corrupt, in-place retries exhausted"));
producer.send(q).get(); // 先确认隔离区写入
consumer.commitSync(Map.of(tp, new OffsetAndMetadata(r.offset() + 1)));
break; // 一次只处理一条
}
四个要点：
- 用
committed重新定位。 另一个进程的position在它自己的内存里，这里拿不到；committed是唯一跨进程可靠的位置。 - 先写隔离区再提交。 反过来的话，写入失败时消息已经被跳过。
- 一次只隔离一条。 批量隔离时如果中途失败，很难判断哪些已经写过；一条一条来，重复执行最多产生重复的隔离记录，不会漏。
- 隔离区用独立主题与独立 ACL。 能读隔离区的人应当能接触原始业务数据，权限要比普通消费者更严。
本机执行的输出：
quarantined key=loan-2 src=0@2 -> library.loan.returned.quarantine@0 committed=3
恢复：位移跳过毒消息，后面的排队消息一次放行
隔离之后，用原来的消费组继续消费。由于 committed 已经推进到 3，消费者从 offset 3 开始：
[13:42:02.987] drain ok partition=0 offset=3 key=loan-3
[13:42:02.998] drain ok partition=0 offset=4 key=loan-4
[13:42:03.002] drain ok partition=0 offset=5 key=loan-5
[13:42:03.006] drain ok partition=0 offset=6 key=loan-6
[13:42:03.011] drain ok partition=0 offset=7 key=loan-7
[13:42:03.019] drain ok partition=0 offset=8 key=loan-8
[13:42:03.029] drain ok partition=0 offset=9 key=loan-9
[13:42:03.033] drain ok partition=0 offset=10 key=loan-10
drained=8 committed=11 position=11 log-end=11 lag=0
8 条健康消息在 46 毫秒内处理完（单次运行，随调度波动，不作基准），LAG 归零：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
libraryd-poison-demo library.loan.returned 0 11 11 0
隔离区里的记录保留了全部定位信息：
QUARANTINE offset=0 key=loan-2 value=returned loan=loan-2 payload=corrupt
original-topic = library.loan.returned
original-partition = 0
original-offset = 2
quarantined-at = 1793252518799
reason = payload=corrupt, in-place retries exhausted in stall window
注意 original-offset=2 是这条消息在主主题里的位置，不是隔离区的位置。人工修复后回灌时，靠它和 key 找到原始上下文；quarantined-at 换算后是 2026-10-29 13:41:58.799（+08:00），用于计算隔离停留时长。
人工修复后回灌：三个幂等前提
回灌是「把修好的消息重新投回主主题」。它天然可能重复，也可能乱序，所以必须假定下游会收到两次、并且可能晚于同 key 的新消息。
前提一：业务唯一键，而不是位移
回灌会产生新的 offset，原来的 offset 只存在于 header 里。去重必须用业务键，例如 loanId + eventType + 事件时间。只在消费端记录「处理过的 offset」是无效的：回灌后 offset 变了，去重表认不出来。
create table consume_dedup (
dedup_key varchar(128) primary key,
event_id varchar(64) not null,
consumed_at timestamp not null
);
dedup_key 由业务字段拼出来，插入成功才继续处理，主键冲突就说明已经处理过。TTL 不能短于回灌窗口：如果去重记录 7 天过期，而人工修复花了 10 天，回灌就会重复生效。
前提二：状态迁移要条件更新，不能直接赋值
毒消息在处理到一半时失败，前面的写库可能已经成功。回灌再执行一次，如果代码是「查到记录 → 改状态 → 保存」，就会把已经前进的状态覆盖回去。
update loan
set status = 'RETURNED', version = version + 1
where loan_id = ?
and status = 'BORROWED'
and version = ?;
影响行数为 0 时不要当成功，要区分「已经被处理过」和「状态不合法」两种情况。前者可以安全忽略，后者要告警。
前提三：能处理旧事件后到
毒消息阻塞期间，同 key 的后续事件可能已经通过别的路径处理完。回灌把更早发生的事件插到后面，消费者如果按到达顺序覆盖状态，就会用旧值盖掉新值。
两个可用的办法：
- 事件带
occurredAt或单调版本号，消费端只接受不小于当前版本的事件； - 状态机只允许合法迁移，比如
RETURNED不能再回到BORROWED，旧事件到达时自然被拒绝。
回灌操作的最小纪律
- 先 dry-run。 打印将要回灌的条数、key 分布、修复前后的差异，不实际发送。
- 限速。 一次回灌几百条，不要瞬间把积压重新灌进主主题，否则等于自己制造一次流量尖峰。
- 带审计 header。
x-redrive-batch、x-fixed-by、x-fix-reason，出问题时能定位到人和批次。 - 回灌后核对。 对比业务表状态与事件条数，确认没有漏处理，也没有重复生效。
- 保留隔离区原记录。 不要回灌成功就删除，保留到业务确认之后。
判据：隔离机制是否合格
| 判据 | 通过条件 |
|---|---|
| 能发现 | 同一位移上重试超过阈值且位移不变时告警，而不是等人反馈业务延迟 |
| 能止损 | 从发现到分区恢复有明确时限，且不依赖重启整个消费者 |
| 能追溯 | 隔离记录带原主题、分区、位移、时间、原因、操作者 |
| 能恢复 | 隔离后 lag 能回到 0，后续消息处理延迟可测量 |
| 能回灌 | 回灌工具支持 dry-run、限速、审计，重复回灌不产生副作用 |
| 不循环 | 隔离/死信不会自动回灌回主主题，除非有人工确认环节 |
最后一条是红线。自动回灌看起来「省人力」，实际上是把一个确定性的失败变成了一个无限循环，而且循环的每一圈都在消耗主链路的容量。
回到 libraryd
libraryd 的退货事件里，payload=corrupt 属于数据损坏，重试没有意义。正确路径是：消费端重试 2 次仍失败 → 写 library.loan.returned.quarantine → 提交位移 → 分区恢复。值班人员从隔离区取出记录，补齐 memberId 和损坏字段，走 dry-run 确认 1 条，再回灌主主题。
回灌时 loanId + eventType 作为去重键，状态更新用条件更新，旧事件按 occurredAt 拒绝。这三件事缺一件，回灌就可能把一次数据损坏变成一次重复扣款。
阅读导航：上一节 6.2 重试主题与退避 ｜ 下一节 7.1 Outbox 模式
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。