public void updateNodeLatencyStats(Integer nodeId, long nowMs, boolean canDrain) {
        // Don't bother with updating stats if the feature is turned off.
        if (partitionAvailabilityTimeoutMs <= 0)
            return;

        // When the sender gets a node (returned by the ready() function) that has data to send
        // but the node is not ready (and so we cannot drain the data), we only update the
        // ready time, then the difference would reflect for how long a node wasn't ready
        // to send the data.  Then we can temporarily remove partitions that are handled by the
        // node from the list of available partitions so that the partitioner wouldn't pick
        // this partition.
        // NOTE: there is no synchronization for metric updates, so drainTimeMs is updated
        // first to avoid accidentally marking a partition unavailable if the reader gets
        // values between updates.
        NodeLatencyStats nodeLatencyStats = nodeStats.computeIfAbsent(nodeId, id -> new NodeLatencyStats(nowMs));
        if (canDrain)
            nodeLatencyStats.drainTimeMs = nowMs;
        nodeLatencyStats.readyTimeMs = nowMs;
    }
Drain all the data for the given nodes and collate them into a list of batches that will fit
within the specified size on a per-node basis. This method attempts to avoid choosing the same
topic-node over and over.

@param metadataSnapshot  The current cluster metadata
@param nodes             The list of node to drain
@param maxSize           The maximum number of bytes to drain
@param now               The current unix time in milliseconds
@return A list of {@link ProducerBatch} for each node specified with total size less than the
requested maxSize.
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java