KafkaProducer超时之谜：消息到底丢没丢？

摘要线上告警 send() 抛了 TimeoutException但 broker 端日志显示消息写进去了——到底丢没丢三个超时参数 max.block.ms、request.timeout.ms、delivery.timeout.ms 分别在哪一层起作用为什么超时重发会导致重复这篇从源码把 KafkaProducer 的超时体系讲透。关键词KafkaProducer、TimeoutException、超时机制、幂等性、request.timeout.ms、delivery.timeout.ms、消息不丢一个排查了三天的线上事故去年有个业务日志里每隔几分钟就蹦一条org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for order-events-0: 120001 ms has passed since batch creation业务同学来找我说是不是消息丢了下游对账差了 300 条。我第一反应不是查代码是问他你开幂等了吗他说没开。这就对了——TimeoutException 不等于消息丢了。很多人看到超时就以为消息没发出去但实际情况可能完全相反。那天排查了三天才搞清楚根因broker GC 停顿 40 秒客户端 request.timeout.ms 等不住主动断开重发。broker 那边其实已经把消息写盘了响应没来得及发回来。重发的消息又写了一次——所以下游对账不是少了是多了。这篇把 KafkaProducer 的超时体系彻底讲清楚。01 三个超时参数分别管哪一段先记住一句话三个超时管三段不同的时间。别搞混。max.block.ms默认 60 秒——管的是 send() 方法本身。消息还在客户端这头连 RecordAccumulator 都没进去。什么时候会等BufferPool 内存满了或者元数据还没拉到。send() 在这里最多等 60 秒等不到就直接抛异常// RecordAccumulator.append() 里的关键逻辑publicRecordAppendResultappend(Stringtopic,intpartition,...){// 申请 ByteBuffer最多等 max.block.msByteBufferbufferfree.allocate(size,maxTimeToBlockMs);if(buffernull){// 等不到内存直接抛——消息根本没进队列thrownewTimeoutException(Failed to allocate size bytes within the configured max blocking time maxTimeToBlockMs ms.);}}request.timeout.ms默认 30 秒——管的是单次请求。消息已经发出去了在等 broker 回响应。这个超时是 InFlightRequests 里维护的。Sender 线程每跑一圈就检查一遍// InFlightRequests.checkTimeout()publicListClientRequestcheckTimeout(longnow,longrequestTimeoutMs){ListClientRequestexpirednewArrayList();for(Map.EntryString,DequeClientRequestentry:requests.entrySet()){DequeClientRequestreqsentry.getValue();IteratorClientRequestitreqs.iterator();while(it.hasNext()){ClientRequestrequestit.next();longelapsednow-request.startTimeMs();if(elapsedrequestTimeoutMs){expired.add(request);it.remove();// 标记这个节点需要断开重连connectionStatus.disconnect(entry.getKey());}}}returnexpired;}注意这里——超时后不是直接判死刑而是标记节点断开然后这个 batch 会被放回 deque 队首下次重发。消息没丢只是这次请求超时了。delivery.timeout.ms默认 120 秒——这是总兜底。从消息创建那一刻开始算到最终成功或失败不能超过这个时间。这个检查在 RecordAccumulator 里每次 poll 待发送 batch 时过一遍// RecordAccumulator.maybeExpire()privatebooleanmaybeExpire(ProducerBatchbatch,longnow){longelapsednow-batch.createdTimeMs();if(elapseddeliveryTimeoutMs){// 总时间到了必须放弃——不管重试了多少次batch.done(-1,-1,newTimeoutException(Expiring batch.recordCount() record(s) for batch.topicPartition(): elapsed ms has passed since batch creation));returntrue;}returnfalse;}线上日志里搜到的那句 “ms has passed since batch creation”就是从这里打出来的。02 超时后消息到底去哪了这是最容易搞混的地方。两种超时两种结果request.timeout.ms 超时——消息大概率还在broker 可能已经处理了请求只是响应在路上堵车了。客户端等不住断开、重连、重发。重发的消息 broker 会再写一次。这就是重复消息的根源——不是 Kafka 不可靠是你没开幂等。delivery.timeout.ms 超时——消息真的没发出去总时间到了batch.done() 触发future 异常完成callback 里 exception 不为 null。内存立即释放。这种情况下消息确实没到达 broker。回到开头那个事故业务日志里的异常消息是 “120001 ms has passed since batch creation”——这是 delivery.timeout.ms 超时。但他下游对账是多了不是少了为什么因为在这 120 秒里前几次 request.timeout 超时已经重发过了。其中某一次 broker 其实收到并写盘了但响应没回来。最后 delivery 超时放弃客户端认为失败但 broker 那边数据已经在。一句话总结request.timeout 超时重发可能导致重复delivery.timeout 超时才是真失败。03 为什么开幂等就不会重复既然超时重发会导致重复Kafka 怎么解决的答案是幂等性——enable.idempotencetrue。核心思路给每个 Producer 分配一个 PIDProducer ID每个 partition 维护一个单调递增的 sequence number。broker 收到请求后检查 sequencePIDabc, partition0, sequence5 第一次请求sequence5 → 接受写入 网络超时重发 第二次请求sequence5 → 已经写过了直接返回成功broker 端记住最近的 sequence重复的请求直接返回成功不重复写盘。这就是幂等性解决重复问题的原理。但注意——幂等只解决单分区、单会话内的重复。如果你 Producer 重启了PID 变了老 sequence 就不认了。跨会话的重复要靠业务方去重比如用业务唯一键。04 线上怎么配这三个参数别用默认值就上线。根据场景调高吞吐核心链路比如订单事件流delivery.timeout.ms调到 3000005 分钟——给重试留足时间request.timeout.ms保持 30000——别太小大集群网络抖动很正常一定开enable.idempotencetrue——别省这个日志/监控类低价值消息delivery.timeout.ms调小30 秒——发不出去就丢别占着内存request.timeout.ms可以 10 秒——快速失败快速降级排查超时问题的正确姿势先看异常消息原文。“ms has passed since batch creation” delivery 超时查 broker GC 和 leader 切换。“request timed out” request 超时查网络延迟和连接状态。“Failed to allocate … bytes” max.block 超时查 buffer.memory 够不够。写在最后KafkaProducer 的超时设计不是一个参数是三层防护max.block.ms 保护业务线程不被阻塞request.timeout.ms 给单次请求设上限delivery.timeout.ms 保证消息最终有定论。理解了这三层下次看到 TimeoutException 就不会慌——先判断是哪一层的超时再决定是查网络、查 broker、还是查内存。