Kafka Producer 背压第一道坎：buffer.memory 打满后 send() 阻塞与 delivery.timeout 计时起点的一次复盘

结论先行：producer 端 buffer.memory 是「应用侧第一个背压点」，它打满后 send() 不是抛异常，而是同步阻塞在 RecordAccumulator.awaitAvailableMemory 上，直到 max.block.ms 才抛 TimeoutException——而这条异常的计时起点，和 delivery.timeout.ms 的计时起点，都不是你直觉里的「我调 send() 的那一刻」。我们一次 4.2 万 msg/s 的事件流降级事故，根因就是这两个语义的偏差：应用以为自己在「异步发」，实际在「同步等」，超时窗口被前段等待吃掉了一半。

一、现象与背景：broker 全绿，producer 却开始抛 TimeoutException

场景：订单状态机事件流，Kafka 3.5.1（KRaft，3 broker，min.insync.replicas=2，ISR=3），topic order-event 16 分区，生产端 6 个 Java 实例（JDK 17，kafka-clients 3.5.1），峰值 4.2 万 msg/s，单条平均 1.6KB，acks=all、enable.idempotence=true。

去年 11 月，broker 一次滚动重启（一次 1 台、间隔 90s、单台 30s 内完成），期间 producer 端监控出现两类反常：

1. 6 个实例里 4 个在 90s 窗口内抛出 org.apache.kafka.common.errors.TimeoutException: Timeout expired while waiting to send，共 2.3 万次；
2. 诡异的是，broker 侧 RequestHandlerAvgIdlePercent 全程 ≥ 70%、UnderReplicatedPartitions=0、网络 RTT 抖动 < 2ms——broker 根本没忙；
3. 更反直觉：被抛 TimeoutException 的消息，端到端耗时（应用埋点）中位数只有 820ms，远低于我们设的 delivery.timeout.ms=30000——为什么 30s 都没到就超时了？

第 3 条是我们排查的入口，也是本文要讲的核心。

二、原理：send() 的两段式阻塞，以及 delivery.timeout 到底从哪一刻开始计时

先把 send(record, callback) 在 KafkaProducer.doSend 里做的事拆成三段：

1. 入队前：RecordAccumulator.append——这一步要先拿 buffer.memory 里的一段内存槽（size 为 batch.size 的槽位），再写入 record。如果剩余内存不足，append 会走 awaitAvailableMemory(timeout)：挂起当前调用线程，直到 max.block.ms（默认 60s）内没有内存释放，才抛 TimeoutException。
2. 入队后：等刷盘——batch 由 Sender 线程按 linger.ms / 批满 / flush() 三种路径触发发送（这一点社区已有讨论，见引用）。这一段的总预算是 delivery.timeout.ms（默认 120s），它从「record 进入 accumulator 的那一刻」开始计时，而不是从「send() 被调用」或「batch 被发出」开始。
3. max.block.ms 与 delivery.timeout.ms 是两个独立预算：前者约束「第 1 段等多久」，后者约束「整个 record 从入队到拿到 ack 或放弃的总时长」。Kafka 客户端在初始化时会强制 max.block.ms <= delivery.timeout.ms，但不要求 max.block.ms + 网络往返 <= delivery.timeout.ms——这是很多团队踩坑的根源。

把上面三条拼起来，事故链条就清楚了：

• broker 滚动重启时，leader 切换 + ISR 收敛要 3–8s，期间 producer 的 Sender 线程在等 broker 响应，in-flight 的 5 个 batch（见引用 #122）全部挂起；
• Sender 卡住 → RecordAccumulator 里已占用的 batch 无法释放 → buffer.memory 剩余槽位持续减少；
• 上游 4.2 万 msg/s 不停，新 record 的 append 拿不到槽位 → 业务线程开始同步阻塞在 awaitAvailableMemory；
• 90s 的 leader 切换窗口里，max.block.ms=60s 先于 delivery.timeout.ms=30s？——不对，我们当时 max.block.ms 用的是默认 60s，delivery.timeout.ms 被配成了 30s。Kafka 客户端在初始化时的实际约束是 max.block.ms 可以大于 delivery.timeout.ms（3.x 客户端只是告警，不强制），于是业务线程在 awaitAvailableMemory 里最多能等 60s；
• 真正抛出的 TimeoutException，是 awaitAvailableMemory(60s) 到期抛的——它不是 delivery.timeout 的超时，是 max.block.ms 的超时，日志里两者长得很像（都是 Timeout expired while waiting to send），但语义完全不同。

第 2 点解释了「为什么端到端 820ms 就抛了」：那条消息在 append 阶段就阻塞了 800ms 左右（等前面 5 个 in-flight 中某一个释放槽位），随后 batch 被发出并 ack，但业务线程在 send() 返回前一直阻塞着，埋点的「端到端」包含了这段同步等待。

三、具体做法：三处配置修正 + 一处上游改造

1. 把 max.block.ms 压到「业务能容忍的同步等待上限」

# 之前
delivery.timeout.ms=30000
max.block.ms=60000        # 默认值，被我们忽视了
# 之后
delivery.timeout.ms=30000
max.block.ms=5000         # 业务 SLO：单条 send 同步等待 ≤ 5s，超过就上游降级

修正后，broker 抖动窗口内，send() 最多阻塞 5s 就快速失败，业务侧有明确的降级分支（写本地 outbox、异步补发），不再出现 60s 级别的线程堆积。

