@Override
    public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
        if (!coordinatorRequestManager.coordinator().isPresent() ||
            membershipManager.shouldSkipHeartbeat()) {
            membershipManager.onHeartbeatRequestSkipped();
            return NetworkClientDelegate.PollResult.EMPTY;
        }
        pollTimer.update(currentTimeMs);
        if (pollTimer.isExpired() && !membershipManager.isLeavingGroup()) {
            logger.warn("Consumer poll timeout has expired. This means the time between " +
                "subsequent calls to poll() was longer than the configured max.poll.interval.ms, " +
                "which typically implies that the poll loop is spending too much time processing " +
                "messages. You can address this either by increasing max.poll.interval.ms or by " +
                "reducing the maximum size of batches returned in poll() with max.poll.records.");

            membershipManager.transitionToSendingLeaveGroup(true);
            NetworkClientDelegate.UnsentRequest leaveHeartbeat = makeHeartbeatRequest(currentTimeMs, true);

            // We can ignore the leave response because we can join before or after receiving the response.
            heartbeatRequestState.reset();
            heartbeatState.reset();
            return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(leaveHeartbeat));
        }

        // Case 1: The member is leaving
        boolean heartbeatNow = membershipManager.state() == MemberState.LEAVING ||
                // Case 2: The member state indicates it should send a heartbeat without waiting for the interval, and there is no heartbeat request currently in-flight
                (membershipManager.shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight());

        if (!heartbeatRequestState.canSendRequest(currentTimeMs) && !heartbeatNow) {
            return new NetworkClientDelegate.PollResult(heartbeatRequestState.timeToNextHeartbeatMs(currentTimeMs));
        }

        NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(currentTimeMs, false);
        return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
    }
This will build a heartbeat request if one must be sent, determined based on the member
state. A heartbeat is sent in the following situations:
<ol>
    <li>Member is part of the consumer group or wants to join it.</li>
    <li>The heartbeat interval has expired, or the member is in a state that indicates
    that it should heartbeat without waiting for the interval.</li>
</ol>
This will also determine the maximum wait time until the next poll based on the member's
state.
<ol>
    <li>If the member is without a coordinator or is in a failed state, the timer is set
    to Long.MAX_VALUE, as there's no need to send a heartbeat.</li>
    <li>If the member cannot send a heartbeat due to either exponential backoff, it will
    return the remaining time left on the backoff timer.</li>
    <li>If the member's heartbeat timer has not expired, It will return the remaining time
    left on the heartbeat timer.</li>
    <li>If the member can send a heartbeat, the timer is set to the current heartbeat interval.</li>
</ol>

@return {@link PollResult} that includes a heartbeat request if one must be sent, and the
time to wait until the next poll.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java