Node selectReadReplica(final TopicPartition partition, final Node leaderReplica, final long currentTimeMs) {
        Optional<Integer> nodeId = subscriptions.preferredReadReplica(partition, currentTimeMs);

        if (nodeId.isPresent()) {
            Optional<Node> node = nodeId.flatMap(id -> metadata.fetch().nodeIfOnline(partition, id));
            if (node.isPresent()) {
                return node.get();
            } else {
                log.trace("Not fetching from {} for partition {} since it is marked offline or is missing from our metadata," +
                        " using the leader instead.", nodeId, partition);
                // Note that this condition may happen due to stale metadata, so we clear preferred replica and
                // refresh metadata.
                requestMetadataUpdate(metadata, subscriptions, partition);
                return leaderReplica;
            }
        } else {
            return leaderReplica;
        }
    }
Determine from which replica to read: the <i>preferred</i> or the <i>leader</i>. The preferred replica is used
iff:

<ul>
    <li>A preferred replica was previously set</li>
    <li>We're still within the lease time for the preferred replica</li>
    <li>The replica is still online/available</li>
</ul>

If any of the above are not met, the leader node is returned.

@param partition {@link TopicPartition} for which we want to fetch data
@param leaderReplica {@link Node} for the leader of the given partition
@param currentTimeMs Current time in milliseconds; used to determine if we're within the optional lease window
@return Replica {@link Node node} from which to request the data
@see SubscriptionState#preferredReadReplica
@see SubscriptionState#updatePreferredReadReplica
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java