public CompletableFuture<Void> leaveGroup() {
        if (isNotInGroup()) {
            if (state == MemberState.FENCED) {
                clearSubscription();
                transitionTo(MemberState.UNSUBSCRIBED);
            }
            return CompletableFuture.completedFuture(null);
        }

        if (state == MemberState.PREPARE_LEAVING || state == MemberState.LEAVING) {
            // Member already leaving. No-op and return existing leave group future that will
            // complete when the ongoing leave operation completes.
            return leaveGroupInProgress.get();
        }

        transitionTo(MemberState.PREPARE_LEAVING);
        CompletableFuture<Void> leaveResult = new CompletableFuture<>();
        leaveGroupInProgress = Optional.of(leaveResult);

        clearSubscription();

        // Transition to ensure that a heartbeat request is sent out to effectively leave the
        // group (even in the case where the member had no assignment to release)
        transitionToSendingLeaveGroup(false);

        // Return future to indicate that the leave group is done when the transition to send
        // the heartbeat has been made.
        return leaveResult;
    }
Release assignment and transition to {@link MemberState#PREPARE_LEAVING} so that a heartbeat
request is sent indicating the broker that the member wants to leave the group. This is
expected to be invoked when the user calls the unsubscribe API.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManager.java