Kafka元数据

这篇文章将重要介绍 Kafka 中的元数据和生产者拉取元数据的过程。生产者在发送消息时，需要获取消息所属的 Topic 在 Broker 节点上的分布情况，然后将消息发送到对应的节点，因此，在生产者发送消息时，需要有当前 Topic 的元数据信息。本篇文章在 Kafka 生产者拉取元数据代码的底子上，对元数据内部的数据结构和生产者拉取元数据的过程进行介绍，并且对生产者拉取元数据的代码增长了注释来方便明确。
1. 元数据

Metadata 是 Kafka 中的元数据类，负责对元数据进行管理，Metadata类内部通过大量synchronized方法来保证多线程下修改元数据的线程安全。以下是 Metadata 内部包含的重要数据结构：

1. public class Metadata implements Closeable {
2.     private final Logger log;
3.     // 重试间隔时间
4.     private final ExponentialBackoff refreshBackoff;
5.     // 元数据过期时间
6.     private final long metadataExpireMs;
7.     // 元数据版本号，每次元数据响应都会加 1
8.     private int updateVersion;  // bumped on every metadata response
9.     // 有新主题(Topic)的版本号，每次有新主题加 1
10.     private int requestVersion; // bumped on every new topic addition
11.     // 上一次更新时间
12.     private long lastRefreshMs;
13.     // 上一次成功更新时间
14.     private long lastSuccessfulRefreshMs;
15.     private long attempts;
16.     private KafkaException fatalException;
17.     private Set<String> invalidTopics;
18.     private Set<String> unauthorizedTopics;
19.     // **当前保存的元数据**
20.     private volatile MetadataSnapshot metadataSnapshot = MetadataSnapshot.empty();
21.     // 需要全部主题更新
22.     private boolean needFullUpdate;
23.     // 需要部分主题更新
24.     private boolean needPartialUpdate;
25.     private long equivalentResponseCount;
26.     private final ClusterResourceListeners clusterResourceListeners;
27.     private boolean isClosed;
28.     private final Map<TopicPartition, Integer> lastSeenLeaderEpochs;
29.     /** Addresses with which the metadata was originally bootstrapped. */
30.     private List<InetSocketAddress> bootstrapAddresses;
31. }

复制代码

MetadataSnapshot 是当前生存的元数据。

1. public class MetadataSnapshot {
2.     // 集群 cluset id
3.     private final String clusterId;
4.     // 节点
5.     private final Map<Integer, Node> nodes;
6.     // Topic
7.     private final Set<String> unauthorizedTopics;
8.     private final Set<String> invalidTopics;
9.     private final Set<String> internalTopics;
10.     // controller 节点
11.     private final Node controller;
12.     // Partition 和对应的 Partition 元数据
13.     private final Map<TopicPartition, PartitionMetadata> metadataByPartition;
14.     // Topic 信息
15.     private final Map<String, Uuid> topicIds;
16.     private final Map<Uuid, String> topicNames;
17.     // Cluster 集群信息
18.     private Cluster clusterInstance;
19. }

复制代码

Broker 节点

1. public class Node {
2.     // 节点 id
3.     private final int id;
4.     // 节点 id string
5.     private final String idString;
6.     // 节点地址
7.     private final String host;
8.     // 节点端口
9.     private final int port;
10.     // 节点机架
11.     private final String rack;
12. }

复制代码

Cluster 集群

1. public final class Cluster {
2.     private final boolean isBootstrapConfigured;
3.     // 节点
4.     private final List<Node> nodes;
5.     private final Set<String> unauthorizedTopics;
6.     private final Set<String> invalidTopics;
7.     private final Set<String> internalTopics;
8.     // Controller 节点
9.     private final Node controller;
10.     // Partition 和对应的 PartitionInfo
11.     private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
12.     // Topic 和对应的 PartitionInfo
13.     private final Map<String, List<PartitionInfo>> partitionsByTopic;
14.     private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
15.     // 节点和对应的 PartitionInfo
16.     private final Map<Integer, List<PartitionInfo>> partitionsByNode;
17.     // 节点 Id 和对应的服务器节点
18.     private final Map<Integer, Node> nodesById;
19.     private final ClusterResource clusterResource;
20.     // Topic 的 Id 和 Name
21.     private final Map<String, Uuid> topicIds;
22.     private final Map<Uuid, String> topicNames;
23. }

