聊聊 Kafka:Producer Metadata 读取与更新机制
一、前言
我们上一篇说了 聊聊 Kafka:Producer 源码解析,这一篇我们来说下 Producer Metadata 的读取与更新机制。上一篇从宏观上介绍了 Producer 的宏观模型,其中通过 waitOnMetadata() 方法获取 topic 的 metadata 信息这一块东西很多,所以单独拎一篇出来讲。
二、Metadata
2.1 什么是 Metadata
Metadata 是指 Kafka 集群的元数据,包含了 Kafka 集群的各种信息,直接看源码便可知:
public class Metadata implements Closeable {
private final Logger log;
// retry.backoff.ms: 默认值为100ms,它用来设定两次重试之间的时间间隔,避免无效的频繁重试。
private final long refreshBackoffMs;
// metadata.max.age.ms: 默认值为300000,如果在这个时间内元数据没有更新的话会被强制更新。
private final long metadataExpireMs;
// 更新版本号,每更新成功1次,version自增1,主要是用于判断metadata是否更新
private int updateVersion;
// 请求版本号,每发送一次请求,version自增1
private int requestVersion;
// 上一次更新的时间(包含更新失败)
private long lastRefreshMs;
// 上一次更新成功的时间
private long lastSuccessfulRefreshMs;
private KafkaException fatalException;
// 非法的topics
private Set<String> invalidTopics;
// 未认证的topics
private Set<String> unauthorizedTopics;
// 元数据信息的Cache缓存
private MetadataCache cache = MetadataCache.empty();
private boolean needFullUpdate;
private boolean needPartialUpdate;
// 会收到metadata updates的Listener列表
private final ClusterResourceListeners clusterResourceListeners;
private boolean isClosed;
// 存储Partition最近一次的leaderEpoch
private final Map<TopicPartition, Integer> lastSeenLeaderEpochs;
}
MetadataCache:Kafka 集群中关于 node、topic 和 partition 的信息。(是只读的)
public class MetadataCache {
private final String clusterId;
private final Map<Integer, Node> nodes;
private final Set<String> unauthorizedTopics;
private final Set<String> invalidTopics;
private final Set<String> internalTopics;
private final Node controller;
private final Map<TopicPartition, PartitionMetadata> metadataByPartition;
private Cluster clusterInstance;
}
关于 topic 的详细信息(leader 所在节点、replica 所在节点、isr 列表)都是在 Cluster 实例中保存的。
// 保存了Kafka集群中部分nodes、topics和partitions的信息
public final class Cluster {
private final boolean isBootstrapConfigured;
// node 列表
private final List<Node> nodes;
// 未认证的topics
private final Set<String> unauthorizedTopics;
// 非法的topics
private final Set<String> invalidTopics;
// kafka内置的topics
private final Set<String> internalTopics;
private final Node controller;
// partition对应的信息,如:leader所在节点、所有的副本、ISR中的副本、offline的副本
private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
// topic和partition信息的对应关系
private final Map<String, List<PartitionInfo>> partitionsByTopic;
// topic和可用partition(leader不为null)的对应关系
private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
// node和partition信息的对应关系
private final Map<Integer, List<PartitionInfo>> partitionsByNode;
// 节点id与节点的对应关系
private final Map<Integer, Node> nodesById;
// 集群信息,里面只有一个clusterId
private final ClusterResource clusterResource;
}
// topic-partition: 包含 topic、partition、leader、replicas、isr
public class PartitionInfo {
private final String topic;
private final int partition;
private final Node leader;
private final Node[] replicas;
private final Node[] inSyncReplicas;
private final Node[] offlineReplicas;
}
看源码不难理解 Metadata 的主要数据结构,我们大概总结下包含哪些信息:
集群中有哪些节点;
集群中有哪些 topic,这些 topic 有哪些 partition;
每个 partition 的 leader 副本分配在哪个节点上,follower 副本分配在哪些节点上;
每个 partition 的 AR 有哪些副本,ISR 有哪些副本;
2.2 Metadata 的应用场景
Metadata 在 Kafka 中非常重要,很多场景中都需要从 Metadata 中获取数据或更新数据,例如:
KafkaProducer 发送一条消息到指定的 topic 中,需要知道分区的数量,要发送的目标分区,目标分区的 leader,leader 所在的节点地址等,这些信息都要从 Metadata 中获取。
当 Kafka 集群中发生了 leader 选举,节点中 partition 或副本发生了变化等,这些场景都需要更新Metadata 中的数据。
三、Producer 的 Metadata 更新流程
Producer 在调用 doSend() 方法时,第一步就是通过 waitOnMetadata 方法获取该 topic 的 metadata 信息。
总结一下以上代码:
首先会从缓存中获取 cluster 信息,并从中获取 partition 信息,如果可以取到则返回当前的 cluster 信息,如果不含有所需要的 partition 信息时就会更新 metadata;
更新 metadata 的操作会在一个 do ….while 循环中进行,直到 metadata 中含有所需 partition 的信息,该循环中主要做了以下事情:
调用 metadata.requestUpdateForTopic() 方法来获取 updateVersion,即上一次更新成功时的 version,并将 needUpdate 设为 true,强制更新;
调用 sender.wakeup() 方法来唤醒 Sender 线程,Sender 线程中又会唤醒 NetworkClient 线程,在 NetworkClient 中会对 UpdateMetadataRequest 请求进行操作,待会下面会详细介绍;
调用 metadata.awaitUpdate(version, remainingWaitMs) 方法来等待 metadata 的更新,通过比较当前的 updateVersion 与步骤 1 中获取的 updateVersion 来判断是否更新成功;
3.1 org.apache.kafka.clients.NetworkClient#poll
上面提到调用 sender.wakeup() 方法来唤醒 Sender 线程,Sender 线程中又会唤醒 NetworkClient 线程,在 NetworkClient 中会对 UpdateMetadataRequest 请求进行操作。在 NetworkClient 中真正处理请求的是 NetworkClient.poll() 方法,接下来让我们通过分析源码来看下 NetworkClient 是如何处理请求的。
3.2 org.apache.kafka.clients.NetworkClient.DefaultMetadataUpdater#maybeUpdate(long)
我们来看下 metadata 是如何更新的
这里你可能会问,老周啊,最小负载节点是啥呀?
别急,我们来看下面这张图,你就理解了。
LeastLoadedNode 指 Kafka 集群中所有 Node 中负载最小的那一个 Node,它是由每个 Node 在 InFlightRequests 中还未确定的请求数决定的,未确定的请求越少则负载越小。如上图所示,Node1 即为 LeastLoadedNode。
3.3 org.apache.kafka.clients.Metadata#updateRequested
下次更新元数据信息的时间:
当前 metadata 信息即将到期的时间
即 timeToExpire 和 距离允许更新 metadata 信息的时间
即 timeToAllowUpdate 中的最大值。timeToExpire:needUpdate 为 true,表示强制更新,此时该值为 0;否则的话,就按照定时更新时间,即元数据信息过期时间(默认是 300000 ms 即 5 分钟)进行周期性更新。
timeToAllowUpdate:默认就是 refreshBackoffMs 的默认值,即 100 ms。
3.4 org.apache.kafka.clients.NetworkClient.DefaultMetadataUpdater#maybeUpdate(long, org.apache.kafka.common.Node)
我们继续跟一下 maybeUpdate 方法:
因此,每次 producer 请求更新 metadata 时,会有以下几种情况:
通道已经 ready,node 可以发送请求,那么就直接发送请求。
如果该 node 正在建立连接,则直接返回。
如果该 node 还没建立连接,则向 broker 初始化连接。
而 KafkaProducer 线程一直是阻塞在两个 while 循环中的,直到 metadata 更新:
sender 线程第一次调用 poll,初始化与 node 的连接。
sender 线程第二次调用 poll,发送 metadata 请求。
sender 线程第三次调用 poll,获取 metadataResponse,并更新 metadata。
3.5 接收 Server 端的响应,更新 Metadata 信息
handleCompletedReceives 是如何处理任何已完成的接收响应,如下:
之后进一步调用 handleSuccessfulResponse。
四、总结
Metadata 会在下面两种情况下进行更新:
强制更新:调用 Metadata.requestUpdate() 将 needFullUpdate 置为 true 来强制更新。
周期性更新:通过 Metadata 的 lastSuccessfulRefreshMs 和 metadataExpireMs 来实现,一般情况下,默认周期时间就是 metadataExpireMs,5 分钟时长。
在 NetworkClient 的 poll() 方法调用时,会去检查两种更新机制,只要达到一种,就会触发更新操作。
Metadata 的强制更新会在以下几种情况下进行:
initConnect 方法调用时,初始化连接;
poll() 方法中对 handleDisconnections() 方法调用来处理连接断开的情况,这时会触发强制更新;
poll() 方法中对 handleTimedOutRequests() 来处理请求超时时;
发送消息时,如果无法找到 partition 的 leader;
处理 Producer 响应(handleProduceResponse),如果返回关于 Metadata 过期的异常,比如:没有 topic-partition 的相关 meta 或者 client 没有权限获取其 metadata。
强制更新主要是用于处理各种异常情况。
好了,Producer Metadata 读取与更新机制就说到这,我们下一期再见。
欢迎大家关注我的公众号【老周聊架构】,Java后端主流技术栈的原理、源码分析、架构以及各种互联网高并发、高性能、高可用的解决方案。