// 无Node表示
private static final Node NO_NODE = new Node(-1, "", -1);

// broker节点id
private final int id;
// broker节点id的字符串形式
private final String idString;
// broker节点的地址
private final String host;
// broker节点的端口
private final int port;
// broker节点的机架
private final String rack;

// hash值，用于hashCode()方法缓存
private int hash = 0;
// 分区编号
private final int partition;
// 主题名称
private final String topic;
...
public int hashCode() {
    if (hash != 0)
        return hash;
    final int prime = 31;
    int result = 1;
    result = prime * result + partition;
    result = prime * result + ((topic == null) ? 0 : topic.hashCode());
    this.hash = result;
    return result;
}

// 主题
private final String topic;
// 分区编号
private final int partition;
// 分区leader副本信息
private final Node leader;
// 全部副本信息
private final Node[] replicas;
// ISR副本信息
private final Node[] inSyncReplicas;

// Kafka集群中的broker节点集合
private final List<Node> nodes;
// 未授权的主题集合
private final Set<String> unauthorizedTopics;
private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
// topic对应的partition信息字典，键为topic名称，值为partition信息集合；存放的partition不一定有Leader副本
private final Map<String, List<PartitionInfo>> partitionsByTopic;
// topic中可用的partition信息字典，键为topic名称，值为可用的partition信息集合；存放的partition必须是有Leader副本的Partition
private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
// broker对应的partition信息字典，键为broker的id，值为partition信息集合
private final Map<Integer, List<PartitionInfo>> partitionsByNode;
// broker对应的Node信息字典，键为broker的id，值为表示该节点的Node实例
private final Map<Integer, Node> nodesById;

// 添加partition信息
public Cluster withPartitions(Map<TopicPartition, PartitionInfo> partitions) {
    Map<TopicPartition, PartitionInfo> combinedPartitions = new HashMap<>(this.partitionsByTopicPartition);
    combinedPartitions.putAll(partitions);
    return new Cluster(this.nodes, combinedPartitions.values(), new HashSet<>(this.unauthorizedTopics));
}

// 根据Node ID获取Node
public Node nodeById(int id) {
    return this.nodesById.get(id);
}

// 根据TopicPartition获取partition的leader分区所在的节点
public Node leaderFor(TopicPartition topicPartition) {
    PartitionInfo info = partitionsByTopicPartition.get(topicPartition);
    if (info == null)
        return null;
    else
        return info.leader();
}

// 根据TopicPartition获取PartitionInfo
public PartitionInfo partition(TopicPartition topicPartition) {
    return partitionsByTopicPartition.get(topicPartition);
}

// 获取主题的分区列表
public List<PartitionInfo> partitionsForTopic(String topic) {
    return this.partitionsByTopic.get(topic);
}

// 获取主题的可用分区列表
public List<PartitionInfo> availablePartitionsForTopic(String topic) {
    return this.availablePartitionsByTopic.get(topic);
}

// 根据Node ID获取该Node上的分区列表
public List<PartitionInfo> partitionsForNode(int nodeId) {
    return this.partitionsByNode.get(nodeId);
}

// 根据主题获取该主题的分区数
public Integer partitionCountForTopic(String topic) {
    List<PartitionInfo> partitionInfos = this.partitionsByTopic.get(topic);
    return partitionInfos == null ? null : partitionInfos.size();
}

// 两次刷新元数据退避时间，避免频繁刷新导致性能消耗
private final long refreshBackoffMs;
// 每隔多久更新一次，默认是300秒
private final long metadataExpireMs;
// 集群元数据版本号，元数据更新成功一次，版本号就自增1
private int version;
// 上一次更新元数据的时间戳
private long lastRefreshMs;
/**
 * 上一次成功更新元数据的时间戳，如果每次更新都成功，
 * lastSuccessfulRefreshMs应该与lastRefreshMs相同，否则lastRefreshMs > lastSuccessfulRefreshMs
 */
private long lastSuccessfulRefreshMs;
// 记录kafka集群的元数据
private Cluster cluster;
// 表示是否强制更新Cluster
private boolean needUpdate;
// 记录当前已知的所有的主题
private final Set<String> topics;
// 监听器集合，用于监听Metadata更新
private final List<Listener> listeners;
// 是否需要更新全部主题的元数据
private boolean needMetadataForAllTopics;