复制代码

1. public class TopicMetadata {
2.     // Topic id
3.     private final Uuid id;
4.     // Topic name
5.     private final String name;
6.     // 分区数量
7.     private final int numPartitions;
8.     // 分区和对应的机架
9.     private final Map<Integer, Set<String>> partitionRacks;
10. }

复制代码

1. public class PartitionMetadata {
2.     // 版本号
3.     private final int version;
4.     // Topic id
5.     private final Uuid topicId;
6. }

复制代码

1. public final class TopicPartition implements Serializable {
2.     // 分区编号
3.     private final int partition;
4.     // Topic
5.     private final String topic;
6. }

复制代码

1. public class PartitionInfo {
2.     // Topic
3.     private final String topic;
4.     // 分区编号
5.     private final int partition;
6.     // Leader 所在的服务器节点
7.     private final Node leader;
8.     // 副本所在的服务器节点
9.     private final Node[] replicas;
10.     // **ISR**: 可以与 leader 保持同步的副本
11.     private final Node[] inSyncReplicas;
12.     private final Node[] offlineReplicas;
13. }

复制代码

2. 发起元数据哀求

客户端调用 send() 来发送一条消息，send() 方法底层调用了 doSend() 方法。doSend() 方法中在发送消息前调用 waitOnMetadata() 哀求元数据。如下为 doSend() 方法中哀求元数据的部分。

1. private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
4.     long nowMs = time.milliseconds();
5.     ClusterAndWaitTime clusterAndWaitTime;
6.     try {
7.         // **请求元数据**
8.         clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);
9.     } catch (KafkaException e) {
10.         if (metadata.isClosed())
11.             throw new KafkaException("Producer closed while send in progress", e);
12.         throw e;
13.     }
16. }

复制代码

waitOnMetadata() 中起首尝试从已生存的元数据中获取 cluster 信息，如果失败，则会发起元数据哀求，调用 awaitUpdate() 阻塞当火线程等待拉取元数据。

