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

Code before:

    private void transitionToUnsubscribed() {
        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);


Review comment (suggestion):
Looks to me this is not really needed here? We use this state when we are taking some action before sending the last HB (mainly callbacks), but in this case there is no async operation to do, just send the last HB right? which is explicitly done in the HB mgr when checking that the poll timer expired

Code after:
        return callbackResult;
    }

    /**
     * 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);