Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java (Java)

Code before:
     *     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.
     */
    @Override
    public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
        if (!coordinatorRequestManager.coordinator().isPresent() ||
            membershipManager.shouldSkipHeartbeat() ||
            pollTimer.isExpired()) {
            membershipManager.onHeartbeatRequestSkipped();
            return NetworkClientDelegate.PollResult.EMPTY;
        }

        pollTimer.update(currentTimeMs);
        if (pollTimer.isExpired()) {
            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.");
            return PollResult.EMPTY;
        }

        boolean heartbeatNow = membershipManager.shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight();

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

        heartbeatRequestState.onSendAttempt(currentTimeMs);
        membershipManager.onHeartbeatRequestSent();
        NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest();
        return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
    }

    /**
     * Returns the delay for which the application thread can safely wait before it should be responsive
     * to results from the request managers. For example, the subscription state can change when heartbeats
     * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
     * responsive to changes.
     *
     * <p>In the event that heartbeats are currently being skipped, this still returns the next heartbeat
     * delay rather than {@code Long.MAX_VALUE} so that the application thread remains responsive.
     */
    @Override
    public long maximumTimeToWait(long currentTimeMs) {

Review comment (suggestion):
I'm thinking what about instead of returning the empty here, we transition the member to "leaving" (`transitionToSendingLeaveGroup`), and that's it. On the following lines that should make the shouldHbNow true, we would send the last HB and transition to unsubscribed (effectively leaving without triggering the callbacks, which is what the old logic does)

Code after:
     *     <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.
     */
    @Override
    public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
        if (!coordinatorRequestManager.coordinator().isPresent() ||
            membershipManager.shouldSkipHeartbeat() ||
            pollTimer.isExpired()) {
            membershipManager.onHeartbeatRequestSkipped();
            return NetworkClientDelegate.PollResult.EMPTY;
        }
        pollTimer.update(currentTimeMs);
        if (pollTimer.isExpired()) {
            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.");
            // This should trigger a heartbeat with leave group epoch
            membershipManager.transitionToStaled();
            NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(currentTimeMs, true);
            // We can ignore the leave response because we can join before or after receiving the response.
            heartbeatRequestState.reset();
            return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
        }

        boolean heartbeatNow = membershipManager.shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight();
        if (!heartbeatRequestState.canSendRequest(currentTimeMs) && !heartbeatNow) {
            return new NetworkClientDelegate.PollResult(heartbeatRequestState.nextHeartbeatMs(currentTimeMs));
        }

        NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(currentTimeMs, false);
        return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
    }

    /**
     * Returns the delay for which the application thread can safely wait before it should be responsive
     * to results from the request managers. For example, the subscription state can change when heartbeats
     * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
     * responsive to changes.
     *
     * <p>In the event that heartbeats are currently being skipped, this still returns the next heartbeat
     * delay rather than {@code Long.MAX_VALUE} so that the application thread remains responsive.
     */