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;  // bumped on every metadata response
    //请求版本号，每发送一次请求，version自增1
    private int requestVersion; // bumped on every new topic addition
    //上一次更新的时间（包含更新失败）
    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;

public class MetadataCache {
    //kafka集群中的node、topic信息
    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 final Map<String, Uuid> topicIds;

    private Cluster clusterInstance;

public final class Cluster {
    
    private final boolean isBootstrapConfigured;
    //集群的节点列表node
    private final List<Node> nodes;
    //未认证的topic
    private final Set<String> unauthorizedTopics;
    //不合理的topic
    private final Set<String> invalidTopics;
    //内部topic
    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和节点的对应关系map
    private final Map<Integer, Node> nodesById;
    //集群信息
    private final ClusterResource clusterResource;
    private final Map<String, Uuid> topicIds;
    private final Map<Uuid, String> topicNames;

public class ClusterResource {

    private final String clusterId;

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;

doSend()方法中调用waitOnMetadata()方法等待更新集群的元数据。（但如果超过了配置的时间metadataExpireMs没有更新， 就会强制更新）。

public List<ClientResponse> poll(long timeout, long now) {
        ensureActive();

        if (!abortedSends.isEmpty()) {
            // If there are aborted sends because of unsupported version exceptions or disconnects,
            // handle them immediately without waiting for Selector#poll.
            List<ClientResponse> responses = new ArrayList<>();
            handleAbortedSends(responses);
            completeResponses(responses);
            return responses;
        }

        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // process completed actions
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleInitiateApiVersionRequests(updatedNow);
        handleTimedOutConnections(responses, updatedNow);
        handleTimedOutRequests(responses, updatedNow);
        completeResponses(responses);

        return responses;
    }

        public long maybeUpdate(long now) {
            // should we update our metadata?
            //获取下一次更新时间，若needUpdate为true返回0立马更新，否则返回剩余过期时间
            long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);

            //计算要等待的时间，有正在处理的请求，则返回默认的请求间隔否则返回0
            long waitForMetadataFetch = hasFetchInProgress() ? defaultRequestTimeoutMs : 0;

            long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
            //大于0说明还需等待一段时间才能更新
            if (metadataTimeout > 0) {
                return metadataTimeout;
            }
            
            // Beware that the behavior of this method and the computation of timeouts for poll() are
            // highly dependent on the behavior of leastLoadedNode.
            Node node = leastLoadedNode(now);
            if (node == null) {
                log.debug("Give up sending metadata request since no node is available");
                return reconnectBackoffMs;
            }

            return maybeUpdate(now, node);
        }