1. private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long nowMs, long maxWaitMs) throws InterruptedException {
2.     // 尝试从元数据中获取 cluster 信息
3.     Cluster cluster = metadata.fetch();
5.     // 判断是否是无效主题
6.     if (cluster.invalidTopics().contains(topic))
7.         throw new InvalidTopicException(topic);
9.     // 将当前主题加入元数据
10.     metadata.add(topic, nowMs);
12.     // 尝试获取分区数量
13.     Integer partitionsCount = cluster.partitionCountForTopic(topic);
14.     // 如果存在则直接从当前保存的元数据中返回
15.     if (partitionsCount != null && (partition == null || partition < partitionsCount))
16.         return new ClusterAndWaitTime(cluster, 0);
18.     // 如果不存在则开始请求元数据
19.     long remainingWaitMs = maxWaitMs;
20.     long elapsed = 0;
21.     // Issue metadata requests until we have metadata for the topic and the requested partition,
22.     // or until maxWaitTimeMs is exceeded. This is necessary in case the metadata
23.     // is stale and the number of partitions for this topic has increased in the meantime.
24.     long nowNanos = time.nanoseconds();
25.     do {
26.         if (partition != null) {
27.             log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
28.         } else {
29.             log.trace("Requesting metadata update for topic {}.", topic);
30.         }
31.         metadata.add(topic, nowMs + elapsed);
32.         // 将 needFullUpdate 或 needPartialUpdate 设定为 true，返回 version
33.         int version = metadata.requestUpdateForTopic(topic);
34.         // 唤醒 sender 线程
35.         sender.wakeup();
36.         try {
37.             // **调用 awaitUpdate() 阻塞当前线程等待拉取元数据**
38.             metadata.awaitUpdate(version, remainingWaitMs);
39.         } catch (TimeoutException ex) {
40.             // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
41.             throw new TimeoutException(
42.                     String.format("Topic %s not present in metadata after %d ms.",
43.                             topic, maxWaitMs));
44.         }
46.         // 尝试从元数据中获取 cluster 信息
47.         cluster = metadata.fetch();
48.         elapsed = time.milliseconds() - nowMs;
49.         if (elapsed >= maxWaitMs) {
50.             throw new TimeoutException(partitionsCount == null ?
51.                     String.format("Topic %s not present in metadata after %d ms.",
52.                             topic, maxWaitMs) :
53.                     String.format("Partition %d of topic %s with partition count %d is not present in metadata after %d ms.",
54.                             partition, topic, partitionsCount, maxWaitMs));
55.         }
56.         metadata.maybeThrowExceptionForTopic(topic);
57.         remainingWaitMs = maxWaitMs - elapsed;
58.         // 尝试获取分区数量
59.         partitionsCount = cluster.partitionCountForTopic(topic);
60.     // 当 获取不到分区信息 或者 当前分区号大于获取到的分区数量，则继续循环
61.     } while (partitionsCount == null || (partition != null && partition >= partitionsCount));
63.     producerMetrics.recordMetadataWait(time.nanoseconds() - nowNanos);
65.     return new ClusterAndWaitTime(cluster, elapsed);
66. }

复制代码

awaitUpdate() 会阻塞主线程，等待 sender 线程返回元数据相应后 (updateVersion() > lastVersion) 才会继续执行。

1. public synchronized void awaitUpdate(final int lastVersion, final long timeoutMs) throws InterruptedException {
2.     long currentTimeMs = time.milliseconds();
3.     long deadlineMs = currentTimeMs + timeoutMs < 0 ? Long.MAX_VALUE : currentTimeMs + timeoutMs;
5.     // **阻塞等待拉取元数据**
6.     time.waitObject(this, () -> {
7.         // Throw fatal exceptions, if there are any. Recoverable topic errors will be handled by the caller.
8.         maybeThrowFatalException();
9.         return updateVersion() > lastVersion || isClosed();
10.     }, deadlineMs);
12.     if (isClosed())
13.         throw new KafkaException("Requested metadata update after close");
14. }

复制代码

3. sender 线程中的元数据哀求

sender 调用了 KafkaClient 接口中的 poll() 方法来进行网络哀求。NetworkClient 实现了 KafkaClient 接口，其内部调用了 metadataUpdater.maybeUpdate() 来哀求元数据，以下为 poll() 方法的具体实现。

1. @Override
2. public List<ClientResponse> poll(long timeout, long now) {
3.     ensureActive();
5.     if (!abortedSends.isEmpty()) {
6.         // If there are aborted sends because of unsupported version exceptions or disconnects,
7.         // handle them immediately without waiting for Selector#poll.
8.         List<ClientResponse> responses = new ArrayList<>();
9.         handleAbortedSends(responses);
10.         completeResponses(responses);
11.         return responses;
12.     }
14.     // **调用 maybeUpdate() 拉取元数据**
15.     long metadataTimeout = metadataUpdater.maybeUpdate(now);
16.     long telemetryTimeout = telemetrySender != null ? telemetrySender.maybeUpdate(now) : Integer.MAX_VALUE;
17.     try {
18.         this.selector.poll(Utils.min(timeout, metadataTimeout, telemetryTimeout, defaultRequestTimeoutMs));
19.     } catch (IOException e) {
20.         log.error("Unexpected error during I/O", e);
21.     }
23.     // process completed actions
24.     long updatedNow = this.time.milliseconds();
25.     List<ClientResponse> responses = new ArrayList<>();
26.     handleCompletedSends(responses, updatedNow);
27.     handleCompletedReceives(responses, updatedNow);
28.     handleDisconnections(responses, updatedNow);
29.     handleConnections();
30.     handleInitiateApiVersionRequests(updatedNow);
31.     handleTimedOutConnections(responses, updatedNow);
32.     handleTimedOutRequests(responses, updatedNow);
33.     completeResponses(responses);
35.     return responses;
36. }