...
// 创建Sender
this.sender = new Sender(client,
        this.metadata,
        this.accumulator,
        config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) == 1, //（max.in.flight.requests.per.connection）
        config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG), //（max.request.size）
        (short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)), //（acks）
        config.getInt(ProducerConfig.RETRIES_CONFIG), //（retries）
        this.metrics,
        new SystemTime(),
        clientId,
        this.requestTimeoutMs);
// 指定Sender线程的名称
String ioThreadName = "kafka-producer-network-thread" + (clientId.length() > 0 ? " | " + clientId : "");
// 启动Sender对应的线程
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

// Sender
public void run() {
    ...
    while (running) {
        try {
            run(time.milliseconds());
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread: ", e);
        }
    }
    ...
}

void run(long now) {
    // 获取Cluster集群元数据信息
    Cluster cluster = metadata.fetch();
    // get the list of partitions with data ready to send
    // 获取当前集群中符合发送消息条件的数据集
    RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

    // if there are any partitions whose leaders are not known yet, force metadata update
    //  如果ReadyCheckResult中标识有unknownLeadersExist，则调用Metadata的requestUpdate方法，标记需要更新Kafka的集群信息
    if (result.unknownLeadersExist)
        this.metadata.requestUpdate();
    ...
    for (ClientRequest request : requests)
        // 使用NetworkClient将ClientRequest对象写入KafkaChannel的send字段
        client.send(request, now);

    // if some partitions are already ready to be sent, the select time would be 0;
    // otherwise if some partition already has some data accumulated but not ready yet,
    // the select time will be the time difference between now and its linger expiry time;
    // otherwise the select time will be the time difference between now and the metadata expiry time;
    // 使用NetworkClient将KafkaChannel的send字段中保存的ClientRequest发送出去
    this.client.poll(pollTimeout, now);
}

// Metadata
public synchronized int requestUpdate() {
    this.needUpdate = true; // 设置为需要更新
    return this.version; // 返回当前集群元数据的版本号
}

public List<ClientResponse> poll(long timeout, long now) {
    // 每次poll()的时候都会判断是否需要更新Metadata
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
        // 执行IO操作
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }

    // process completed actions
    long updatedNow = this.time.milliseconds();
    // 响应队列
    List<ClientResponse> responses = new ArrayList<>();
    // 处理completeSends队列
    handleCompletedSends(responses, updatedNow);
    // 处理completeReceives队列
    handleCompletedReceives(responses, updatedNow);
    // 处理disconnected列表
    handleDisconnections(responses, updatedNow);
    // 处理connected列表
    handleConnections();
    // 处理InFlightRequest中超时请求
    handleTimedOutRequests(responses, updatedNow);

    // invoke callbacks
    // 遍历ClientResponse
    for (ClientResponse response : responses) {
        // 如果对应的ClientRequest有回调就执行回调
        if (response.request().hasCallback()) {
            try {
                response.request().callback().onComplete(response);
            } catch (Exception e) {
                log.error("Uncaught error in request completion:", e);
            }
        }
    }

    return responses;
}

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    TopicPartition tp = null;
    try {
        // first make sure the metadata for the topic is available
        // 获取Kafka集群信息，会唤醒Sender线程更新Kafka集群的元数据
        long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
        ...
    } catch (ApiException e) {
        ...
    }
}

private long waitOnMetadata(String topic, long maxWaitMs) throws InterruptedException {
    // add topic to metadata topic list if it is not there already.
    // 如果Metadata对象的topics集合中没有该topic，就将其加入
    if (!this.metadata.containsTopic(topic))
        this.metadata.add(topic);

    // fetch()返回的是Cluster对象，判断Cluster对象中是否有该topic的分区信息，如果有就可以直接返回
    if (metadata.fetch().partitionsForTopic(topic) != null)
        return 0;

    long begin = time.milliseconds();
    long remainingWaitMs = maxWaitMs;
    // 如果没有拉取到相应主题的元数据，将会重复拉取
    while (metadata.fetch().partitionsForTopic(topic) == null) {
        log.trace("Requesting metadata update for topic {}.", topic);
        // 设置Metadata的needUpdate字段，并得到当前的元数据版本号
        int version = metadata.requestUpdate();
        // 唤醒Sender线程，唤醒的NetworkClient中的Selector，最终唤醒的是NIO的Selector
        sender.wakeup();
        // 等待更新，会阻塞
        metadata.awaitUpdate(version, remainingWaitMs);
        long elapsed = time.milliseconds() - begin;
        if (elapsed >= maxWaitMs)
            // 等待超过最大超时时间，直接抛出异常
            throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
        if (metadata.fetch().unauthorizedTopics().contains(topic))
            // 检查主题的授权信息，如果未授权，直接抛出异常
            throw new TopicAuthorizationException(topic);
        // 更新超时时间
        remainingWaitMs = maxWaitMs - elapsed;
    }
    return time.milliseconds() - begin;
}

