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

Code before:
        transitionTo(MemberState.UNSUBSCRIBED);
        leaveGroupInProgress.get().complete(null);
        leaveGroupInProgress = Optional.empty();
    }

    /**
     * @return True if there are no assignments waiting to be resolved from metadata or reconciled.
     */
    private boolean allPendingAssignmentsReconciled() {
        return assignmentUnresolved.isEmpty() && assignmentReadyToReconcile.isEmpty();
    }

    @Override
    public boolean shouldSkipHeartbeat() {
        MemberState state = state();
        return state == MemberState.UNSUBSCRIBED || state == MemberState.FATAL;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void transitionToStaled() {
        transitionTo(MemberState.PREPARE_LEAVING);
        memberEpoch = ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
        transitionTo(MemberState.LEAVING);
        leaveGroupInProgress = Optional.of(CompletableFuture.completedFuture(null));
        transitionTo(MemberState.STALED);
    }

    /**
     * Reconcile the assignment that has been received from the server and for which topic names
     * are resolved, kept in the {@link #assignmentReadyToReconcile}. This will commit if needed,
     * trigger the callbacks and update the subscription state. Note that only one reconciliation
     * can be in progress at a time. If there is already another one in progress when this is
     * triggered, it will be no-op, and the assignment will be reconciled on the next
     * reconciliation loop.
     */
    boolean reconcile() {
        if (reconciliationInProgress) {
            log.debug("Ignoring reconciliation attempt. Another reconciliation is already in progress. Assignment " +
                    assignmentReadyToReconcile + " will be handled in the next reconciliation loop.");
            return false;
        }

        // Make copy of the assignment to reconcile as it could change as new assignments or metadata updates are received
        SortedSet<TopicIdPartition> assignedTopicIdPartitions = new TreeSet<>(TOPIC_ID_PARTITION_COMPARATOR);
        assignedTopicIdPartitions.addAll(assignmentReadyToReconcile);

        SortedSet<TopicPartition> ownedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
        ownedPartitions.addAll(subscriptions.assignedPartitions());

Review comment (suggestion):
Similar, do we need this? We use this LEAVING state to signal the HB manager that it should send a HB now, but in this staled member case, that last HB is sent based on the check that the poll timer expired (not based on any state that should HB now)

Code after:

    /**
     * Reset member epoch to the value required for the leave the group heartbeat request, and
     * transition to the {@link MemberState#LEAVING} state so that a heartbeat
     * request is sent out with it.
     */
    private void transitionToSendingLeaveGroup() {
        if (state == MemberState.FATAL) {
            log.warn("Member {} with epoch {} won't send leave group request because it is in " +
                    "FATAL state", memberId, memberEpoch);
            return;
        }
        memberEpoch = groupInstanceId.isPresent() ?
                ConsumerGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH :
                ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
        commitRequestManager.updateMemberInformation(this.memberId, this.memberEpoch);
        currentAssignment = new HashMap<>();
        transitionTo(MemberState.LEAVING);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean shouldHeartbeatNow() {
        MemberState state = state();
        return state == MemberState.ACKNOWLEDGING || state == MemberState.LEAVING || state == MemberState.JOINING;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onHeartbeatRequestSent() {
        MemberState state = state();
        if (isStaled()) {
            log.debug("Member {} is staled and is therefore leaving the group.  It will rejoin upon the next poll.", memberEpoch);
            transitionToJoining();
            return;
        }

        if (state == MemberState.ACKNOWLEDGING) {
            if (allPendingAssignmentsReconciled()) {
                transitionTo(MemberState.STABLE);
            } else {
                log.debug("Member {} with epoch {} transitioned to {} after a heartbeat was sent " +
                        "to ack a previous reconciliation. New assignments are ready to " +
                        "be reconciled.", memberId, memberEpoch, MemberState.RECONCILING);
                transitionTo(MemberState.RECONCILING);
            }
        } else if (state == MemberState.LEAVING) {