复制代码

maybeUpdate() 内部起首根据更新标记和前次更新元数据的时间间隔等信息计算是否需要再次更新元数据，如果需要更新，则选取一个较为空闲的节点发起元数据哀求。

1. @Override
2. public long maybeUpdate(long now) {
3.     // 根据更新标记，过期时间和退避时间，计算出下一次要更新元数据的时间。
4.     long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
5.     // 是否有正在进行的元数据请求，如果有，将 waitForMetadataFetch 设定为 defaultRequestTimeoutMs(默认30s)
6.     long waitForMetadataFetch = hasFetchInProgress() ? defaultRequestTimeoutMs : 0;
8.     // 计算元数据超时时间
9.     long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
10.     if (metadataTimeout > 0) {
11.         return metadataTimeout;
12.     }
14.     // Beware that the behavior of this method and the computation of timeouts for poll() are
15.     // highly dependent on the behavior of leastLoadedNode.
16.     LeastLoadedNode leastLoadedNode = leastLoadedNode(now);
18.     // Rebootstrap if needed and configured.
19.     if (metadataRecoveryStrategy == MetadataRecoveryStrategy.REBOOTSTRAP
20.             && !leastLoadedNode.hasNodeAvailableOrConnectionReady()) {
21.         metadata.rebootstrap();
23.         leastLoadedNode = leastLoadedNode(now);
24.     }
26.     if (leastLoadedNode.node() == null) {
27.         log.debug("Give up sending metadata request since no node is available");
28.         return reconnectBackoffMs;
29.     }
31.     // **调用 private maybeUpdate(long now, Node node) 向选中的节点请求元数据**
32.     return maybeUpdate(now, leastLoadedNode.node());
33. }
36. // **判断是否需要更新元数据的条件**
37. // 计算下一次要更新元数据的时间
38. public synchronized long timeToNextUpdate(long nowMs) {
39.     // updateRequested() 检查是否标记了 needFullUpdate 或者 needPartialUpdate
40.     // 如果已标记，则将过期时间 timeToExpire 设定为 0。
41.     // 如果未标记，则根据 lastSuccessfulRefreshMs 和 metadataExpireMs 计算当前保存的元数据的过期时间 timeToExpire。
42.     long timeToExpire = updateRequested() ? 0 : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0);
44.     // timeToAllowUpdate() 根据退避时间计算允许下一次更新的时间
45.     // 最终返回 过期时间 和 允许下一次更新的时间 中较大的值。
46.     return Math.max(timeToExpire, timeToAllowUpdate(nowMs));
47. }
49. // 如果 needFullUpdate 或者 needPartialUpdate 为 true，则表示需要立即更新元数据
50. public synchronized boolean updateRequested() {
51.     return this.needFullUpdate || this.needPartialUpdate;
52. }
54. public synchronized long timeToAllowUpdate(long nowMs) {
55.     // 计算重试的退避时间，在退避时间内不进行重试。
56.     long backoffForAttempts = Math.max(this.lastRefreshMs +
57.             this.refreshBackoff.backoff(this.attempts > 0 ? this.attempts - 1 : 0) - nowMs, 0);
59.     // Periodic updates based on expiration resets the equivalent response count so exponential backoff is not used
60.     if (Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0) == 0) {
61.         this.equivalentResponseCount = 0;
62.     }
64.     // 计算重复返回相同响应的退避时间
65.     long backoffForEquivalentResponseCount = Math.max(this.lastRefreshMs +
66.             (this.equivalentResponseCount > 0 ? this.refreshBackoff.backoff(this.equivalentResponseCount - 1) : 0) - nowMs, 0);
68.     return Math.max(backoffForAttempts, backoffForEquivalentResponseCount);
69. }