public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
    if (maxWaitMs < 0) {
        throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milli seconds");
    }
    long begin = System.currentTimeMillis();
    long remainingWaitMs = maxWaitMs;
    // 比较版本号
    while (this.version <= lastVersion) {
        if (remainingWaitMs != 0)
            // 进入wait阻塞，等待更新成功被notify或超时，进入下一次while循环
            wait(remainingWaitMs);
        long elapsed = System.currentTimeMillis() - begin;
        if (elapsed >= maxWaitMs)
            throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
        remainingWaitMs = maxWaitMs - elapsed;
    }
}

// 选择负载最小的Node节点
public Node leastLoadedNode(long now) {
    // 获取存放在Cluster中的Node节点列表
    List<Node> nodes = this.metadataUpdater.fetchNodes();
    int inflight = Integer.MAX_VALUE;
    Node found = null;

    // 获得随机偏移
    int offset = this.randOffset.nextInt(nodes.size());
    for (int i = 0; i < nodes.size(); i++) {
        // 根据随机偏移随机选一个Node
        int idx = (offset + i) % nodes.size();
        Node node = nodes.get(idx);
        // 获得该Node节点中请求队列的中的请求数量
        int currInflight = this.inFlightRequests.inFlightRequestCount(node.idString());
        if (currInflight == 0 && this.connectionStates.isConnected(node.idString())) {
            // 如果该Node节点的请求队列中没有请求，且该Node节点是连接的，就选该节点
            // if we find an established connection with no in-flight requests we can stop right away
            return node;
        } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) {
            // otherwise if this is the best we have found so far, record that
            // 如果不符合，则重新选择
            inflight = currInflight;
            found = node;
        }
    }

    return found;
}

private void doSend(ClientRequest request, long now) {
    request.setSendTimeMs(now);
    // 将请求添加到inFlightRequest队列中等待响应
    this.inFlightRequests.add(request);
    // 将请求交由KafkaChannel处理
    selector.send(request.request());
}

public void add(ClientRequest request) {
    /**
     * 以ClientRequest对象中封装的RequestSend的destination为键从requests中获取对应的Deque
     * 这个destination的值其实是当时发送请求的目的Node节点的id，可以回顾下面的方法：
     * {@link NetworkClient.DefaultMetadataUpdater#maybeUpdate}
     * requests的结构是Map<String, Deque<ClientRequest>>
     */
    Deque<ClientRequest> reqs = this.requests.get(request.request().destination());
    // 如果没有获取到就创建一个
    if (reqs == null) {
        reqs = new ArrayDeque<>();
        this.requests.put(request.request().destination(), reqs);
    }
    // 将ClientRequest对象添加到对应的Deque中
    reqs.addFirst(request);
}

// 内部封装了java.nio.channels.Selector
public void send(Send send) {
    KafkaChannel channel = channelOrFail(send.destination());
    try {
        /**
         * 将send对象缓存到KafkaChannel的send字段中，同时添加OP_WRITE事件的关注
         * send对象实际类型是RequestSend对象，其中封装了具体的请求数据，包括请求头和请求体
         * 这里只是将RequestSend对象用KafkaChannel的send字段记录下来
         * 具体的发送会在Selector.poll()方法中进行
         * KafkaChannel每次只会发送一个RequestSend对象
         */
        channel.setSend(send);
    } catch (CancelledKeyException e) {
        this.failedSends.add(send.destination());
        close(channel);
    }
}

// 注意：该方法会将RequestSend对象与KafkaChannel关联起来，使用KafkaChannel的send属性进行记录
public void setSend(Send send) {
    if (this.send != null)
        throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
    // 记录send
    this.send = send;
    // 添加OP_WRITE事件关注
    this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
}

public boolean canSendMore (String node) {
    Deque＜ClientRequest＞ queue = requests.get(node);
    return queue == null  // 队列未创建
            || queue.isEmpty() // 队列为空
            || (queue.peekFirst().request().completed() // 队首请求已完成
                ＆＆ queue.size() ＜ this.maxInFlightRequestsPerConnection); // 队列长度小于maxInFlightRequestsPerConnection
}

