栈：Kafka 3.5（3 broker，ISR=3）、Spring Kafka 3.1、K8s 1.29、Java 17、MySQL 8.0 结论先行：
consumer.wakeup()不是"关闭开关"，它只能让当前阻塞的那一次poll()抛WakeupException，而且这个"打断令牌"只消费一次、不排队；它对你"已经 poll 出来、还在内存里"的那批消息毫无办法。所以优雅停机的关键不是 wakeup，而是"安全 drain 掉 in-flight 批"，靠 in-flight 计数器轮询等待，而不是 sleep 一个拍脑袋的时长。
一、现象与背景：sleep 等 in-flight，发布窗口内还是丢
我们一条订单事件流：topic order-pay 32 分区，消费组 12 个 K8s 实例（Deployment，maxSurge=1 / maxUnavailable=0），日均约 4500 万条，max.poll.records=500，业务是「读 batch → 逐条写 MySQL → 整批 commit offset」。K8s 侧 terminationGracePeriodSeconds=30。
滚动发布时偶发丢消息，统计下来大约每 50 次发布丢 1～2 次，涉及 100～300 条。我们最初的 SIGTERM handler 是这么写的：
onSIGTERM() {
consumer.wakeup(); // 让 poll() 提前返回
sleep(5s); // 等 in-flight 跑完（不靠谱）
consumer.close(5s); // 超时直接抛
}
问题就出在 sleep(5s)：它是时间维度，而 in-flight 是数量维度，两者没有必然关系。慢 SQL、一次 800ms 的 Young GC、下游 DB 抖一下，5s 就等着 in-flight 跑不完了，剩余几百条既没 commit、也没落库，进程被 SIGKILL 直接蒸发。
二、根因：wakeup 管不到"内存里那批"，in-flight 才是真身
把一次消费拆成三态：
- 未 poll：还在 broker 里，丢了也能重放，安全；
- 已 poll 未处理：
poll()返回了 500 条，躺在 JVM 堆里——这才是真正的 in-flight； - 已处理未 commit：写库了但 offset 还没推。
wakeup() 的机制：它不关闭 consumer，而是让当前正在阻塞的那一个 poll()（或 close()）立刻抛 WakeupException。两个关键事实：
- 它只消费一次：
wakeup()是"一次性打断令牌"，不会排队累积。你 handler 里 wakeup 之后，如果还想再poll()一次拿最后一批数据，第二次 poll 会正常阻塞而不是抛异常——令牌已经用掉了。 - 它对内存里的 batch 无能为力：
wakeup()影响的是"下一次 poll 取不取新数据"，对"已经取到手、正在 for 循环处理"的那 500 条，它一点忙都帮不上。这批必须业务自己 drain。
所以优雅停机的正确问题不是"怎么让 poll 停下来"，而是"怎么确认内存里那批 in-flight 已经安全处理完（或安全回滚）再 close"。
三、具体做法：in-flight 计数器 + 状态机 + K8s preStop
核心思路：用 AtomicLong 记 in-flight 数量，进入处理 +count、结束 -count；SIGTERM 时置一个 DRAINING 标志停止接新批，然后轮询 in-flight 归零，配一个超时兜底。
class GracefulDrainer {
private final AtomicLong inFlight = new AtomicLong();
private final AtomicBoolean draining = new AtomicBoolean(false);
private final KafkaConsumer<String, Order> consumer;
// 业务处理入口：drain 的账记在这里
void process(ConsumerRecords<String, Order> records) {
inFlight.addAndGet(records.count()); // 进入 +1
try {
for (ConsumerRecord<String, Order> r : records) handle(r); // 写库
commitAndFlush(); // 先落库、再 commit（顺序见引用[1]）
} finally {
inFlight.addAndGet(-records.count()); // 离开 -1（含异常路径）
}
}
// SIGTERM handler：drain 而非 sleep
void onSigTerm(long drainTimeoutMs) {
draining.set(true); // 1. 让 poll 主循环检查此标志，不再取新批
consumer.wakeup(); // 2. 让当前阻塞的 poll 抛 WakeupException
long deadline = System.currentTimeMillis() + drainTimeoutMs;
while (inFlight.get() > 0 && System.currentTimeMillis() < deadline) {
Thread.sleep(50); // 轮询 in-flight，而非盲目睡固定时长
}
if (inFlight.get() > 0) log.warn("drain 超时, inFlight={}", inFlight.get());
consumer.close(Duration.ofSeconds(5));
}
}
poll 主循环里加一道闸：
while (running && !draining.get()) {
ConsumerRecords<String, Order> rec = consumer.poll(Duration.ofMillis(200));
if (rec.isEmpty()) continue;
drainer.process(rec);
}
K8s 侧配合两点：
terminationGracePeriodSeconds: 90 # 要 > drain 超时(20s) + close 余量
lifecycle:
preStop:
exec:
command: ["sh","-c","curl -s -XPOST localhost:8080/admin/drain"]
preStop 的钩子顺序在 K8s 里是「先跑 preStop → 再发 SIGTERM → 等 grace 到期后 SIGKILL」，所以把 drain 逻辑挂到 preStop 的 HTTP 端点、让 SIGTERM handler 只做"确认 + close"，能拿到更长的有效窗口。drain 超时我建议设 20s，terminationGracePeriodSeconds 留到 90s，给 GC 和 close 留余量。
四、踩过的坑与反直觉之处
- wakeup 是一次性的：想"先 wakeup 让 poll 返回，再 poll 拿最后一批"——不行，令牌已消费，第二次 poll 会正常阻塞。要拿最后一批，只能在 handler 里直接调一次
poll()，而不是靠 wakeup 触发。 - in-flight 的"真身"是内存里那批：只盯着
wakeup + close是隔靴搔痒，必须业务侧自己记账 drain。 - drain 完成前千万别 commit 未落库的批。引用江寒山-175 那篇攒批落库的结论很到位：「先 commit、再 flush」会把 Kafka 的 at-least-once 悄悄降级成 at-most-once[1]——丢得无声无息，offset 对账还是平的。
- 优雅停机不是银弹，只覆盖"正常 SIGTERM"这一条路径。引用江星河-885 的量化记录：
acknowledge()默认在「业务方法 return、事务尚未 commit」之间有个 5–22ms 的崩溃窗口，Full GC / SIGKILL 都能落进去[2]。OOMKilled、节点宕机这些不走 SIGTERM 的场景，drain 帮不了你，必须配合幂等兜底，否则优雅停机只是"把丢的概率调小"，不是"调成零"。
五、结论与可复用清单
- drain 用"数量"（in-flight 计数器）而非"时间"（sleep）来判定：
AtomicLong进出加减，SIGTERM 时轮询归零 + 超时兜底。 - wakeup 只打断当前一次阻塞、不排队：不要指望它帮你"取最后一批"，要取就显式再 poll 一次。
- 顺序铁律：先落库、再 commit；commit 前崩溃 = 可重放，commit 后崩溃 = 静默丢。
- K8s 侧：
preStop挂 drain 端点拿更长窗口，terminationGracePeriodSeconds必须 > drain 超时 + close 余量。 - 优雅停机 ≠ 零丢失：它只覆盖正常停机路径，OOMKILL/SIGKILL 仍会命中崩溃窗口，幂等是最后一道闸，二者缺一不可。
一句话总结：把"等 in-flight 跑完"从 sleep(5s) 换成"数清楚还剩多少条、数到零再走"，再给 OOM 这种不走 SIGTERM 的路径留一道幂等——优雅停机才算真正立住。
[1] 江寒山-175《Kafka 攒批落库：offset 提交与批量 flush 谁先谁后，决定了你是丢还是重》 [2] 江星河-885《Spring Kafka acknowledge() 与 @Transactional 的时序漏洞》