public void onUnloaded() {
        groups.values().forEach(group -> {
            switch (group.type()) {
                case CONSUMER:
                    ConsumerGroup consumerGroup = (ConsumerGroup) group;
                    log.info("[GroupId={}] Unloaded group metadata for group epoch {}.",
                        consumerGroup.groupId(), consumerGroup.groupEpoch());
                    break;
                case CLASSIC:
                    ClassicGroup classicGroup = (ClassicGroup) group;
                    log.info("[GroupId={}] Unloading group metadata for generation {}.",
                        classicGroup.groupId(), classicGroup.generationId());

                    classicGroup.transitionTo(DEAD);
                    switch (classicGroup.previousState()) {
                        case EMPTY:
                        case DEAD:
                            break;
                        case PREPARING_REBALANCE:
                            classicGroup.allMembers().forEach(member -> {
                                classicGroup.completeJoinFuture(member, new JoinGroupResponseData()
                                    .setMemberId(member.memberId())
                                    .setErrorCode(NOT_COORDINATOR.code()));
                            });

                            break;
                        case COMPLETING_REBALANCE:
                        case STABLE:
                            classicGroup.allMembers().forEach(member -> {
                                classicGroup.completeSyncFuture(member, new SyncGroupResponseData()
                                    .setErrorCode(NOT_COORDINATOR.code()));
                            });
                    }
                    break;
                case SHARE:
                    ShareGroup shareGroup = (ShareGroup) group;
                    log.info("[GroupId={}] Unloaded group metadata for group epoch {}.",
                        shareGroup.groupId(), shareGroup.groupEpoch());
                    break;

                default:
                    log.warn("onUnloaded group with an unknown group type {}.", group.type());
                    break;
            }
        });
    }
@Test
    public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .build();
        ClassicGroup group = context.createClassicGroup("group-id");

        // Set up a group in with a leader, follower, and a pending member.
        // Have the pending member join the group and both the pending member
        // and the follower sync. We should have 2 members awaiting sync.
        GroupMetadataManagerTestContext.PendingMemberGroupResult pendingGroupResult = context.setupGroupWithPendingMember(group);
        String pendingMemberId = pendingGroupResult.pendingMemberResponse.memberId();

        // Compete join group for the pending member
        JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
            .withGroupId("group-id")
            .withMemberId(pendingMemberId)
            .withDefaultProtocolTypeAndProtocols()
            .build();

        GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(request);

        assertTrue(joinResult.records.isEmpty());
        assertTrue(joinResult.joinFuture.isDone());
        assertEquals(Errors.NONE.code(), joinResult.joinFuture.get().errorCode());
        assertEquals(3, group.allMembers().size());
        assertEquals(0, group.numPendingJoinMembers());

        // Follower and pending send SyncGroup request.
        // Follower and pending member should be awaiting sync while the leader is pending sync.
        GroupMetadataManagerTestContext.SyncResult followerSyncResult = context.sendClassicGroupSync(
            new GroupMetadataManagerTestContext.SyncGroupRequestBuilder()
                .withGroupId("group-id")
                .withMemberId(pendingGroupResult.followerId)
                .withGenerationId(joinResult.joinFuture.get().generationId())
                .build());

        GroupMetadataManagerTestContext.SyncResult pendingMemberSyncResult = context.sendClassicGroupSync(
            new GroupMetadataManagerTestContext.SyncGroupRequestBuilder()
                .withGroupId("group-id")
                .withMemberId(pendingMemberId)
                .withGenerationId(joinResult.joinFuture.get().generationId())
                .build());

        assertFalse(followerSyncResult.syncFuture.isDone());
        assertFalse(pendingMemberSyncResult.syncFuture.isDone());
        assertTrue(group.isInState(COMPLETING_REBALANCE));

        context.onUnloaded();

        assertTrue(group.isInState(DEAD));
        assertTrue(followerSyncResult.syncFuture.isDone());
        assertTrue(pendingMemberSyncResult.syncFuture.isDone());
        assertEquals(new SyncGroupResponseData()
            .setAssignment(EMPTY_ASSIGNMENT)
            .setErrorCode(NOT_COORDINATOR.code()), followerSyncResult.syncFuture.get());
        assertEquals(new SyncGroupResponseData()
            .setAssignment(EMPTY_ASSIGNMENT)
            .setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
    }