public List<ClientResponse> poll(long timeout, long now) {
    // 每次poll()的时候都会判断是否需要更新Metadata
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
        // 执行IO操作
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }
    ...
}

if (channel.ready() && key.isWritable()) {
    // Channel可写，处理OP_WRITE事件
    Send send = channel.write();
    /**
     * channel.write()将KafkaChannel中保存的send字段发送出去，
     * 如果发送成功就会返回send，然后将其添加到completedSends集合，等待后续处理
     * 如果发送未完成会返回null，然后在下一次处理中继续发送
     */
    if (send != null) {
        this.completedSends.add(send);
        this.sensors.recordBytesSent(channel.id(), send.size());
    }
}

// KafkaChannel类
public Send write() throws IOException {
    Send result = null;
    // send()方法发送时，如果发送完成会返回true，否则返回false
    if (send != null && send(send)) {
        // 完成后使用result记录send，用于返回，并将send置为null，用于下一次发送
        result = send;
        send = null;
    }
    return result;
}

private boolean send(Send send) throws IOException {
    // 发送send，此处的send对象是RequestSend类型
    send.writeTo(transportLayer);
    // 检查是否完成
    if (send.completed())
        // 如果完成就移除对OP_WRITE事件的关注
        transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

    return send.completed();
}

// 需要注意的继承关系：RequestSend -继承-> NetworkSend -继承-> ByteBufferSend -实现-> Send
// 因此上面的RequestSend调用的writeTo()方法其实继承自ByteBufferSend类，
// 传入的channel是TransportLayer类型对象，这个后面会讲解
public long writeTo(GatheringByteChannel channel) throws IOException {
    // 使用Channel写出数据
    long written = channel.write(buffers);
    if (written < 0)
        throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");
    remaining -= written;
    // This is temporary workaround. As Send , Receive interfaces are being used by BlockingChannel.
    // Once BlockingChannel is removed we can make Send, Receive to work with transportLayer rather than
    // GatheringByteChannel or ScatteringByteChannel.
    if (channel instanceof TransportLayer)
        pending = ((TransportLayer) channel).hasPendingWrites();

    return written;
}

// Selector类
/* if channel is ready read from any connections that have readable data */
if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
    // Channel可读，处理OP_READ事件
    NetworkReceive networkReceive;
    // 循环接收，直到1个response完全接收到，才会从while循环退出
    while ((networkReceive = channel.read()) != null)
        /**
         * 读取信息并将读到的NetworkReceive添加到stagedReceives集合中保存
         * 若读取到一个完整的NetworkReceive，则将其添加到stagedReceives集合中保存
         * 若读取不到一个完整的NetworkReceive，则返回null，下次处理OP_READ事件时，
         * 继续读取，知道读取到一个完整的NetworkReceive
         */
        addToStagedReceives(channel, networkReceive);
}

private void addToStagedReceives(KafkaChannel channel, NetworkReceive receive) {
    // 如果stagedReceives中还没有与指定KafkaChannel对应的ArrayDeque，就创建一个新的
    if (!stagedReceives.containsKey(channel))
        stagedReceives.put(channel, new ArrayDeque<NetworkReceive>());

    // 获取stagedReceives中指定KafkaChannel对应的Deque
    Deque<NetworkReceive> deque = stagedReceives.get(channel);
    // 将接收的NetworkReceive对象添加到该Deque中
    deque.add(receive);
}

// Selector类
public void poll(long timeout) throws IOException {
    ...
    // 调用select()方法
    int readyKeys = select(timeout);
    ...
    if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
        // 处理IO事件
        pollSelectionKeys(this.nioSelector.selectedKeys(), false);
        pollSelectionKeys(immediatelyConnectedKeys, true);
    }
    // 将statedReceives复制到completedReceives集合中
    addToCompletedReceives();
    ...
}

private void addToCompletedReceives() {
    if (!this.stagedReceives.isEmpty()) {
        // 如果stagedReceives集合不为空，则遍历该集合
        Iterator<Map.Entry<KafkaChannel, Deque<NetworkReceive>>> iter = this.stagedReceives.entrySet().iterator();
        while (iter.hasNext()) {
            // 取出对应的键值对
            Map.Entry<KafkaChannel, Deque<NetworkReceive>> entry = iter.next();
            // 获取KafkaChannel
            KafkaChannel channel = entry.getKey();
            if (!channel.isMute()) {
                // 判断KafkaChannel是否是mute状态，如果不是才表示此时KafkaChannel已经完成了读写操作
                Deque<NetworkReceive> deque = entry.getValue();
                // 获取队首networkReceive并添加到completedReceives
                NetworkReceive networkReceive = deque.poll();
                this.completedReceives.add(networkReceive);
                this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
                // 如果队列空了，移除键值对
                if (deque.isEmpty())
                    iter.remove();
            }
        }
    }
}

