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 testClassicGroupOnUnloadedEmptyAndPreparingRebalance() throws Exception {
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .build();

        ClassicGroup emptyGroup = context.createClassicGroup("empty-group");
        assertTrue(emptyGroup.isInState(EMPTY));

        ClassicGroup preparingGroup = context.createClassicGroup("preparing-group");
        JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
            .withGroupId("preparing-group")
            .withMemberId(UNKNOWN_MEMBER_ID)
            .withDefaultProtocolTypeAndProtocols()
            .build();

        // preparing-group should have 2 members.
        GroupMetadataManagerTestContext.JoinResult joinResult1 = context.sendClassicGroupJoin(request);
        GroupMetadataManagerTestContext.JoinResult joinResult2 = context.sendClassicGroupJoin(request);

        assertFalse(joinResult1.joinFuture.isDone());
        assertFalse(joinResult2.joinFuture.isDone());
        assertTrue(preparingGroup.isInState(PREPARING_REBALANCE));
        assertEquals(2, preparingGroup.numMembers());

        context.onUnloaded();

        assertTrue(emptyGroup.isInState(DEAD));
        assertTrue(preparingGroup.isInState(DEAD));
        assertTrue(joinResult1.joinFuture.isDone());
        assertTrue(joinResult2.joinFuture.isDone());
        assertEquals(new JoinGroupResponseData()
            .setMemberId(joinResult1.joinFuture.get().memberId())
            .setMembers(Collections.emptyList())
            .setErrorCode(NOT_COORDINATOR.code()), joinResult1.joinFuture.get());

        assertEquals(new JoinGroupResponseData()
            .setMemberId(joinResult2.joinFuture.get().memberId())
            .setMembers(Collections.emptyList())
            .setErrorCode(NOT_COORDINATOR.code()), joinResult2.joinFuture.get());
    }