KafkaProducer 元数据管理send() 卡住时它在等什么摘要上一篇讲了 KafkaProducer 构造器不发任何网络请求。这篇拆waitOnMetadata真实源码——第一次 send() 为什么卡 60 秒ProducerMetadata怎么从只记 bootstrap 地址到知道每个分区 leader以及STABLE / NEED_UPDATE / IN_FLIGHT三态状态机。基于 kafka-clients 3.9。关键词Kafka、KafkaProducer、ProducerMetadata、Metadata、Cluster、waitOnMetadata、awaitUpdate、Sender、max.block.ms先回答那个 60 秒上一篇结尾留了个场景bootstrap.servers 地址写错new KafkaProducer不报错第一条消息 send 进去卡 60 秒抛TimeoutException(Topic my-topic not present in metadata after 60000 ms.)。这 60 秒里业务线程干了什么答案是它在一个 Condition 上等一个永远不会到来的 version 递增——因为 bootstrap 地址错了Sender 根本连不上任何节点MetadataResponse 永远回不来。整个客户端不是卡死是设计上就允许你等这么久。要讲明白这件事得先看清元数据在客户端里到底是哪几个对象、各自管什么。01 两个对象Metadata 管何时刷Cluster 管长什么样很多人以为 ProducerMetadata 就是个 Map其实不是。kafka-clients 里这层拆成了两个类职责分得很死Metadata是基类管什么时候该刷新——它持有一个Cluster cluster字段、一个单调递增的int version、一个boolean needUpdate、一个boolean requestInProgress、一个订阅 topic 的SetString topics。它本身不存节点和分区信息只存调度状态。ProducerMetadata extends Metadata加了时间戳字段lastRefreshMs、lastSuccessfulRefreshMs、lastPartialUpdateMs。定期刷新就靠这些时间戳和metadata.max.age.ms比大小。Cluster才是数据本体——不可变快照。它里面是MapInteger, Node nodesById和MapTopicPartition, PartitionInfo byTp。PartitionInfo里有leader、replicas、inSyncReplicas、offlineReplicas。关键设计Cluster 不可变。update 时不是改旧的是 new 一个新 Cluster 整个替换掉 Metadata 里的引用——读线程永远不会看到半更新状态。读路径和写路径是分开的业务线程调metadata.fetch()直接拿 Cluster 引用无锁Sender 线程拿到 MetadataResponse 后调metadata.update(...)持锁写新 Cluster、version、listeners.signalAll()。读不抢写写不阻塞读——这是 send() 高吞吐的关键之一。构造完的 Cluster 长什么样只有 bootstrap.servers 解析出来的那几个InetSocketAddress转成的 Node没有任何 topic、没有任何分区、leader 全是 null。这就是为什么第一次 send 必须阻塞。02 第一次 send()waitOnMetadata 的真实代码看 KafkaProducer.send() 的链路。表面是一行producer.send(record)实际走到doSend()第一个关键调用就是waitOnMetadata()。源码核心逻辑kafka-clients 3.9略有简化privateClusterAndWaitTimewaitOnMetadata(Stringtopic,Integerpartition,longnow,longremainingWaitMs){Clusterclustermetadata.fetch();IntegerpartitionsCountcluster.partitionCountForTopic(topic);// 已经知道分区数直接返回if(partitionsCount!null(partitionnull||partitionpartitionsCount)){returnnewClusterAndWaitTime(cluster,0);}// 不知道这个 topic需要刷元数据longnowMstime.milliseconds();longremainingMath.max(0,remainingWaitMs);longversionmetadata.requestUpdate();// ① 标记需要更新sender.wakeup();// ② 叫醒 Sender 线程try{metadata.awaitUpdate(version,remaining);// ③ 阻塞等新版本}catch(TimeoutExceptionex){thrownewTimeoutException(Topic topic not present in metadata after ...);}returnnewClusterAndWaitTime(metadata.fetch(),waitTime);}三个动作逐一看①metadata.requestUpdate()返回当前 version。这个调用不发网络请求只是把needUpdate置 true返回一个旧版本号 v。业务线程拿着 v 去等——只等比 v 新的版本。②sender.wakeup()。Sender 线程在NetworkClient.poll()里阻塞等网络事件wakeup 让它下一轮醒过来检查metadata.updateRequested()。这就是为什么业务线程和 Sender 线程之间不需要直接通信——一个置标记一个轮询。③metadata.awaitUpdate(version, remaining)。这是真正阻塞的地方publicsynchronizedvoidawaitUpdate(finallonglastVersion,longmaxWaitMs){longbegintime.milliseconds();longremainingWaitMsmaxWaitMs;while(versionlastVersion){if(time.milliseconds()-beginmaxWaitMs)thrownewTimeoutException(Failed to update metadata after ...);time.wait(remainingWaitMs);// Condition.awaitremainingWaitMsmaxWaitMs-(time.milliseconds()-begin);}}它是synchronized方法wait 在 this 的 Condition 上。多个业务线程同时进来全都阻塞在这把锁的条件队列上——但只触发一次 RPC。因为requestUpdate()里如果requestInProgresstrue就不重复标记只记一笔。Sender 那一侧呢它的runOnce()里有这么一段if(metadata.updateRequested()!hasInFlightFetch()){Nodenodeclient.leastLoadedNode(time.milliseconds());if(node!null){ListNodenodesfetchTargets...;NetworkClientUtils.maybeUpdateMetadata(now,...);}}leastLoadedNode()选连接数最少的节点——bootstrap 阶段就是连得上的那个不挑。MetadataRequest 发过去Broker 返回全集群元数据。回来后Sender.handleResponse()调metadata.update(newCluster)新建 Cluster、version、listeners.signalAll()——所有阻塞在awaitUpdate上的业务线程同时被唤醒各自重新读 Cluster这次leader ! null继续走RecordAccumulator.append()。超时分支也在这awaitUpdate里 while 循环检查到时间到抛 TimeoutException。等元数据和等缓冲共享同一个max.block.ms预算——不是先等元数据 60 秒再等缓冲 60 秒是一共就 60 秒。03 三条路径只是设标记决策只有一个很多教程把元数据刷新讲成三条独立机制其实它们最终都汇入同一个状态机三个状态STABLEneedUpdatefalse, requestInProgressfalse。Cluster 是新的业务线程直接读。NEED_UPDATEneedUpdatetrue。Sender 下一轮会发 MetadataRequest。IN_FLIGHT请求已发出等响应。这期间其他线程调requestUpdate()只更新标记不重复发请求。三条路径触发的都是从 STABLE 走到 NEED_UPDATE定时刷新。Sender.runOnce()里每次检查now - lastSuccessfulRefreshMs metadata.max.age.ms默认 3000005 分钟。到点主动发全量——解决 Broker 上下线、leader 迁移、新 topic 创建这些客户端没遇到的变化。被动失效。不等 5 分钟发送时发现leader null、收到LEADER_NOT_AVAILABLE/NOT_LEADER_OR_FOLLOWER、连接 Broker 失败、TopicAuthorizationException——任何一个都直接requestUpdate(topic)。连接失败还有指数退避reconnect.backoff.ms默认 50ms最大reconnect.backoff.max.ms默认 1s避免疯狂重试打挂集群。订阅集合变更。topics.add(topic)时自动触发一次全量更新。元数据请求只问订阅过的 topic不是全集群拉取——这是为什么客户端规模大了以后元数据请求量可控。为什么要这么设计因为发元数据请求是个重操作要新建 TCP、序列化全集群拓扑。如果业务线程一发现 leader 没了就发一个并发场景下就是请求风暴。现在的模型是谁都可以设标记只有 Sender 线程一个决策点发请求in-flight 时排队。04 配置和 JMX 怎么看和元数据相关的配置按排查优先级排max.block.ms60000send() 在元数据和缓冲上花的总时间预算。这就是上一篇那个 60 秒超时。metadata.max.age.ms300000元数据最大存活。拓扑稳定维持默认扩缩容频繁调到 30~60 秒。request.timeout.ms30000单次 RPC 超时。比 max.block.ms 小所以一次请求超时后客户端还会重试几次直到总预算用完。retry.backoff.ms100刷新失败后的退避。线上区分元数据问题还是缓冲问题看 JMX 三个指标kafka.producer:typeProducer-metrics,client-idxxx,namebuffer-available-bytes持续为 0 → 缓冲耗尽和元数据无关。...namemetadata-age持续走高 → 元数据一直刷不新查连通性或 topic 是否存在。...namerecord-send-rate为 0 但...namerecord-network-send-rate也低 → 消息进缓冲了但 Sender 发不出去多半是元数据没拿到。05 排查清单把前面的内容收成可执行的清单send() 卡 60 秒 TimeoutException先telnet bootstrap-host port看通不通、topic 在 Broker 上存不存在、metadata-age在不在涨。别上来就把 max.block.ms 调到 300000——那是让客户端多等 5 分钟根因没解决。首次 send 比后续慢几百毫秒到几秒正常第一次要走完整时序。如果超过 5 秒查网络到 bootstrap 的 RTT。偶发 NOT_LEADER_OR_FOLLOWERleader 在切换客户端自动刷新重试。频繁出现就查 Broker 是不是在滚动重启。元数据请求量异常查是不是 send 了大量不同 topic——每个新 topic 触发一次全量更新。JMX 指标对不上确认client.id显式配了否则多个 Producer 的指标混在一起。写在最后元数据这层最迷惑人的地方是它看起来不工作——new 完不连、send 了卡 60 秒、topic 不存在不报错。这些都不是 bug是懒加载加单飞请求的设计结果。理解了requestUpdate/awaitUpdate/version这三个动作就理解了 KafkaProducer 为什么是这种行为。