private void handleCompletedReceives(List<ClientResponse> responses, long now) {
    // 遍历completeReceives中的NetworkReceive
    for (NetworkReceive receive : this.selector.completedReceives()) {
        // 获取返回响应的NodeId，这里的source对应于前年发送数据时，RequestSend的destination字段
        String source = receive.source();
        // 从inFlightRequests中取出对应的ClientRequest
        ClientRequest req = inFlightRequests.completeNext(source);
        // 解析响应
        Struct body = parseResponse(receive.payload(), req.request().header());
        /**
         * 调用metadataUpdater.maybeHandleCompletedReceive()方法检查是否是Metadata更新的响应
         * 如果是，就处理MetadataResponse，会更新Metadata中的记录的集群元数据，并唤醒所有等待Metadata更新完成的线程
         * 如果不是会返回false
         */
        if (!metadataUpdater.maybeHandleCompletedReceive(req, now, body))
            // 如果不是MetadataResponse，则创建ClientResponse并添加到responses集合
            responses.add(new ClientResponse(req, now, false, body));
    }
}

private void handleResponse(RequestHeader header, Struct body, long now) {
    // 修改metadataFetchInProgress
    this.metadataFetchInProgress = false;
    // 解析MetadataResponse
    MetadataResponse response = new MetadataResponse(body);
    // 根据response返回的数据，创建新的Cluster对象
    Cluster cluster = response.cluster();
    // check if any topics metadata failed to get updated
    Map<String, Errors> errors = response.errors();
    if (!errors.isEmpty())
        log.warn("Error while fetching metadata with correlation id {} : {}", header.correlationId(), errors);

    // don't update the cluster if there are no valid nodes...the topic we want may still be in the process of being
    // created which means we will get errors and no nodes until it exists
    if (cluster.nodes().size() > 0) {
        // 首先通知Metadata上的监听器，然后更新cluster字段，最后唤醒等待Metadata更新完成的线程
        this.metadata.update(cluster, now);
    } else {
        log.trace("Ignoring empty metadata response with correlation id {}.", header.correlationId());
        // 更新Metadata失败，将lastRefreshMs字段更新
        this.metadata.failedUpdate(now);
    }
}

public synchronized void update(Cluster cluster, long now) {
    this.needUpdate = false;
    this.lastRefreshMs = now;
    this.lastSuccessfulRefreshMs = now;
    this.version += 1;

    // 通知所有的监听器，数据要更新了
    for (Listener listener: listeners)
        listener.onMetadataUpdate(cluster);

    // Do this after notifying listeners as subscribed topics' list can be changed by listeners
    // 更新Cluster字段
    this.cluster = this.needMetadataForAllTopics ? getClusterForCurrentTopics(cluster) : cluster;

    // 唤醒等待Metadata更新完成的线程
    notifyAll();
    log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster);
}

public long maybeUpdate(long now) {
    /**
     * 调用Metadata.timeToNextUpdate()方法，其中会检测needUpdate的值、退避时间、是否长时间未更新
     * 最终得到一个下次更新集群元数据的时间戳
     */
    // should we update our metadata?
    long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
    // 获取下次尝试重新连接服务端的时间戳
    long timeToNextReconnectAttempt = Math.max(this.lastNoNodeAvailableMs + metadata.refreshBackoff() - now, 0);
    // 检测metadataFetchInProgress字段，判断是否已经发送了请求
    long waitForMetadataFetch = this.metadataFetchInProgress ? Integer.MAX_VALUE : 0;
    // if there is no node available to connect, back off refreshing metadata
    // 计算当前距离下次可以发送MetadataRequest请求的时间差
    long metadataTimeout = Math.max(Math.max(timeToNextMetadataUpdate, timeToNextReconnectAttempt),
            waitForMetadataFetch);

    if (metadataTimeout == 0) {
        // 允许发送MetadataRequest请求
        // Beware that the behavior of this method and the computation of timeouts for poll() are
        // highly dependent on the behavior of leastLoadedNode.
        // 找到负载最小的Node，若没有可用就返回null
        Node node = leastLoadedNode(now);
        // 将跟新Metadata的请求发送给这个Node；这里只会创建并缓存MetadataRequest，等待下次poll()方法发送请求
        maybeUpdate(now, node);
    }

    return metadataTimeout;
}