2. buffer.memory 按「峰值积压 × 单条大小 × 安全系数」反推，而不是拍脑袋

我们的算式：

单实例峰值 = 42000 / 6 = 7000 msg/s
单条 1.6KB，1 秒积压 = 11.2 MB
leader 切换最坏 8s 积压 = 90 MB
安全系数 1.5 → 135 MB

原来默认 buffer.memory=33554432（32MB），只够单实例 2.9 秒的峰值——这是事故的直接诱因。改成 buffer.memory=144179200（137.5MB）后，8s 抖动窗口内不再打满。

3. linger.ms / batch.size 与 buffer.memory 的配比要联动

社区已有讨论（见引用 #519）：linger.ms 是 batch 的等待窗口，不是延迟旋钮。我们最终配置：

linger.ms=5
batch.size=32768          # 32KB，比默认 16KB 大一倍
buffer.memory=144179200   # ~137.5MB
compression.type=lz4      # 见下文

「少而大」的批次（32KB / batch，平均 20 条）比「多而小」的批次（16KB / batch，平均 10 条）在 broker 侧的 RequestHandlerAvgIdlePercent 上稳定在 75%+，比 16KB 时高约 4 个百分点。

4. 上游加「send 失败即降级」的分支

try {
    Future<RecordMetadata> f = producer.send(record);
    f.get(5, TimeUnit.SECONDS);   // 兜底：send 返回后 5s 内必须拿到 ack
} catch (TimeoutException | ExecutionException e) {
    outboxDao.insert(record);    // 写本地 outbox，异步补发
    metrics.increment("producer.send.degraded");
    return;
}

这一条不是 Kafka 配置问题，但没有它，前面三处配置都只是「让事故变慢」，不是「让事故可恢复」。

四、踩过的坑与反直觉之处

1. max.block.ms 和 delivery.timeout.ms 在 3.x 客户端里不强制前者 ≤ 后者，初始化时只有 log.warn，不是 log.error，生产环境没人看日志。我们当时 max.block.ms=60s > delivery.timeout.ms=30s，客户端没拦，直接上线。建议：所有 producer 配置评审时，把这两个值写进 checklist，人工核对。
2. TimeoutException: Timeout expired while waiting to send 这个异常信息，在「入队等内存」和「等待 broker ack」两种场景下长得几乎一样，日志里无法区分。排查时我们最初以为是 ack 超时，走了 2 天弯路，最后靠 Sender 线程的 JFR 栈（KafkaProducer.doSend → RecordAccumulator.append → awaitAvailableMemory）才定位到是 buffer 耗尽。建议：生产环境开 log.level=org.apache.kafka.clients.producer:DEBUG，或在 awaitAvailableMemory 前后加埋点，区分「等内存」和「等 ack」两类超时。
3. compression.type 默认是 producer，会跟随 broker 的 compression.type。我们 broker 默认 zstd，producer 端 CPU 在压缩上吃了 12% 单核——对 1.6KB 的小消息，zstd 的压缩比优势（比 lz4 多压 8%）远不够覆盖 CPU 成本。改成 lz4 后，producer CPU 从 42% 降到 31%，broker 侧磁盘写入量只多了 6%（因为 broker 端还是会按自己的 compression.type 重压缩落盘）。小消息场景下，lz4 是更稳的默认值。
4. max.in.flight=5（幂等钳制）+ acks=all + leader 切换，三件事叠加时，in-flight 的 5 个 batch 会「整批挂起」，不是「一个一个超时」。这意味着 broker 抖动的「有效时长」= 最慢那个 batch 的恢复时间，而不是平均值。我们观测到 5 个 batch 的恢复时间离散度在 1.2s–8.5s，buffer.memory 的容量必须按 8.5s 这个尾值算，不是按 3s 中位数算。

五、结论与可复用清单

结论：Kafka producer 的「异步」是有条件的异步——send() 在 buffer 有空间时是异步的，在 buffer 打满时是同步阻塞的。buffer.memory 不是「越大越好」，而是「按最坏抖动窗口反推的最小值」；max.block.ms 不是「默认 60s 就完事」，而是「业务能容忍的同步等待上限」。

可复用清单：

1. buffer.memory ≥ 单实例峰值 msg/s × 单条大小 × 最坏 leader 切换时长 × 1.5；
2. max.block.ms ≤ 业务 SLO 里「单条 send 同步等待上限」，且人工核对 ≤ delivery.timeout.ms；
3. compression.type 小消息场景用 lz4，大消息（>8KB）再考虑 zstd；
4. 生产环境开 org.apache.kafka.clients.producer:DEBUG 或加 awaitAvailableMemory 埋点，区分「等内存」和「等 ack」两类超时；
5. 上游必须有「send 失败 → 本地 outbox → 异步补发」的降级分支，不要指望 Kafka 客户端自己兜底；
6. 监控项：producer.buffer.expired_records（buffer 耗尽丢弃）、producer.request.timeout_ms（等 ack）、producer.record_error（入队失败），三者分开看，不要合成一个「producer 错误率」。

引用

• 批次刷盘的三种触发路径与 linger.ms 语义：见社区文章 #519「Kafka 生产者 linger.ms 不是延迟开关」。
• enable.idempotence=true 把 max.in.flight 钳到 5、以及 in-flight 的 5 个 batch 在 leader 切换时的整批挂起行为：见社区文章 #122「Kafka 生产者幂等为什么把 max.in.flight 锁死在 5」。