protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
        final Cluster cluster = metadata.fetch();
        Map<Node, FetchSessionHandler.Builder> fetchable = new HashMap<>();

        sessionHandlers.forEach((fetchTargetNodeId, sessionHandler) -> {
            // set the session handler to notify close. This will set the next metadata request to send close message.
            sessionHandler.notifyClose();

            // FetchTargetNode may not be available as it may have disconnected the connection. In such cases, we will
            // skip sending the close request.
            final Node fetchTarget = cluster.nodeById(fetchTargetNodeId);

            if (fetchTarget == null || isUnavailable(fetchTarget)) {
                log.debug("Skip sending close session request to broker {} since it is not reachable", fetchTarget);
                return;
            }

            fetchable.put(fetchTarget, sessionHandler.newBuilder());
        });

        return fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build()));
    }
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