public synchronized long timeToNextUpdate(long nowMs) {
    /**
     * 元数据是否过期，判断条件：
     * 1. needUpdate被置为true
     * 2. 上次更新时间距离当前时间已经超过了指定的元数据过期时间阈值metadataExpireMs（metadata.max.age.ms），默认是300秒
     */
    long timeToExpire = needUpdate ? 0 : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0);
    /**
     * 允许更新的时间点，计算方式：
     * 上次更新时间 + 退避时间 - 当前时间的间隔
     * 即要求上次更新时间与当前时间的间隔不能大于退避时间，如果大于则需要等待
     */
    long timeToAllowUpdate = this.lastRefreshMs + this.refreshBackoffMs - nowMs;
    return Math.max(timeToExpire, timeToAllowUpdate);
}

void run(long now) {
    // 获取Cluster集群元数据信息
    Cluster cluster = metadata.fetch();
    // get the list of partitions with data ready to send
    // 获取当前集群中符合发送消息条件的数据集
    RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

    // if there are any partitions whose leaders are not known yet, force metadata update
    //  如果ReadyCheckResult中标识有unknownLeadersExist，则调用Metadata的requestUpdate方法，标记需要更新Kafka的集群信息
    if (result.unknownLeadersExist)
        this.metadata.requestUpdate();
    ...
}

private void initiateConnect(Node node, long now) {
    String nodeConnectionId = node.idString();
    try {
        log.debug("Initiating connection to node {} at {}:{}.", node.id(), node.host(), node.port());
        this.connectionStates.connecting(nodeConnectionId, now);
        // 发起连接
        selector.connect(nodeConnectionId,
                            new InetSocketAddress(node.host(), node.port()),
                            this.socketSendBuffer,
                            this.socketReceiveBuffer);
    } catch (IOException e) {
        /* attempt failed, we'll try again after the backoff */
        connectionStates.disconnected(nodeConnectionId, now);
        /* maybe the problem is our metadata, update it */
        // 如果发起连接失败，可能是由于集群元数据发生了变化，这里会触发一次Metadata的更新，最终会调用Metadata的requestUpdate()方法
        metadataUpdater.requestUpdate();
        log.debug("Error connecting to node {} at {}:{}:", node.id(), node.host(), node.port(), e);
    }
}

public List<ClientResponse> poll(long timeout, long now) {
    // 每次poll()的时候都会判断是否需要更新Metadata
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
        // 执行IO操作
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }
    ...
    // 处理disconnected列表
    handleDisconnections(responses, updatedNow);
    ...
}

private void handleDisconnections(List<ClientResponse> responses, long now) {
    // 更新连接状态，并清理掉InFlightRequests中断开连接的Node对应的ClientRequest
    for (String node : this.selector.disconnected()) {
        log.debug("Node {} disconnected.", node);
        processDisconnection(responses, node, now);
    }
    // we got a disconnect so we should probably refresh our metadata and see if that broker is dead
    if (this.selector.disconnected().size() > 0)
        // 有连接断开了，标识需要更新集群元数据，最终会调用Metadata的requestUpdate()方法
        metadataUpdater.requestUpdate();
}

// NetworkClient类
public List<ClientResponse> poll(long timeout, long now) {
    // 每次poll()的时候都会判断是否需要更新Metadata
    long metadataTimeout = metadataUpdater.maybeUpdate(now);
    try {
        // 执行IO操作
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }
    ...
    // 处理InFlightRequest中超时请求
    handleTimedOutRequests(responses, updatedNow);
    ...
}

private void handleTimedOutRequests(List<ClientResponse> responses, long now) {
    // 从inFlightRequests中获取请求超时的Node的集合
    List<String> nodeIds = this.inFlightRequests.getNodesWithTimedOutRequests(now, this.requestTimeoutMs);
    // 遍历Node关闭连接并更新状态
    for (String nodeId : nodeIds) {
        // close connection to the node
        this.selector.close(nodeId);
        log.debug("Disconnecting from node {} due to request timeout.", nodeId);
        processDisconnection(responses, nodeId, now);
    }

    // we disconnected, so we should probably refresh our metadata
    // 有超时请求，因此需要更新Metadata元数据，最终会调用Metadata的requestUpdate()方法
    if (nodeIds.size() > 0)
        metadataUpdater.requestUpdate();
}