毒消息与 DLQ 怎么设计

难度与前置

难度：进阶
前置：排过消费失败死循环或 Lag 飙升；知道单条坏消息能卡死分区。
读完能做什么：设计重试 + DLQ 隔离毒消息，保留可观测与回放，并用命令/对账验收。

适用场景

反序列化失败、Schema 不兼容、必填字段缺失导致单条反复抛错
业务校验失败（金额为负、未知枚举）不想堵后续合法消息
需要审计「坏消息」并支持人工/工具回放

不适用：

下游短暂故障（应重试/退避，不应直接进 DLQ）
全链路 Exactly-Once 事务边界问题（见 EOS），与毒消息分层处理
流量型 Lag（扩容/倾斜），不是单条 Poison

真实场景（一条）

点击流某分区一条畸形 JSON 导致 Flink 源算子持续失败重启，Lag 百万级。引入：可重试错误最多 3 次 → 进 DLQ Topic（保留原 topic/partition/offset/headers）→ 主链路跳过并提交位点；DLQ 告警 + 日批回放工具。主链路 5 分钟恢复，坏消息可追溯。

怎么做（步骤清单）

1. 先分类错误

类型	例子	策略
瞬时	下游 503、超时	有限重试 + 退避，不进 DLQ
毒消息	反序列化失败、硬校验失败	有限重试后 DLQ + 跳过
系统性	权限、依赖全挂	停消费/熔断，别把海量灌 DLQ

text

DLQ 不是垃圾桶：只收「这条消息本身无法在当前版本处理」的记录

2. DLQ Topic 设计

text

命名：{origin}.dlq 或 dlq.{origin}
分区：可与源相同或按 hash(key)；务必带追溯字段
保留：≥ 源 Topic 或更长（审计需要）
ACL：生产限消费组服务账号；读开放给数据/oncall

3. 信封（Envelope）建议字段

json

{
  "original_topic": "click_evt",
  "original_partition": 3,
  "original_offset": 987654321,
  "original_key": "...",
  "original_timestamp": 1720000000000,
  "error_type": "DESERIALIZE",
  "error_message": "Unexpected character...",
  "retry_count": 3,
  "consumer_group": "flink-click-etl",
  "first_fail_ts": 1720000001000,
  "payload_base64": "..."
}

Headers 同步打：dlq_reason、dlq_src_offset，方便不拆 body 过滤。

4. 消费侧伪流程

text

poll → deserialize
  fail → retry_count++ → 未超限：seek/不提交并退避
         超限：send DLQ → commit 源 offset（跳过）→ metric++
validate business
  fail → 同上（可与反序列化共用 DLQ，用 error_type 区分）
process → commit（与 EOS/手动提交模型对齐）

Flink：侧输出（side output）进 DLQ sink；主输出继续；注意 Checkpoint 与「跳过」语义一致。

5. 重试与幂等

重试必须有上限与抖动退避
进 DLQ 后源位点必须前进，否则永远卡死
回放 DLQ 时写主链路要幂等（见数仓幂等 / 实时去重）

6. 告警与治理

指标	阈值示例
dlq_messages_per_min	> 10 告警
dlq_ratio	> 0.1% 告警
连续毒消息同 partition	触发 Schema 回滚检查

日批：Top error_type、样例 payload、是否关联 Schema 演进事故。

7. 回放工具

bash

# 示意：从 DLQ 过滤后写回修复 Topic 或直接重处理
kafka-console-consumer --bootstrap-server $BS \
  --topic click_evt.dlq --from-beginning | \
  jq 'select(.error_type=="DESERIALIZE")' > /tmp/dlq_sample.json
# 修复 schema / 补丁作业后，专用 replay job 读 DLQ 写主链路

常见坑（至少牢记这些）

无限重试不跳过
一条毒消息卡死分区，Lag 表象像「下游慢」。
瞬时错误也进 DLQ
下游抖一下，海量「假毒」淹没真问题。
DLQ 不带原 offset
无法回溯、无法证明已跳过哪条，审计失败。
跳过却不提交位点
重启后重复毒消息 / 重复打 DLQ。
DLQ 与主 Topic 同生命周期乱删
合规与回放窗口被破坏。
多消费者组共用一个 DLQ 不打 group 标签
分不清谁产生的毒消息。

如何验收（可量化）

[ ] 注入 1 条畸形消息：主链路继续，Lag 不卡死
[ ] DLQ 中可查到该消息的 topic/partition/offset
[ ] 指标：dlq_count +1；主处理成功率恢复
[ ] 回放演练：修复后 DLQ 消息可重处理且幂等
[ ] 告警在注入后按阈值触发

bash

# 注入与观察（演练环境）
echo 'not-a-json' | kafka-console-producer --broker-list $BS --topic click_evt
# 观察消费组 Lag 应继续前进
kafka-consumer-groups --bootstrap-server $BS --describe --group flink-click-etl
# DLQ 应有新消息
kafka-console-consumer --bootstrap-server $BS --topic click_evt.dlq --timeout-ms 10000

sql

-- 若 DLQ 入湖
SELECT error_type, COUNT(*) AS cnt
FROM ods_click_evt_dlq_di
WHERE dt='${biz_date}'
GROUP BY error_type
ORDER BY cnt DESC;

text

验收窗口：
- 主作业 checkpoint 持续成功
- poison 注入后 1 分钟内 dlq_ratio 可见
- 主业务指标无长时间断崖

毒消息与 DLQ 怎么设计

适用场景

真实场景（一条）

怎么做（步骤清单）

1. 先分类错误

2. DLQ Topic 设计

3. 信封（Envelope）建议字段

4. 消费侧伪流程

5. 重试与幂等

6. 告警与治理

7. 回放工具

常见坑（至少牢记这些）

如何验收（可量化）

相关面试题

附：与 Schema 门禁联动

毒消息与 DLQ 怎么设计 ​

适用场景 ​

真实场景（一条） ​

怎么做（步骤清单） ​

1. 先分类错误 ​

2. DLQ Topic 设计 ​

3. 信封（Envelope）建议字段 ​

4. 消费侧伪流程 ​

5. 重试与幂等 ​

6. 告警与治理 ​

7. 回放工具 ​

常见坑（至少牢记这些） ​

如何验收（可量化） ​

相关面试题 ​

附：与 Schema 门禁联动 ​

毒消息与 DLQ 怎么设计

适用场景

真实场景（一条）

怎么做（步骤清单）

1. 先分类错误

2. DLQ Topic 设计

3. 信封（Envelope）建议字段

4. 消费侧伪流程

5. 重试与幂等

6. 告警与治理

7. 回放工具

常见坑（至少牢记这些）

如何验收（可量化）

相关面试题

附：与 Schema 门禁联动