复制代码

private maybeUpdate(long now, Node node) 方法将元数据哀求封装成 ClientRequest (与生产者发送消息类似)，并发送哀求。

1. private long maybeUpdate(long now, Node node) {
2.     String nodeConnectionId = node.idString();
4.     if (canSendRequest(nodeConnectionId, now)) {
5.         // **将元数据请求封装成 ClientRequest，并发送请求**
6.         Metadata.MetadataRequestAndVersion requestAndVersion = metadata.newMetadataRequestAndVersion(now);
7.         MetadataRequest.Builder metadataRequest = requestAndVersion.requestBuilder;
8.         log.debug("Sending metadata request {} to node {}", metadataRequest, node);
9.         sendInternalMetadataRequest(metadataRequest, nodeConnectionId, now);
10.         inProgress = new InProgressData(requestAndVersion.requestVersion, requestAndVersion.isPartialUpdate);
11.         return defaultRequestTimeoutMs;
12.     }
14.     // If there's any connection establishment underway, wait until it completes. This prevents
15.     // the client from unnecessarily connecting to additional nodes while a previous connection
16.     // attempt has not been completed.
17.     if (isAnyNodeConnecting()) {
18.         // Strictly the timeout we should return here is "connect timeout", but as we don't
19.         // have such application level configuration, using reconnect backoff instead.
20.         return reconnectBackoffMs;
21.     }
23.     if (connectionStates.canConnect(nodeConnectionId, now)) {
24.         // We don't have a connection to this node right now, make one
25.         log.debug("Initialize connection to node {} for sending metadata request", node);
26.         initiateConnect(node, now);
27.         return reconnectBackoffMs;
28.     }
30.     // connected, but can't send more OR connecting
31.     // In either case, we just need to wait for a network event to let us know the selected
32.     // connection might be usable again.
33.     return Long.MAX_VALUE;
34. }

复制代码

4. 处理元数据相应并更新元数据

在节点返回元数据相应之后，由 sender 线程 poll() 方法中的 handleCompletedReceives() 处理元数据。与生产者发送消息类似，起首将已经完成的哀求从 inFlightRequests 中移除，然后调用 handleSuccessfulResponse() 来处理元数据相应中的信息。

1. private void handleCompletedReceives(List<ClientResponse> responses, long now) {
2.     for (NetworkReceive receive : this.selector.completedReceives()) {
3.         String source = receive.source();
4.         // 将已经完成的请求从 inFlightRequests 中移除
5.         InFlightRequest req = inFlightRequests.completeNext(source);
7.         AbstractResponse response = parseResponse(receive.payload(), req.header);
8.         if (throttleTimeSensor != null)
9.             throttleTimeSensor.record(response.throttleTimeMs(), now);
11.         if (log.isDebugEnabled()) {
12.             log.debug("Received {} response from node {} for request with header {}: {}",
13.                 req.header.apiKey(), req.destination, req.header, response);
14.         }
16.         maybeThrottle(response, req.header.apiVersion(), req.destination, now);
17.         
18.         // **如果是元数据响应，则调用  handleSuccessfulResponse() 来处理元数据响应中的信息**
19.         if (req.isInternalRequest && response instanceof MetadataResponse)
20.             metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse) response);
21.         
22.     }
23. }

复制代码

handleSuccessfulResponse() 中更新当前生存的元数据。

1. @Override
2. public void handleSuccessfulResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
3.    
4.     // 更新元数据
5.     this.metadata.update(inProgress.requestVersion, response, inProgress.isPartialUpdate, now);
8. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。